Skip to content

Update data load templates #42

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Zhimin-arya merged 1 commit into from
Jul 21, 2026
Merged

Update data load templates #42

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 7 additions & 5 deletions flows/Load_OMOP_CDM_v53.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -106,17 +106,19 @@
}
],
"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": "" }
],
"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" }
]
}
12 changes: 7 additions & 5 deletions flows/Load_OMOP_CDM_v54.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"

@Zhimin-arya Zhimin-arya Jul 16, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

attribute_definition is not included in OMOP CDM 5.4.

},
"type": "python_node",
"width": 350,
Expand Down Expand Up @@ -106,17 +106,19 @@
}
],
"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": "" }
],
"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" }
]
}