diff --git a/flows/Load_OMOP_CDM_v53.json b/flows/Load_OMOP_CDM_v53.json index 80c3edd..8f09e0a 100644 --- a/flows/Load_OMOP_CDM_v53.json +++ b/flows/Load_OMOP_CDM_v53.json @@ -36,7 +36,7 @@ "data": { "name": "TruncateTables", "description": "Truncate tables by setting boolean flag", - "python_code": "def exec(myinput) -> dict[str, str]:\n \"\"\"\n Execute database table truncation and schema mapping for CDM and vocabulary tables.\n \n This function creates a mapping of OMOP CDM (Common Data Model) and vocabulary tables\n to their respective database schemas, and optionally truncates these tables before\n returning the mapping.\n \n Args:\n myinput: Input parameter (currently unused, reserved for future functionality).\n \n Returns:\n dict[str, str]: A dictionary mapping table names to their schema names.\n Keys are table names (e.g., 'person', 'concept'), values are schema names.\n \n Raises:\n Exception: If any tables fail truncation and skip_if_error is False, or if\n truncation fails for any tables when skip_if_error is True (after attempting\n all tables).\n \n Note:\n - Requires global variables: cdm_schema, vocab_schema, database_code, truncate_tables, skip_truncate_if_error\n - truncate_tables flag controls whether tables are truncated (default: True)\n - skip_if_error flag controls error handling during truncation (default: False)\n \"\"\"\n\n # From shared variables\n cdm_schema_var: str = cdm_schema \n vocab_schema_var: str = vocab_schema\n database_code_var: str = database_code\n truncate_tables_var: bool = eval(truncate_tables) # If set to true, tables will be truncated\n skip_if_error: bool = eval(skip_truncate_if_error) # If set to true, tables that failed to be truncated will be ignored\n\n # List of OMOP CDM clinical data tables\n cdm_table_list: list[str] = [\n \"person\",\n \"observation_period\",\n \"visit_occurrence\",\n \"visit_detail\",\n \"condition_occurrence\",\n \"drug_exposure\",\n \"procedure_occurrence\",\n \"device_exposure\",\n \"measurement\",\n \"observation\",\n \"death\",\n \"note\",\n \"note_nlp\",\n \"specimen\",\n \"fact_relationship\",\n \"location\",\n \"care_site\",\n \"provider\",\n \"payer_plan_period\",\n \"cost\",\n \"drug_era\",\n \"dose_era\",\n \"condition_era\",\n \"metadata\",\n \"cdm_source\",\n \"attribute_definition\"\n ]\n\n # List of OMOP vocabulary tables\n vocab_table_list: list[str] = [\n \"concept\",\n \"vocabulary\",\n \"domain\",\n \"concept_class\",\n \"concept_relationship\",\n \"relationship\",\n \"concept_synonym\",\n \"concept_ancestor\",\n \"source_to_concept_map\",\n \"drug_strength\"\n ]\n\n # Mapping of table names to their respective schemas\n table_schema_map: dict[str, str] = {\n **{table: cdm_schema_var for table in cdm_table_list},\n **{table: vocab_schema_var for table in vocab_table_list},\n }\n\n if truncate_tables:\n truncate_success: dict[str, bool] = {}\n\n dbdao = DBDao(database_code=database_code_var)\n\n for table, schema in table_schema_map.items():\n try: \n dbdao.truncate_table(schema, table)\n except Exception as e:\n truncate_success[table] = False\n if skip_if_error:\n continue\n else:\n raise\n else:\n truncate_success[table] = True\n\n false_count = sum(1 for v in truncate_success.values() if v is False)\n if false_count > 0:\n false_keys = [k for k, v in truncate_success.items() if v is False]\n raise Exception(f\"Some tables failed truncation: {false_keys}\")\n\n return table_schema_map\n return table_schema_map" + "python_code": "def exec(myinput) -> dict[str, str]:\n \"\"\"\n Execute database table truncation and schema mapping for CDM and vocabulary tables.\n \n This function creates a mapping of OMOP CDM (Common Data Model) and vocabulary tables\n to their respective database schemas, and optionally truncates these tables before\n returning the mapping.\n \n Args:\n myinput: Input parameter (currently unused, reserved for future functionality).\n \n Returns:\n dict[str, str]: A dictionary mapping table names to their schema names.\n Keys are table names (e.g., 'person', 'concept'), values are schema names.\n \n Raises:\n Exception: If any tables fail truncation and skip_if_error is False, or if\n truncation fails for any tables when skip_if_error is True (after attempting\n all tables).\n \n Note:\n - Requires global variables: cdm_schema, vocab_schema, database_code, truncate_tables, skip_truncate_if_error\n - truncate_tables flag controls whether tables are truncated (default: True)\n - skip_if_error flag controls error handling during truncation (default: False)\n \"\"\"\n\n # From shared variables\n cdm_schema_var: str = cdm_schema \n vocab_schema_var: str = vocab_schema\n database_code_var: str = database_code\n truncate_tables_var: bool = eval(truncate_tables) # If set to true, tables will be truncated\n skip_if_error: bool = eval(skip_truncate_if_error) # If set to true, tables that failed to be truncated will be ignored\n\n # List of OMOP CDM clinical data tables\n cdm_table_list: list[str] = [\n \"person\",\n \"observation_period\",\n \"visit_occurrence\",\n \"visit_detail\",\n \"condition_occurrence\",\n \"drug_exposure\",\n \"procedure_occurrence\",\n \"device_exposure\",\n \"measurement\",\n \"observation\",\n \"death\",\n \"note\",\n \"note_nlp\",\n \"specimen\",\n \"fact_relationship\",\n \"location\",\n \"care_site\",\n \"provider\",\n \"payer_plan_period\",\n \"cost\",\n \"drug_era\",\n \"dose_era\",\n \"condition_era\",\n \"metadata\",\n \"cdm_source\",\n \"attribute_definition\"\n ]\n\n # List of OMOP vocabulary tables\n vocab_table_list: list[str] = [\n \"concept\",\n \"vocabulary\",\n \"domain\",\n \"concept_class\",\n \"concept_relationship\",\n \"relationship\",\n \"concept_synonym\",\n \"concept_ancestor\",\n \"source_to_concept_map\",\n \"drug_strength\"\n ]\n\n # Mapping of table names to their respective schemas\n table_schema_map: dict[str, str] = {\n **{table: cdm_schema_var for table in cdm_table_list},\n **{table: vocab_schema_var for table in vocab_table_list},\n }\n\n if truncate_tables_var:\n truncate_success: dict[str, bool] = {}\n\n dbdao = DBDao(database_code=database_code_var)\n\n for table, schema in table_schema_map.items():\n try: \n dbdao.truncate_table(schema, table)\n except Exception as e:\n truncate_success[table] = False\n if skip_if_error:\n continue\n else:\n raise\n else:\n truncate_success[table] = True\n\n false_count = sum(1 for v in truncate_success.values() if v is False)\n if false_count > 0:\n false_keys = [k for k, v in truncate_success.items() if v is False]\n raise Exception(f\"Some tables failed truncation: {false_keys}\")\n\n return table_schema_map\n return table_schema_map" }, "type": "python_node", "width": 350, @@ -106,11 +106,8 @@ } ], "variables": [ - { "key": "database_code", "value": "demo_database" }, - { "key": "cdm_schema", "value": "cdmdefault" }, - { "key": "vocab_schema", "value": "cdmdefault" }, { "key": "delimiter", "value": "," }, - { "key": "encoding", "value": "utf+8" }, + { "key": "encoding", "value": "utf-8" }, { "key": "truncate_tables", "value": "True" }, { "key": "skip_truncate_if_error", "value": "True" }, { "key": "exclude_tables", "value": "" } @@ -118,5 +115,10 @@ "importLibs": [ "from _shared_flow_utils.dao.DBDao import DBDao", "from _shared_flow_utils.api.SupabaseStorageAPI import SupabaseStorageAPI" + ], + "databases": [{ "code": "demo_database", "name": "database_code" }], + "schemas": [ + { "name": "cdm_schema", "schema": "cdmdefault" }, + { "name": "vocab_schema", "schema": "cdmdefault" } ] } diff --git a/flows/Load_OMOP_CDM_v54.json b/flows/Load_OMOP_CDM_v54.json index 3f60123..ce868e6 100644 --- a/flows/Load_OMOP_CDM_v54.json +++ b/flows/Load_OMOP_CDM_v54.json @@ -36,7 +36,7 @@ "data": { "name": "TruncateTables", "description": "Truncate tables by setting boolean flag", - "python_code": "def exec(myinput) -> dict[str, str]:\n \"\"\"\n Execute database table truncation and schema mapping for CDM and vocabulary tables.\n \n This function creates a mapping of OMOP CDM (Common Data Model) and vocabulary tables\n to their respective database schemas, and optionally truncates these tables before\n returning the mapping.\n \n Args:\n myinput: Input parameter (currently unused, reserved for future functionality).\n \n Returns:\n dict[str, str]: A dictionary mapping table names to their schema names.\n Keys are table names (e.g., 'person', 'concept'), values are schema names.\n \n Raises:\n Exception: If any tables fail truncation and skip_if_error is False, or if\n truncation fails for any tables when skip_if_error is True (after attempting\n all tables).\n \n Note:\n - Requires global variables: cdm_schema, vocab_schema, database_code, truncate_tables, skip_truncate_if_error\n - truncate_tables flag controls whether tables are truncated (default: True)\n - skip_if_error flag controls error handling during truncation (default: False)\n \"\"\"\n\n # From shared variables\n cdm_schema_var: str = cdm_schema \n vocab_schema_var: str = vocab_schema\n database_code_var: str = database_code\n truncate_tables_var: bool = eval(truncate_tables) # If set to true, tables will be truncated\n skip_if_error: bool = eval(skip_truncate_if_error) # If set to true, tables that failed to be truncated will be ignored\n\n # List of OMOP CDM clinical data tables\n cdm_table_list: list[str] = [\n \"person\",\n \"observation_period\",\n \"visit_occurrence\",\n \"visit_detail\",\n \"condition_occurrence\",\n \"drug_exposure\",\n \"procedure_occurrence\",\n \"device_exposure\",\n \"measurement\",\n \"observation\",\n \"death\",\n \"note\",\n \"note_nlp\",\n \"specimen\",\n \"fact_relationship\",\n \"location\",\n \"care_site\",\n \"provider\",\n \"payer_plan_period\",\n \"cost\",\n \"drug_era\",\n \"dose_era\",\n \"condition_era\",\n \"episode\",\n \"episode_event\",\n \"metadata\",\n \"cdm_source\",\n \"attribute_definition\"\n ]\n\n # List of OMOP vocabulary tables\n vocab_table_list: list[str] = [\n \"concept\",\n \"vocabulary\",\n \"domain\",\n \"concept_class\",\n \"concept_relationship\",\n \"relationship\",\n \"concept_synonym\",\n \"concept_ancestor\",\n \"source_to_concept_map\",\n \"drug_strength\"\n ]\n\n # Mapping of table names to their respective schemas\n table_schema_map: dict[str, str] = {\n **{table: cdm_schema_var for table in cdm_table_list},\n **{table: vocab_schema_var for table in vocab_table_list},\n }\n\n if truncate_tables:\n truncate_success: dict[str, bool] = {}\n\n dbdao = DBDao(database_code=database_code_var)\n\n for table, schema in table_schema_map.items():\n try: \n dbdao.truncate_table(schema, table)\n except Exception as e:\n truncate_success[table] = False\n if skip_if_error:\n continue\n else:\n raise\n else:\n truncate_success[table] = True\n\n false_count = sum(1 for v in truncate_success.values() if v is False)\n if false_count > 0:\n false_keys = [k for k, v in truncate_success.items() if v is False]\n raise Exception(f\"Some tables failed truncation: {false_keys}\")\n\n return table_schema_map\n return table_schema_map" + "python_code": "def exec(myinput) -> dict[str, str]:\n \"\"\"\n Execute database table truncation and schema mapping for CDM and vocabulary tables.\n \n This function creates a mapping of OMOP CDM (Common Data Model) and vocabulary tables\n to their respective database schemas, and optionally truncates these tables before\n returning the mapping.\n \n Args:\n myinput: Input parameter (currently unused, reserved for future functionality).\n \n Returns:\n dict[str, str]: A dictionary mapping table names to their schema names.\n Keys are table names (e.g., 'person', 'concept'), values are schema names.\n \n Raises:\n Exception: If any tables fail truncation and skip_if_error is False, or if\n truncation fails for any tables when skip_if_error is True (after attempting\n all tables).\n \n Note:\n - Requires global variables: cdm_schema, vocab_schema, database_code, truncate_tables, skip_truncate_if_error\n - truncate_tables flag controls whether tables are truncated (default: True)\n - skip_if_error flag controls error handling during truncation (default: False)\n \"\"\"\n\n # From shared variables\n cdm_schema_var: str = cdm_schema \n vocab_schema_var: str = vocab_schema\n database_code_var: str = database_code\n truncate_tables_var: bool = eval(truncate_tables) # If set to true, tables will be truncated\n skip_if_error: bool = eval(skip_truncate_if_error) # If set to true, tables that failed to be truncated will be ignored\n\n # List of OMOP CDM clinical data tables\n cdm_table_list: list[str] = [\n \"person\",\n \"observation_period\",\n \"visit_occurrence\",\n \"visit_detail\",\n \"condition_occurrence\",\n \"drug_exposure\",\n \"procedure_occurrence\",\n \"device_exposure\",\n \"measurement\",\n \"observation\",\n \"death\",\n \"note\",\n \"note_nlp\",\n \"specimen\",\n \"fact_relationship\",\n \"location\",\n \"care_site\",\n \"provider\",\n \"payer_plan_period\",\n \"cost\",\n \"drug_era\",\n \"dose_era\",\n \"condition_era\",\n \"episode\",\n \"episode_event\",\n \"metadata\",\n \"cdm_source\"\n ]\n\n # List of OMOP vocabulary tables\n vocab_table_list: list[str] = [\n \"concept\",\n \"vocabulary\",\n \"domain\",\n \"concept_class\",\n \"concept_relationship\",\n \"relationship\",\n \"concept_synonym\",\n \"concept_ancestor\",\n \"source_to_concept_map\",\n \"drug_strength\"\n ]\n\n # Mapping of table names to their respective schemas\n table_schema_map: dict[str, str] = {\n **{table: cdm_schema_var for table in cdm_table_list},\n **{table: vocab_schema_var for table in vocab_table_list},\n }\n\n if truncate_tables_var:\n truncate_success: dict[str, bool] = {}\n\n dbdao = DBDao(database_code=database_code_var)\n\n for table, schema in table_schema_map.items():\n try: \n dbdao.truncate_table(schema, table)\n except Exception as e:\n truncate_success[table] = False\n if skip_if_error:\n continue\n else:\n raise\n else:\n truncate_success[table] = True\n\n false_count = sum(1 for v in truncate_success.values() if v is False)\n if false_count > 0:\n false_keys = [k for k, v in truncate_success.items() if v is False]\n raise Exception(f\"Some tables failed truncation: {false_keys}\")\n\n return table_schema_map\n return table_schema_map" }, "type": "python_node", "width": 350, @@ -106,11 +106,8 @@ } ], "variables": [ - { "key": "database_code", "value": "demo_database" }, - { "key": "cdm_schema", "value": "cdmdefault" }, - { "key": "vocab_schema", "value": "cdmdefault" }, { "key": "delimiter", "value": "," }, - { "key": "encoding", "value": "utf+8" }, + { "key": "encoding", "value": "utf-8" }, { "key": "truncate_tables", "value": "True" }, { "key": "skip_truncate_if_error", "value": "True" }, { "key": "exclude_tables", "value": "" } @@ -118,5 +115,10 @@ "importLibs": [ "from _shared_flow_utils.dao.DBDao import DBDao", "from _shared_flow_utils.api.SupabaseStorageAPI import SupabaseStorageAPI" + ], + "databases": [{ "code": "demo_database", "name": "database_code" }], + "schemas": [ + { "name": "cdm_schema", "schema": "cdmdefault" }, + { "name": "vocab_schema", "schema": "cdmdefault" } ] }