lineage tests

dataprep/tests/lineage/dependency_example/a_table.sql

@@ -0,0 +1,2 @@
+SELECT subject_id, gender
+FROM `physionet-data.mimiciii_derived.no_dob`;

dataprep/tests/lineage/dependency_example/aa_table.sql

@@ -0,0 +1,6 @@
+CREATE VIEW aa_table AS
+SELECT a.subject_id, b.gender
+FROM `physionet-data.mimiciii_derived.a_table` a, `physionet-data.mimiciii_derived.no_dob` b;
+CREATE TABLE a_table AS
+SELECT subject_id, gender
+FROM `physionet-data.mimiciii_derived.no_dob`;

dataprep/tests/lineage/dependency_example/basic_patient_info.sql

@@ -0,0 +1,11 @@
+-- ------------------------------------------------------------------
+-- Title: Retrieves basic patient information from the patients table
+-- Notes: this query does not specify a schema. To run it on your local
+-- MIMIC schema, run the following command:
+-- SET SEARCH_PATH TO mimiciii;
+-- Where "mimiciii" is the name of your schema, and may be different.
+-- ------------------------------------------------------------------
+
+
+SELECT subject_id, gender, dob
+FROM `physionet-data.mimiciii_clinical.patients`;

dataprep/tests/lineage/dependency_example/from_aa_table.sql

@@ -0,0 +1 @@
+SELECT * FROM aa_table;

dataprep/tests/lineage/dependency_example/no_dob.sql

@@ -0,0 +1,2 @@
+SELECT subject_id, gender, dob
+FROM `physionet-data.mimiciii_derived.basic_patient_info`;

dataprep/tests/lineage/test_read_sql.py

@@ -0,0 +1,32 @@
+# type: ignore
+from os import environ
+import os
+import pytest
+
+from ...lineage import lineagex
+
+
+@pytest.mark.skipif(
+ environ.get("DB_URL", "") == "",
+ reason="Skip tests that requires database setup and sql query specified",
+)
+def test_read_sql() -> None:
+ db_url = environ["DB_URL"]
+ sql = os.path.join(os.getcwd(), 'dependency_example')
+ lx = lineagex(
+ sql,
+ "mimiciii_derived",
+ db_url,
+ "mimiciii_clinical, public"
+ )
+ print("dependency test with database connection", lx)
+ lx = lineagex(
+ sql=sql,
+ target_schema="mimiciii_derived",
+ search_path_schema="mimiciii_clinical, public"
+ )
+ print("dependency test without database connection", lx)
+
+
+if __name__ == "__main__":
+ test_read_sql()