Update templates to use database variables

flows/Load_OMOP_CDM_v53.json

@@ -85,7 +85,7 @@
       "data": {
         "name": "IngestTables",
         "description": "Ingest into OMOP CDM tables",
-        "python_code": "from typing import Any, Dict, Optional\nimport pandas as pd\n\n\ndef exec(myinput: Dict[str, Any]) -> None:\n  \"\"\"\n  Ingest CDM DataFrames into their corresponding database tables.\n  \n  This function takes transformed OMOP CDM DataFrames and inserts them into\n  the appropriate database schemas using SQLAlchemy's to_sql method.\n  \n  Args:\n    myinput: Dictionary containing workflow results with:\n      - 'LoadAndTransformFiles': Contains 'result' with Dict[str, pd.DataFrame]\n        mapping table names to their DataFrames\n      - 'TruncateTables': Contains 'result' with Dict[str, str] mapping\n        table names to their schema names\n        \n  Returns:\n    None\n    \n  Note:\n    - Requires global variables: cdm_schema, vocab_schema, database_code\n    - Requires DBDao class to be available in scope\n    - Uses 'append' mode for table insertion\n    - Disposes database connection after completion\n  \"\"\"\n  # Extract the dictionary of DataFrames from the LoadAndTransformFiles step\n  df_list: Dict[str, pd.DataFrame] = myinput.get(\"LoadAndTransformFiles\").result\n  \n  # Extract the table-to-schema mapping from the TruncateTables step\n  table_schema_map: Dict[str, str] = myinput.get(\"TruncateTables\").result\n\n  # Get configuration from global variables\n  database_code_var: str = database_code\n  cdm_schema_var: str = cdm_schema \n  vocab_schema_var: str = vocab_schema\n  \n\n  # Initialize database connection (not using cache database)\n  dbdao = DBDao(use_cache_db=False, database_code=database_code_var)\n  dbconn = dbdao.engine\n\n  # Insert each DataFrame into its corresponding database table\n  for table, df in df_list.items():\n    # Insert DataFrame into the database table\n    # - Uses the schema from table_schema_map\n    # - Appends to existing data (table should already be truncated)\n    # - Does not include DataFrame index as a column\n    result: Optional[int] = df.to_sql(\n      table, \n      dbconn, \n      schema=table_schema_map[table],\n      if_exists='append',\n      index=False\n    )\n  \n  # Clean up database connection\n  dbconn.dispose()"
+        "python_code": "from typing import Any, Dict, Optional\nimport pandas as pd\n\n\ndef exec(myinput: Dict[str, Any]) -> None:\n  \"\"\"\n  Ingest CDM DataFrames into their corresponding database tables.\n  \n  This function takes transformed OMOP CDM DataFrames and inserts them into\n  the appropriate database schemas using SQLAlchemy's to_sql method.\n  \n  Args:\n    myinput: Dictionary containing workflow results with:\n      - 'LoadAndTransformFiles': Contains 'result' with Dict[str, pd.DataFrame]\n        mapping table names to their DataFrames\n      - 'TruncateTables': Contains 'result' with Dict[str, str] mapping\n        table names to their schema names\n        \n  Returns:\n    None\n    \n  Note:\n    - Requires global variables: cdm_schema, vocab_schema, database_code\n    - Requires DBDao class to be available in scope\n    - Uses 'append' mode for table insertion\n    - Disposes database connection after completion\n  \"\"\"\n  # Extract the dictionary of DataFrames from the LoadAndTransformFiles step\n  df_list: Dict[str, pd.DataFrame] = myinput.get(\"LoadAndTransformFiles\").result\n  \n  # Extract the table-to-schema mapping from the TruncateTables step\n  table_schema_map: Dict[str, str] = myinput.get(\"TruncateTables\").result\n\n  # Get configuration from global variables\n  database_code_var: str = database_code\n  cdm_schema_var: str = cdm_schema \n  vocab_schema_var: str = vocab_schema\n  \n\n  # Initialize database connection (not using cache database)\n  dbdao = DBDao(database_code=database_code_var)\n  dbconn = dbdao.engine\n\n  # Insert each DataFrame into its corresponding database table\n  for table, df in df_list.items():\n    # Insert DataFrame into the database table\n    # - Uses the schema from table_schema_map\n    # - Appends to existing data (table should already be truncated)\n    # - Does not include DataFrame index as a column\n    result: Optional[int] = df.to_sql(\n      table, \n      dbconn, \n      schema=table_schema_map[table],\n      if_exists='append',\n      index=False\n    )\n  \n  # Clean up database connection\n  dbconn.dispose()"
       },
       "type": "python_node",
       "width": 350,

flows/Load_OMOP_CDM_v54.json

@@ -85,7 +85,7 @@
       "data": {
         "name": "IngestTables",
         "description": "Ingest into OMOP CDM tables",
-        "python_code": "from typing import Any, Dict, Optional\nimport pandas as pd\n\n\ndef exec(myinput: Dict[str, Any]) -> None:\n  \"\"\"\n  Ingest CDM DataFrames into their corresponding database tables.\n  \n  This function takes transformed OMOP CDM DataFrames and inserts them into\n  the appropriate database schemas using SQLAlchemy's to_sql method.\n  \n  Args:\n    myinput: Dictionary containing workflow results with:\n      - 'LoadAndTransformFiles': Contains 'result' with Dict[str, pd.DataFrame]\n        mapping table names to their DataFrames\n      - 'TruncateTables': Contains 'result' with Dict[str, str] mapping\n        table names to their schema names\n        \n  Returns:\n    None\n    \n  Note:\n    - Requires global variables: cdm_schema, vocab_schema, database_code\n    - Requires DBDao class to be available in scope\n    - Uses 'append' mode for table insertion\n    - Disposes database connection after completion\n  \"\"\"\n  # Extract the dictionary of DataFrames from the LoadAndTransformFiles step\n  df_list: Dict[str, pd.DataFrame] = myinput.get(\"LoadAndTransformFiles\").result\n  \n  # Extract the table-to-schema mapping from the TruncateTables step\n  table_schema_map: Dict[str, str] = myinput.get(\"TruncateTables\").result\n\n  # Get configuration from global variables\n  database_code_var: str = database_code\n  cdm_schema_var: str = cdm_schema \n  vocab_schema_var: str = vocab_schema\n  \n\n  # Initialize database connection (not using cache database)\n  dbdao = DBDao(use_cache_db=False, database_code=database_code_var)\n  dbconn = dbdao.engine\n\n  # Insert each DataFrame into its corresponding database table\n  for table, df in df_list.items():\n    # Insert DataFrame into the database table\n    # - Uses the schema from table_schema_map\n    # - Appends to existing data (table should already be truncated)\n    # - Does not include DataFrame index as a column\n    result: Optional[int] = df.to_sql(\n      table, \n      dbconn, \n      schema=table_schema_map[table],\n      if_exists='append',\n      index=False\n    )\n  \n  # Clean up database connection\n  dbconn.dispose()"
+        "python_code": "from typing import Any, Dict, Optional\nimport pandas as pd\n\n\ndef exec(myinput: Dict[str, Any]) -> None:\n  \"\"\"\n  Ingest CDM DataFrames into their corresponding database tables.\n  \n  This function takes transformed OMOP CDM DataFrames and inserts them into\n  the appropriate database schemas using SQLAlchemy's to_sql method.\n  \n  Args:\n    myinput: Dictionary containing workflow results with:\n      - 'LoadAndTransformFiles': Contains 'result' with Dict[str, pd.DataFrame]\n        mapping table names to their DataFrames\n      - 'TruncateTables': Contains 'result' with Dict[str, str] mapping\n        table names to their schema names\n        \n  Returns:\n    None\n    \n  Note:\n    - Requires global variables: cdm_schema, vocab_schema, database_code\n    - Requires DBDao class to be available in scope\n    - Uses 'append' mode for table insertion\n    - Disposes database connection after completion\n  \"\"\"\n  # Extract the dictionary of DataFrames from the LoadAndTransformFiles step\n  df_list: Dict[str, pd.DataFrame] = myinput.get(\"LoadAndTransformFiles\").result\n  \n  # Extract the table-to-schema mapping from the TruncateTables step\n  table_schema_map: Dict[str, str] = myinput.get(\"TruncateTables\").result\n\n  # Get configuration from global variables\n  database_code_var: str = database_code\n  cdm_schema_var: str = cdm_schema \n  vocab_schema_var: str = vocab_schema\n  \n\n  # Initialize database connection (not using cache database)\n  dbdao = DBDao(database_code=database_code_var)\n  dbconn = dbdao.engine\n\n  # Insert each DataFrame into its corresponding database table\n  for table, df in df_list.items():\n    # Insert DataFrame into the database table\n    # - Uses the schema from table_schema_map\n    # - Appends to existing data (table should already be truncated)\n    # - Does not include DataFrame index as a column\n    result: Optional[int] = df.to_sql(\n      table, \n      dbconn, \n      schema=table_schema_map[table],\n      if_exists='append',\n      index=False\n    )\n  \n  # Clean up database connection\n  dbconn.dispose()"
       },
       "type": "python_node",
       "width": 350,