From bb2606c7ff0e433c8f45967d210311bf3d137628 Mon Sep 17 00:00:00 2001 From: imilinovic Date: Mon, 5 Jan 2026 16:57:49 +0100 Subject: [PATCH 1/7] fix migrate --- python/migrate.py | 618 ++++++++++++++++++++++++++++++---------------- 1 file changed, 411 insertions(+), 207 deletions(-) diff --git a/python/migrate.py b/python/migrate.py index d44990ca6..b514ddeab 100644 --- a/python/migrate.py +++ b/python/migrate.py @@ -1,6 +1,7 @@ import base64 import csv import datetime +import hashlib import io import json import os @@ -41,6 +42,30 @@ class Constants: USERNAME = "username" +def _get_query_hash(query: str, config: mgp.Map, params: mgp.Nullable[mgp.Any] = None) -> str: + """ + Create a hash from query, config, and params to use as a cache key. + + :param query: The query string (or table name, endpoint, file path, etc.) + :param config: Configuration map + :param params: Optional query parameters + """ + config_dict = dict(config) if config else {} + config_str = json.dumps(config_dict, sort_keys=True, default=str) + + params_str = "" + if params is not None: + if isinstance(params, dict): + params_str = json.dumps(params, sort_keys=True, default=str) + elif isinstance(params, (list, tuple)): + params_str = json.dumps(list(params), sort_keys=False, default=str) + else: + params_str = str(params) + + hash_input = f"{query}|{config_str}|{params_str}" + return hashlib.sha256(hash_input.encode('utf-8')).hexdigest() + + # MYSQL mysql_dict = {} @@ -62,23 +87,24 @@ def init_migrate_mysql( if _query_is_table(table_or_sql): table_or_sql = f"SELECT * FROM {table_or_sql};" - thread_id = threading.get_native_id() - if thread_id not in mysql_dict: - mysql_dict[thread_id] = {} + query_hash = _get_query_hash(table_or_sql, config, params) + + # check if query is already running + if query_hash in mysql_dict: + raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") - if Constants.CURSOR not in mysql_dict[thread_id]: - mysql_dict[thread_id][Constants.CURSOR] = None + if query_hash not in mysql_dict: + mysql_dict[query_hash] = {} - if mysql_dict[thread_id][Constants.CURSOR] is None: - connection = mysql_connector.connect(**config) - cursor = connection.cursor() - cursor.execute(table_or_sql, params=params) + connection = mysql_connector.connect(**config) + cursor = connection.cursor() + cursor.execute(table_or_sql, params=params) - mysql_dict[thread_id][Constants.CONNECTION] = connection - mysql_dict[thread_id][Constants.CURSOR] = cursor - mysql_dict[thread_id][Constants.COLUMN_NAMES] = [ - column[Constants.I_COLUMN_NAME] for column in cursor.description - ] + mysql_dict[query_hash][Constants.CONNECTION] = connection + mysql_dict[query_hash][Constants.CURSOR] = cursor + mysql_dict[query_hash][Constants.COLUMN_NAMES] = [ + column[Constants.I_COLUMN_NAME] for column in cursor.description + ] def mysql( @@ -105,24 +131,43 @@ def mysql( """ global mysql_dict - thread_id = threading.get_native_id() - cursor = mysql_dict[thread_id][Constants.CURSOR] - column_names = mysql_dict[thread_id][Constants.COLUMN_NAMES] + if len(config_path) > 0: + config = _combine_config(config=config, config_path=config_path) + + if _query_is_table(table_or_sql): + table_or_sql = f"SELECT * FROM {table_or_sql};" + + query_hash = _get_query_hash(table_or_sql, config, params) + cursor = mysql_dict[query_hash][Constants.CURSOR] + column_names = mysql_dict[query_hash][Constants.COLUMN_NAMES] rows = cursor.fetchmany(Constants.BATCH_SIZE) - return [mgp.Record(row=_name_row_cells_mysql(row, column_names)) for row in rows] + result = [mgp.Record(row=_name_row_cells_mysql(row, column_names)) for row in rows] + + # if results are empty, cleanup the query since cleanup doesn't accept any parameters + if not result: + _cleanup_mysql_by_hash(query_hash) + + return result -def cleanup_migrate_mysql(): +def _cleanup_mysql_by_hash(query_hash: str): + """Internal cleanup function that takes a query hash.""" global mysql_dict + + if query_hash in mysql_dict: + mysql_dict[query_hash][Constants.CURSOR] = None + mysql_dict[query_hash][Constants.CONNECTION].commit() + mysql_dict[query_hash][Constants.CONNECTION].close() + mysql_dict[query_hash][Constants.CONNECTION] = None + mysql_dict[query_hash][Constants.COLUMN_NAMES] = None + mysql_dict.pop(query_hash, None) - thread_id = threading.get_native_id() - mysql_dict[thread_id][Constants.CURSOR] = None - mysql_dict[thread_id][Constants.CONNECTION].commit() - mysql_dict[thread_id][Constants.CONNECTION].close() - mysql_dict[thread_id][Constants.CONNECTION] = None - mysql_dict[thread_id][Constants.COLUMN_NAMES] = None + +def cleanup_migrate_mysql(): + """Cleanup function called by mgp framework (no parameters).""" + pass mgp.add_batch_read_proc(mysql, init_migrate_mysql, cleanup_migrate_mysql) @@ -151,23 +196,24 @@ def init_migrate_sql_server( if _query_is_table(table_or_sql): table_or_sql = f"SELECT * FROM {table_or_sql};" - thread_id = threading.get_native_id() - if thread_id not in sql_server_dict: - sql_server_dict[thread_id] = {} + query_hash = _get_query_hash(table_or_sql, config, params) - if Constants.CURSOR not in sql_server_dict[thread_id]: - sql_server_dict[thread_id][Constants.CURSOR] = None + # check if query is already running + if query_hash in sql_server_dict: + raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") - if sql_server_dict[thread_id][Constants.CURSOR] is None: - connection = pyodbc.connect(**config) - cursor = connection.cursor() - cursor.execute(table_or_sql, *params) + if query_hash not in sql_server_dict: + sql_server_dict[query_hash] = {} - sql_server_dict[thread_id][Constants.CONNECTION] = connection - sql_server_dict[thread_id][Constants.CURSOR] = cursor - sql_server_dict[thread_id][Constants.COLUMN_NAMES] = [ - column[Constants.I_COLUMN_NAME] for column in cursor.description - ] + connection = pyodbc.connect(**config) + cursor = connection.cursor() + cursor.execute(table_or_sql, *params) + + sql_server_dict[query_hash][Constants.CONNECTION] = connection + sql_server_dict[query_hash][Constants.CURSOR] = cursor + sql_server_dict[query_hash][Constants.COLUMN_NAMES] = [ + column[Constants.I_COLUMN_NAME] for column in cursor.description + ] def sql_server( @@ -193,23 +239,47 @@ def sql_server( """ global sql_server_dict - thread_id = threading.get_native_id() - cursor = sql_server_dict[thread_id][Constants.CURSOR] - column_names = sql_server_dict[thread_id][Constants.COLUMN_NAMES] + if params: + _check_params_type(params, (list, tuple)) + else: + params = [] + + if len(config_path) > 0: + config = _combine_config(config=config, config_path=config_path) + + if _query_is_table(table_or_sql): + table_or_sql = f"SELECT * FROM {table_or_sql};" + + query_hash = _get_query_hash(table_or_sql, config, params) + cursor = sql_server_dict[query_hash][Constants.CURSOR] + column_names = sql_server_dict[query_hash][Constants.COLUMN_NAMES] rows = cursor.fetchmany(Constants.BATCH_SIZE) - return [mgp.Record(row=_name_row_cells(row, column_names)) for row in rows] + result = [mgp.Record(row=_name_row_cells(row, column_names)) for row in rows] + + # if results are empty, cleanup the query since cleanup doesn't accept any parameters + if not result: + _cleanup_sql_server_by_hash(query_hash) + + return result -def cleanup_migrate_sql_server(): +def _cleanup_sql_server_by_hash(query_hash: str): + """Internal cleanup function that takes a query hash.""" global sql_server_dict + + if query_hash in sql_server_dict: + sql_server_dict[query_hash][Constants.CURSOR] = None + sql_server_dict[query_hash][Constants.CONNECTION].commit() + sql_server_dict[query_hash][Constants.CONNECTION].close() + sql_server_dict[query_hash][Constants.CONNECTION] = None + sql_server_dict[query_hash][Constants.COLUMN_NAMES] = None + sql_server_dict.pop(query_hash, None) - thread_id = threading.get_native_id() - sql_server_dict[thread_id][Constants.CURSOR] = None - sql_server_dict[thread_id][Constants.CONNECTION].commit() - sql_server_dict[thread_id][Constants.CONNECTION].close() - sql_server_dict[thread_id][Constants.CONNECTION] = None - sql_server_dict[thread_id][Constants.COLUMN_NAMES] = None + +def cleanup_migrate_sql_server(): + """Cleanup function called by mgp framework (no parameters).""" + pass mgp.add_batch_read_proc(sql_server, init_migrate_sql_server, cleanup_migrate_sql_server) @@ -242,29 +312,30 @@ def init_migrate_oracle_db( # To prevent query execution from hanging config["disable_oob"] = True - thread_id = threading.get_native_id() - if thread_id not in oracle_db_dict: - oracle_db_dict[thread_id] = {} + query_hash = _get_query_hash(table_or_sql, config, params) - if Constants.CURSOR not in oracle_db_dict[thread_id]: - oracle_db_dict[thread_id][Constants.CURSOR] = None + # check if query is already running + if query_hash in oracle_db_dict: + raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") - if oracle_db_dict[thread_id][Constants.CURSOR] is None: - connection = oracledb.connect(**config) - cursor = connection.cursor() + if query_hash not in oracle_db_dict: + oracle_db_dict[query_hash] = {} - if not params: - cursor.execute(table_or_sql) - elif isinstance(params, (list, tuple)): - cursor.execute(table_or_sql, params) - else: - cursor.execute(table_or_sql, **params) + connection = oracledb.connect(**config) + cursor = connection.cursor() - oracle_db_dict[thread_id][Constants.CONNECTION] = connection - oracle_db_dict[thread_id][Constants.CURSOR] = cursor - oracle_db_dict[thread_id][Constants.COLUMN_NAMES] = [ - column[Constants.I_COLUMN_NAME] for column in cursor.description - ] + if not params: + cursor.execute(table_or_sql) + elif isinstance(params, (list, tuple)): + cursor.execute(table_or_sql, params) + else: + cursor.execute(table_or_sql, **params) + + oracle_db_dict[query_hash][Constants.CONNECTION] = connection + oracle_db_dict[query_hash][Constants.CURSOR] = cursor + oracle_db_dict[query_hash][Constants.COLUMN_NAMES] = [ + column[Constants.I_COLUMN_NAME] for column in cursor.description + ] def oracle_db( @@ -291,23 +362,46 @@ def oracle_db( global oracle_db_dict - thread_id = threading.get_native_id() - cursor = oracle_db_dict[thread_id][Constants.CURSOR] - column_names = oracle_db_dict[thread_id][Constants.COLUMN_NAMES] + if len(config_path) > 0: + config = _combine_config(config=config, config_path=config_path) + + if _query_is_table(table_or_sql): + table_or_sql = f"SELECT * FROM {table_or_sql}" + + if not config: + config = {} + config["disable_oob"] = True + + query_hash = _get_query_hash(table_or_sql, config, params) + cursor = oracle_db_dict[query_hash][Constants.CURSOR] + column_names = oracle_db_dict[query_hash][Constants.COLUMN_NAMES] rows = cursor.fetchmany(Constants.BATCH_SIZE) - return [mgp.Record(row=_name_row_cells(row, column_names)) for row in rows] + result = [mgp.Record(row=_name_row_cells(row, column_names)) for row in rows] + + # if results are empty, cleanup the query since cleanup doesn't accept any parameters + if not result: + _cleanup_oracle_db_by_hash(query_hash) + + return result -def cleanup_migrate_oracle_db(): +def _cleanup_oracle_db_by_hash(query_hash: str): + """Internal cleanup function that takes a query hash.""" global oracle_db_dict + + if query_hash in oracle_db_dict: + oracle_db_dict[query_hash][Constants.CURSOR] = None + oracle_db_dict[query_hash][Constants.CONNECTION].commit() + oracle_db_dict[query_hash][Constants.CONNECTION].close() + oracle_db_dict[query_hash][Constants.CONNECTION] = None + oracle_db_dict[query_hash][Constants.COLUMN_NAMES] = None + oracle_db_dict.pop(query_hash, None) - thread_id = threading.get_native_id() - oracle_db_dict[thread_id][Constants.CURSOR] = None - oracle_db_dict[thread_id][Constants.CONNECTION].commit() - oracle_db_dict[thread_id][Constants.CONNECTION].close() - oracle_db_dict[thread_id][Constants.CONNECTION] = None - oracle_db_dict[thread_id][Constants.COLUMN_NAMES] = None + +def cleanup_migrate_oracle_db(): + """Cleanup function called by mgp framework (no parameters).""" + pass mgp.add_batch_read_proc(oracle_db, init_migrate_oracle_db, cleanup_migrate_oracle_db) @@ -336,23 +430,24 @@ def init_migrate_postgresql( if _query_is_table(table_or_sql): table_or_sql = f"SELECT * FROM {table_or_sql};" - thread_id = threading.get_native_id() - if thread_id not in postgres_dict: - postgres_dict[thread_id] = {} + query_hash = _get_query_hash(table_or_sql, config, params) - if Constants.CURSOR not in postgres_dict[thread_id]: - postgres_dict[thread_id][Constants.CURSOR] = None + # check if query is already running + if query_hash in postgres_dict: + raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") - if postgres_dict[thread_id][Constants.CURSOR] is None: - connection = psycopg2.connect(**config) - cursor = connection.cursor() - cursor.execute(table_or_sql, params) + if query_hash not in postgres_dict: + postgres_dict[query_hash] = {} + + connection = psycopg2.connect(**config) + cursor = connection.cursor() + cursor.execute(table_or_sql, params) - postgres_dict[thread_id][Constants.CONNECTION] = connection - postgres_dict[thread_id][Constants.CURSOR] = cursor - postgres_dict[thread_id][Constants.COLUMN_NAMES] = [ - column.name for column in cursor.description - ] + postgres_dict[query_hash][Constants.CONNECTION] = connection + postgres_dict[query_hash][Constants.CURSOR] = cursor + postgres_dict[query_hash][Constants.COLUMN_NAMES] = [ + column.name for column in cursor.description + ] def postgresql( @@ -378,24 +473,48 @@ def postgresql( """ global postgres_dict - thread_id = threading.get_native_id() - cursor = postgres_dict[thread_id][Constants.CURSOR] - column_names = postgres_dict[thread_id][Constants.COLUMN_NAMES] + if params: + _check_params_type(params, (list, tuple)) + else: + params = [] + + if len(config_path) > 0: + config = _combine_config(config=config, config_path=config_path) + + if _query_is_table(table_or_sql): + table_or_sql = f"SELECT * FROM {table_or_sql};" + + query_hash = _get_query_hash(table_or_sql, config, params) + cursor = postgres_dict[query_hash][Constants.CURSOR] + column_names = postgres_dict[query_hash][Constants.COLUMN_NAMES] rows = cursor.fetchmany(Constants.BATCH_SIZE) - return [mgp.Record(row=_name_row_cells(row, column_names)) for row in rows] + result = [mgp.Record(row=_name_row_cells(row, column_names)) for row in rows] + + # if results are empty, cleanup the query since cleanup doesn't accept any parameters + if not result: + _cleanup_postgresql_by_hash(query_hash) + + return result -def cleanup_migrate_postgresql(): +def _cleanup_postgresql_by_hash(query_hash: str): + """Internal cleanup function that takes a query hash.""" global postgres_dict + + if query_hash in postgres_dict: + postgres_dict[query_hash][Constants.CURSOR] = None + postgres_dict[query_hash][Constants.CONNECTION].commit() + postgres_dict[query_hash][Constants.CONNECTION].close() + postgres_dict[query_hash][Constants.CONNECTION] = None + postgres_dict[query_hash][Constants.COLUMN_NAMES] = None + postgres_dict.pop(query_hash, None) - thread_id = threading.get_native_id() - postgres_dict[thread_id][Constants.CURSOR] = None - postgres_dict[thread_id][Constants.CONNECTION].commit() - postgres_dict[thread_id][Constants.CONNECTION].close() - postgres_dict[thread_id][Constants.CONNECTION] = None - postgres_dict[thread_id][Constants.COLUMN_NAMES] = None + +def cleanup_migrate_postgresql(): + """Cleanup function called by mgp framework (no parameters).""" + pass mgp.add_batch_read_proc(postgresql, init_migrate_postgresql, cleanup_migrate_postgresql) @@ -432,6 +551,12 @@ def init_migrate_s3( bucket_name, *key_parts = file_path_no_protocol.split("/") s3_key = "/".join(key_parts) + query_hash = _get_query_hash(file_path, config) + + # check if query is already running + if query_hash in s3_dict: + raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") + # Initialize S3 client s3_client = boto3.client( "s3", @@ -456,12 +581,9 @@ def init_migrate_s3( csv_reader = csv.reader(text_stream) column_names = next(csv_reader) # First row contains column names - thread_id = threading.get_native_id() - if thread_id not in s3_dict: - s3_dict[thread_id] = {} - - s3_dict[thread_id][Constants.CURSOR] = csv_reader - s3_dict[thread_id][Constants.COLUMN_NAMES] = column_names + s3_dict[query_hash] = {} + s3_dict[query_hash][Constants.CURSOR] = csv_reader + s3_dict[query_hash][Constants.COLUMN_NAMES] = column_names def s3( @@ -480,9 +602,12 @@ def s3( """ global s3_dict - thread_id = threading.get_native_id() - csv_reader = s3_dict[thread_id][Constants.CURSOR] - column_names = s3_dict[thread_id][Constants.COLUMN_NAMES] + if len(config_path) > 0: + config = _combine_config(config=config, config_path=config_path) + + query_hash = _get_query_hash(file_path, config) + csv_reader = s3_dict[query_hash][Constants.CURSOR] + column_names = s3_dict[query_hash][Constants.COLUMN_NAMES] batch_rows = [] for _ in range(Constants.BATCH_SIZE): @@ -492,17 +617,24 @@ def s3( except StopIteration: break + # if results are empty, cleanup the query since cleanup doesn't accept any parameters + if not batch_rows: + _cleanup_s3_by_hash(query_hash) + return batch_rows -def cleanup_migrate_s3(): - """ - Clean up S3 dictionary references per-thread. - """ +def _cleanup_s3_by_hash(query_hash: str): + """Internal cleanup function that takes a query hash.""" global s3_dict + + if query_hash in s3_dict: + s3_dict.pop(query_hash, None) + - thread_id = threading.get_native_id() - s3_dict.pop(thread_id, None) +def cleanup_migrate_s3(): + """Cleanup function called by mgp framework (no parameters).""" + pass mgp.add_batch_read_proc(s3, init_migrate_s3, cleanup_migrate_s3) @@ -519,13 +651,16 @@ def init_migrate_neo4j( ): global neo4j_dict - thread_id = threading.get_native_id() - if thread_id not in neo4j_dict: - neo4j_dict[thread_id] = {} - if len(config_path) > 0: config = _combine_config(config=config, config_path=config_path) + query = _formulate_cypher_query(label_or_rel_or_query) + query_hash = _get_query_hash(query, config, params) + + # check if query is already running + if query_hash in neo4j_dict: + raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") + uri = _build_neo4j_uri(config) username = config.get(Constants.USERNAME, "neo4j") password = config.get(Constants.PASSWORD, "password") @@ -539,14 +674,14 @@ def init_migrate_neo4j( else: session = driver.session() - query = _formulate_cypher_query(label_or_rel_or_query) # Neo4j expects params to be a dict or None cypher_params = params if params is not None else {} result = session.run(query, parameters=cypher_params) - neo4j_dict[thread_id][Constants.DRIVER] = driver - neo4j_dict[thread_id][Constants.SESSION] = session - neo4j_dict[thread_id][Constants.RESULT] = result + neo4j_dict[query_hash] = {} + neo4j_dict[query_hash][Constants.DRIVER] = driver + neo4j_dict[query_hash][Constants.SESSION] = session + neo4j_dict[query_hash][Constants.RESULT] = result def neo4j( @@ -566,8 +701,12 @@ def neo4j( """ global neo4j_dict - thread_id = threading.get_native_id() - result = neo4j_dict[thread_id][Constants.RESULT] + if len(config_path) > 0: + config = _combine_config(config=config, config_path=config_path) + + query = _formulate_cypher_query(label_or_rel_or_query) + query_hash = _get_query_hash(query, config, params) + result = neo4j_dict[query_hash][Constants.RESULT] # Fetch up to BATCH_SIZE records batch = [] @@ -579,20 +718,30 @@ def neo4j( if len(batch) >= Constants.BATCH_SIZE: break + # if results are empty, cleanup the query since cleanup doesn't accept any parameters + if not batch: + _cleanup_neo4j_by_hash(query_hash) + return batch -def cleanup_migrate_neo4j(): +def _cleanup_neo4j_by_hash(query_hash: str): + """Internal cleanup function that takes a query hash.""" global neo4j_dict + + if query_hash in neo4j_dict: + session = neo4j_dict[query_hash].get(Constants.SESSION) + driver = neo4j_dict[query_hash].get(Constants.DRIVER) + if session: + session.close() + if driver: + driver.close() + neo4j_dict.pop(query_hash, None) - thread_id = threading.get_native_id() - session = neo4j_dict[thread_id].get(Constants.SESSION) - driver = neo4j_dict[thread_id].get(Constants.DRIVER) - if session: - session.close() - if driver: - driver.close() - neo4j_dict.pop(thread_id, None) + +def cleanup_migrate_neo4j(): + """Cleanup function called by mgp framework (no parameters).""" + pass mgp.add_batch_read_proc(neo4j, init_migrate_neo4j, cleanup_migrate_neo4j) @@ -612,6 +761,12 @@ def init_migrate_arrow_flight( if len(config_path) > 0: config = _combine_config(config=config, config_path=config_path) + query_hash = _get_query_hash(query, config) + + # check if query is already running + if query_hash in flight_dict: + raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") + host = config.get(Constants.HOST, None) port = config.get(Constants.PORT, None) username = config.get(Constants.USERNAME, "") @@ -633,13 +788,9 @@ def init_migrate_arrow_flight( flight.FlightDescriptor.for_command(query), options ) - # Store connection per thread - thread_id = threading.get_native_id() - if thread_id not in flight_dict: - flight_dict[thread_id] = {} - - flight_dict[thread_id][Constants.CONNECTION] = client - flight_dict[thread_id][Constants.CURSOR] = iter( + flight_dict[query_hash] = {} + flight_dict[query_hash][Constants.CONNECTION] = client + flight_dict[query_hash][Constants.CURSOR] = iter( _fetch_flight_data(client, flight_info, options) ) @@ -671,8 +822,11 @@ def arrow_flight( """ global flight_dict - thread_id = threading.get_native_id() - cursor = flight_dict[thread_id][Constants.CURSOR] + if len(config_path) > 0: + config = _combine_config(config=config, config_path=config_path) + + query_hash = _get_query_hash(query, config) + cursor = flight_dict[query_hash][Constants.CURSOR] batch = [] for _ in range(Constants.BATCH_SIZE): try: @@ -681,18 +835,24 @@ def arrow_flight( except StopIteration: break + # if results are empty, cleanup the query since cleanup doesn't accept any parameters + if not batch: + _cleanup_arrow_flight_by_hash(query_hash) + return batch -def cleanup_migrate_arrow_flight(): - """ - Close the Flight connection per-thread. - """ +def _cleanup_arrow_flight_by_hash(query_hash: str): + """Internal cleanup function that takes a query hash.""" global flight_dict + + if query_hash in flight_dict: + flight_dict.pop(query_hash, None) - thread_id = threading.get_native_id() - if thread_id in flight_dict: - flight_dict.pop(thread_id, None) + +def cleanup_migrate_arrow_flight(): + """Cleanup function called by mgp framework (no parameters).""" + pass mgp.add_batch_read_proc( @@ -709,16 +869,19 @@ def init_migrate_duckdb(query: str, setup_queries: mgp.Nullable[List[str]] = Non Initialize an in-memory DuckDB connection and execute the query. :param query: SQL query to execute - :param config: Unused but kept for consistency with other migration functions - :param config_path: Unused but kept for consistency with other migration functions + :param setup_queries: Optional list of setup queries to execute before the main query """ global duckdb_dict - thread_id = threading.get_native_id() - if thread_id not in duckdb_dict: - duckdb_dict[thread_id] = {} + # Create hash from query and setup_queries + setup_queries_str = json.dumps(setup_queries, sort_keys=False) if setup_queries else "" + query_hash = hashlib.sha256(f"{query}|{setup_queries_str}".encode('utf-8')).hexdigest() - # Ensure a fresh in-memory DuckDB instance for each thread + # check if query is already running + if query_hash in duckdb_dict: + raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") + + # Ensure a fresh in-memory DuckDB instance for each query connection = duckDB.connect() cursor = connection.cursor() if setup_queries is not None: @@ -727,9 +890,10 @@ def init_migrate_duckdb(query: str, setup_queries: mgp.Nullable[List[str]] = Non cursor.execute(query) - duckdb_dict[thread_id][Constants.CONNECTION] = connection - duckdb_dict[thread_id][Constants.CURSOR] = cursor - duckdb_dict[thread_id][Constants.COLUMN_NAMES] = [ + duckdb_dict[query_hash] = {} + duckdb_dict[query_hash][Constants.CONNECTION] = connection + duckdb_dict[query_hash][Constants.CURSOR] = cursor + duckdb_dict[query_hash][Constants.COLUMN_NAMES] = [ desc[0] for desc in cursor.description ] @@ -741,31 +905,39 @@ def duckdb( Fetch rows from DuckDB in batches. :param query: SQL query to execute - :param config: Unused but kept for consistency with other migration functions - :param config_path: Unused but kept for consistency with other migration functions + :param setup_queries: Optional list of setup queries to execute before the main query :return: The result table as a stream of rows """ global duckdb_dict - thread_id = threading.get_native_id() - cursor = duckdb_dict[thread_id][Constants.CURSOR] - column_names = duckdb_dict[thread_id][Constants.COLUMN_NAMES] + setup_queries_str = json.dumps(setup_queries, sort_keys=False) if setup_queries else "" + query_hash = hashlib.sha256(f"{query}|{setup_queries_str}".encode('utf-8')).hexdigest() + cursor = duckdb_dict[query_hash][Constants.CURSOR] + column_names = duckdb_dict[query_hash][Constants.COLUMN_NAMES] rows = cursor.fetchmany(Constants.BATCH_SIZE) - return [mgp.Record(row=_name_row_cells(row, column_names)) for row in rows] + result = [mgp.Record(row=_name_row_cells(row, column_names)) for row in rows] + + # if results are empty, cleanup the query since cleanup doesn't accept any parameters + if not result: + _cleanup_duckdb_by_hash(query_hash) + + return result -def cleanup_migrate_duckdb(): - """ - Clean up DuckDB dictionary references per-thread. - """ +def _cleanup_duckdb_by_hash(query_hash: str): + """Internal cleanup function that takes a query hash.""" global duckdb_dict + + if query_hash in duckdb_dict: + if Constants.CONNECTION in duckdb_dict[query_hash]: + duckdb_dict[query_hash][Constants.CONNECTION].close() + duckdb_dict.pop(query_hash, None) - thread_id = threading.get_native_id() - if thread_id in duckdb_dict: - if Constants.CONNECTION in duckdb_dict[thread_id]: - duckdb_dict[thread_id][Constants.CONNECTION].close() - duckdb_dict.pop(thread_id, None) + +def cleanup_migrate_duckdb(): + """Cleanup function called by mgp framework (no parameters).""" + pass mgp.add_batch_read_proc(duckdb, init_migrate_duckdb, cleanup_migrate_duckdb) @@ -782,19 +954,22 @@ def init_migrate_memgraph( ): global memgraph_dict - thread_id = threading.get_native_id() - if thread_id not in memgraph_dict: - memgraph_dict[thread_id] = {} - if len(config_path) > 0: config = _combine_config(config=config, config_path=config_path) - memgraph_db = Memgraph(**config) query = _formulate_cypher_query(label_or_rel_or_query) + query_hash = _get_query_hash(query, config, params) + + # check if query is already running + if query_hash in memgraph_dict: + raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") + + memgraph_db = Memgraph(**config) cursor = memgraph_db.execute_and_fetch(query, params) - memgraph_dict[thread_id][Constants.CONNECTION] = memgraph_db - memgraph_dict[thread_id][Constants.CURSOR] = cursor + memgraph_dict[query_hash] = {} + memgraph_dict[query_hash][Constants.CONNECTION] = memgraph_db + memgraph_dict[query_hash][Constants.CURSOR] = cursor def memgraph( @@ -814,23 +989,39 @@ def memgraph( """ global memgraph_dict - thread_id = threading.get_native_id() - cursor = memgraph_dict[thread_id][Constants.CURSOR] + if len(config_path) > 0: + config = _combine_config(config=config, config_path=config_path) + + query = _formulate_cypher_query(label_or_rel_or_query) + query_hash = _get_query_hash(query, config, params) + cursor = memgraph_dict[query_hash][Constants.CURSOR] - return [ + result = [ mgp.Record(row=row) for row in (next(cursor, None) for _ in range(Constants.BATCH_SIZE)) if row is not None ] + + # if results are empty, cleanup the query since cleanup doesn't accept any parameters + if not result: + _cleanup_memgraph_by_hash(query_hash) + + return result -def cleanup_migrate_memgraph(): +def _cleanup_memgraph_by_hash(query_hash: str): + """Internal cleanup function that takes a query hash.""" global memgraph_dict + + if query_hash in memgraph_dict: + if Constants.CONNECTION in memgraph_dict[query_hash]: + memgraph_dict[query_hash][Constants.CONNECTION].close() + memgraph_dict.pop(query_hash, None) - thread_id = threading.get_native_id() - if Constants.CONNECTION in memgraph_dict[thread_id]: - memgraph_dict[thread_id][Constants.CONNECTION].close() - memgraph_dict.pop(thread_id, None) + +def cleanup_migrate_memgraph(): + """Cleanup function called by mgp framework (no parameters).""" + pass mgp.add_batch_read_proc(memgraph, init_migrate_memgraph, cleanup_migrate_memgraph) @@ -858,6 +1049,12 @@ def init_migrate_servicenow( if len(config_path) > 0: config = _combine_config(config=config, config_path=config_path) + query_hash = _get_query_hash(endpoint, config, params) + + # check if query is already running + if query_hash in servicenow_dict: + raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") + auth = (config.get(Constants.USERNAME), config.get(Constants.PASSWORD)) headers = {"Accept": "application/json"} @@ -868,11 +1065,8 @@ def init_migrate_servicenow( if not data: raise ValueError("No data found in ServiceNow response") - thread_id = threading.get_native_id() - if thread_id not in servicenow_dict: - servicenow_dict[thread_id] = {} - - servicenow_dict[thread_id][Constants.CURSOR] = iter(data) + servicenow_dict[query_hash] = {} + servicenow_dict[query_hash][Constants.CURSOR] = iter(data) def servicenow( @@ -892,8 +1086,11 @@ def servicenow( """ global servicenow_dict - thread_id = threading.get_native_id() - data_iter = servicenow_dict[thread_id][Constants.CURSOR] + if len(config_path) > 0: + config = _combine_config(config=config, config_path=config_path) + + query_hash = _get_query_hash(endpoint, config, params) + data_iter = servicenow_dict[query_hash][Constants.CURSOR] batch_rows = [] for _ in range(Constants.BATCH_SIZE): @@ -903,17 +1100,24 @@ def servicenow( except StopIteration: break + # if results are empty, cleanup the query since cleanup doesn't accept any parameters + if not batch_rows: + _cleanup_servicenow_by_hash(query_hash) + return batch_rows -def cleanup_migrate_servicenow(): - """ - Clean up ServiceNow dictionary references per-thread. - """ +def _cleanup_servicenow_by_hash(query_hash: str): + """Internal cleanup function that takes a query hash.""" global servicenow_dict + + if query_hash in servicenow_dict: + servicenow_dict.pop(query_hash, None) - thread_id = threading.get_native_id() - servicenow_dict.pop(thread_id, None) + +def cleanup_migrate_servicenow(): + """Cleanup function called by mgp framework (no parameters).""" + pass mgp.add_batch_read_proc(servicenow, init_migrate_servicenow, cleanup_migrate_servicenow) From 32513d52c322cbec48cb249f27a66652730a21d5 Mon Sep 17 00:00:00 2001 From: imilinovic Date: Mon, 5 Jan 2026 17:05:32 +0100 Subject: [PATCH 2/7] check --- python/migrate.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/migrate.py b/python/migrate.py index b514ddeab..aa6d585e7 100644 --- a/python/migrate.py +++ b/python/migrate.py @@ -50,7 +50,7 @@ def _get_query_hash(query: str, config: mgp.Map, params: mgp.Nullable[mgp.Any] = :param config: Configuration map :param params: Optional query parameters """ - config_dict = dict(config) if config else {} + config_dict = dict(config) config_str = json.dumps(config_dict, sort_keys=True, default=str) params_str = "" From 194d34f9e65b76c769d44177dffff762854a0ba7 Mon Sep 17 00:00:00 2001 From: imilinovic Date: Mon, 5 Jan 2026 17:16:53 +0100 Subject: [PATCH 3/7] run black --- python/migrate.py | 118 +++++++++++++++++++++++++++++----------------- 1 file changed, 74 insertions(+), 44 deletions(-) diff --git a/python/migrate.py b/python/migrate.py index aa6d585e7..88d764f2c 100644 --- a/python/migrate.py +++ b/python/migrate.py @@ -42,10 +42,12 @@ class Constants: USERNAME = "username" -def _get_query_hash(query: str, config: mgp.Map, params: mgp.Nullable[mgp.Any] = None) -> str: +def _get_query_hash( + query: str, config: mgp.Map, params: mgp.Nullable[mgp.Any] = None +) -> str: """ Create a hash from query, config, and params to use as a cache key. - + :param query: The query string (or table name, endpoint, file path, etc.) :param config: Configuration map :param params: Optional query parameters @@ -61,9 +63,9 @@ def _get_query_hash(query: str, config: mgp.Map, params: mgp.Nullable[mgp.Any] = params_str = json.dumps(list(params), sort_keys=False, default=str) else: params_str = str(params) - + hash_input = f"{query}|{config_str}|{params_str}" - return hashlib.sha256(hash_input.encode('utf-8')).hexdigest() + return hashlib.sha256(hash_input.encode("utf-8")).hexdigest() # MYSQL @@ -91,7 +93,9 @@ def init_migrate_mysql( # check if query is already running if query_hash in mysql_dict: - raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") + raise RuntimeError( + f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one." + ) if query_hash not in mysql_dict: mysql_dict[query_hash] = {} @@ -144,18 +148,18 @@ def mysql( rows = cursor.fetchmany(Constants.BATCH_SIZE) result = [mgp.Record(row=_name_row_cells_mysql(row, column_names)) for row in rows] - + # if results are empty, cleanup the query since cleanup doesn't accept any parameters if not result: _cleanup_mysql_by_hash(query_hash) - + return result def _cleanup_mysql_by_hash(query_hash: str): """Internal cleanup function that takes a query hash.""" global mysql_dict - + if query_hash in mysql_dict: mysql_dict[query_hash][Constants.CURSOR] = None mysql_dict[query_hash][Constants.CONNECTION].commit() @@ -200,7 +204,9 @@ def init_migrate_sql_server( # check if query is already running if query_hash in sql_server_dict: - raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") + raise RuntimeError( + f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one." + ) if query_hash not in sql_server_dict: sql_server_dict[query_hash] = {} @@ -256,18 +262,18 @@ def sql_server( rows = cursor.fetchmany(Constants.BATCH_SIZE) result = [mgp.Record(row=_name_row_cells(row, column_names)) for row in rows] - + # if results are empty, cleanup the query since cleanup doesn't accept any parameters if not result: _cleanup_sql_server_by_hash(query_hash) - + return result def _cleanup_sql_server_by_hash(query_hash: str): """Internal cleanup function that takes a query hash.""" global sql_server_dict - + if query_hash in sql_server_dict: sql_server_dict[query_hash][Constants.CURSOR] = None sql_server_dict[query_hash][Constants.CONNECTION].commit() @@ -316,7 +322,9 @@ def init_migrate_oracle_db( # check if query is already running if query_hash in oracle_db_dict: - raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") + raise RuntimeError( + f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one." + ) if query_hash not in oracle_db_dict: oracle_db_dict[query_hash] = {} @@ -378,18 +386,18 @@ def oracle_db( rows = cursor.fetchmany(Constants.BATCH_SIZE) result = [mgp.Record(row=_name_row_cells(row, column_names)) for row in rows] - + # if results are empty, cleanup the query since cleanup doesn't accept any parameters if not result: _cleanup_oracle_db_by_hash(query_hash) - + return result def _cleanup_oracle_db_by_hash(query_hash: str): """Internal cleanup function that takes a query hash.""" global oracle_db_dict - + if query_hash in oracle_db_dict: oracle_db_dict[query_hash][Constants.CURSOR] = None oracle_db_dict[query_hash][Constants.CONNECTION].commit() @@ -434,7 +442,9 @@ def init_migrate_postgresql( # check if query is already running if query_hash in postgres_dict: - raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") + raise RuntimeError( + f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one." + ) if query_hash not in postgres_dict: postgres_dict[query_hash] = {} @@ -491,18 +501,18 @@ def postgresql( rows = cursor.fetchmany(Constants.BATCH_SIZE) result = [mgp.Record(row=_name_row_cells(row, column_names)) for row in rows] - + # if results are empty, cleanup the query since cleanup doesn't accept any parameters if not result: _cleanup_postgresql_by_hash(query_hash) - + return result def _cleanup_postgresql_by_hash(query_hash: str): """Internal cleanup function that takes a query hash.""" global postgres_dict - + if query_hash in postgres_dict: postgres_dict[query_hash][Constants.CURSOR] = None postgres_dict[query_hash][Constants.CONNECTION].commit() @@ -555,7 +565,9 @@ def init_migrate_s3( # check if query is already running if query_hash in s3_dict: - raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") + raise RuntimeError( + f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one." + ) # Initialize S3 client s3_client = boto3.client( @@ -620,14 +632,14 @@ def s3( # if results are empty, cleanup the query since cleanup doesn't accept any parameters if not batch_rows: _cleanup_s3_by_hash(query_hash) - + return batch_rows def _cleanup_s3_by_hash(query_hash: str): """Internal cleanup function that takes a query hash.""" global s3_dict - + if query_hash in s3_dict: s3_dict.pop(query_hash, None) @@ -659,7 +671,9 @@ def init_migrate_neo4j( # check if query is already running if query_hash in neo4j_dict: - raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") + raise RuntimeError( + f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one." + ) uri = _build_neo4j_uri(config) username = config.get(Constants.USERNAME, "neo4j") @@ -721,14 +735,14 @@ def neo4j( # if results are empty, cleanup the query since cleanup doesn't accept any parameters if not batch: _cleanup_neo4j_by_hash(query_hash) - + return batch def _cleanup_neo4j_by_hash(query_hash: str): """Internal cleanup function that takes a query hash.""" global neo4j_dict - + if query_hash in neo4j_dict: session = neo4j_dict[query_hash].get(Constants.SESSION) driver = neo4j_dict[query_hash].get(Constants.DRIVER) @@ -765,7 +779,9 @@ def init_migrate_arrow_flight( # check if query is already running if query_hash in flight_dict: - raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") + raise RuntimeError( + f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one." + ) host = config.get(Constants.HOST, None) port = config.get(Constants.PORT, None) @@ -838,14 +854,14 @@ def arrow_flight( # if results are empty, cleanup the query since cleanup doesn't accept any parameters if not batch: _cleanup_arrow_flight_by_hash(query_hash) - + return batch def _cleanup_arrow_flight_by_hash(query_hash: str): """Internal cleanup function that takes a query hash.""" global flight_dict - + if query_hash in flight_dict: flight_dict.pop(query_hash, None) @@ -874,12 +890,18 @@ def init_migrate_duckdb(query: str, setup_queries: mgp.Nullable[List[str]] = Non global duckdb_dict # Create hash from query and setup_queries - setup_queries_str = json.dumps(setup_queries, sort_keys=False) if setup_queries else "" - query_hash = hashlib.sha256(f"{query}|{setup_queries_str}".encode('utf-8')).hexdigest() + setup_queries_str = ( + json.dumps(setup_queries, sort_keys=False) if setup_queries else "" + ) + query_hash = hashlib.sha256( + f"{query}|{setup_queries_str}".encode("utf-8") + ).hexdigest() # check if query is already running if query_hash in duckdb_dict: - raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") + raise RuntimeError( + f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one." + ) # Ensure a fresh in-memory DuckDB instance for each query connection = duckDB.connect() @@ -910,25 +932,29 @@ def duckdb( """ global duckdb_dict - setup_queries_str = json.dumps(setup_queries, sort_keys=False) if setup_queries else "" - query_hash = hashlib.sha256(f"{query}|{setup_queries_str}".encode('utf-8')).hexdigest() + setup_queries_str = ( + json.dumps(setup_queries, sort_keys=False) if setup_queries else "" + ) + query_hash = hashlib.sha256( + f"{query}|{setup_queries_str}".encode("utf-8") + ).hexdigest() cursor = duckdb_dict[query_hash][Constants.CURSOR] column_names = duckdb_dict[query_hash][Constants.COLUMN_NAMES] rows = cursor.fetchmany(Constants.BATCH_SIZE) result = [mgp.Record(row=_name_row_cells(row, column_names)) for row in rows] - + # if results are empty, cleanup the query since cleanup doesn't accept any parameters if not result: _cleanup_duckdb_by_hash(query_hash) - + return result def _cleanup_duckdb_by_hash(query_hash: str): """Internal cleanup function that takes a query hash.""" global duckdb_dict - + if query_hash in duckdb_dict: if Constants.CONNECTION in duckdb_dict[query_hash]: duckdb_dict[query_hash][Constants.CONNECTION].close() @@ -962,7 +988,9 @@ def init_migrate_memgraph( # check if query is already running if query_hash in memgraph_dict: - raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") + raise RuntimeError( + f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one." + ) memgraph_db = Memgraph(**config) cursor = memgraph_db.execute_and_fetch(query, params) @@ -1001,18 +1029,18 @@ def memgraph( for row in (next(cursor, None) for _ in range(Constants.BATCH_SIZE)) if row is not None ] - + # if results are empty, cleanup the query since cleanup doesn't accept any parameters if not result: _cleanup_memgraph_by_hash(query_hash) - + return result def _cleanup_memgraph_by_hash(query_hash: str): """Internal cleanup function that takes a query hash.""" global memgraph_dict - + if query_hash in memgraph_dict: if Constants.CONNECTION in memgraph_dict[query_hash]: memgraph_dict[query_hash][Constants.CONNECTION].close() @@ -1053,7 +1081,9 @@ def init_migrate_servicenow( # check if query is already running if query_hash in servicenow_dict: - raise RuntimeError(f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one.") + raise RuntimeError( + f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one." + ) auth = (config.get(Constants.USERNAME), config.get(Constants.PASSWORD)) headers = {"Accept": "application/json"} @@ -1103,14 +1133,14 @@ def servicenow( # if results are empty, cleanup the query since cleanup doesn't accept any parameters if not batch_rows: _cleanup_servicenow_by_hash(query_hash) - + return batch_rows def _cleanup_servicenow_by_hash(query_hash: str): """Internal cleanup function that takes a query hash.""" global servicenow_dict - + if query_hash in servicenow_dict: servicenow_dict.pop(query_hash, None) From 92d844b8db809a7bc9c802843d4862c205b29b00 Mon Sep 17 00:00:00 2001 From: imilinovic Date: Mon, 5 Jan 2026 18:04:11 +0100 Subject: [PATCH 4/7] linter --- python/migrate.py | 1 - 1 file changed, 1 deletion(-) diff --git a/python/migrate.py b/python/migrate.py index 88d764f2c..a0d7c5578 100644 --- a/python/migrate.py +++ b/python/migrate.py @@ -6,7 +6,6 @@ import json import os import re -import threading from decimal import Decimal from typing import Any, Dict, List From 1e431bfa330297d3933e29f6cc55847c86da553b Mon Sep 17 00:00:00 2001 From: imilinovic Date: Thu, 8 Jan 2026 13:42:48 +0100 Subject: [PATCH 5/7] cleanup --- python/migrate.py | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/python/migrate.py b/python/migrate.py index a0d7c5578..ecb2d0e78 100644 --- a/python/migrate.py +++ b/python/migrate.py @@ -96,8 +96,7 @@ def init_migrate_mysql( f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one." ) - if query_hash not in mysql_dict: - mysql_dict[query_hash] = {} + mysql_dict[query_hash] = {} connection = mysql_connector.connect(**config) cursor = connection.cursor() @@ -207,8 +206,7 @@ def init_migrate_sql_server( f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one." ) - if query_hash not in sql_server_dict: - sql_server_dict[query_hash] = {} + sql_server_dict[query_hash] = {} connection = pyodbc.connect(**config) cursor = connection.cursor() @@ -325,8 +323,7 @@ def init_migrate_oracle_db( f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one." ) - if query_hash not in oracle_db_dict: - oracle_db_dict[query_hash] = {} + oracle_db_dict[query_hash] = {} connection = oracledb.connect(**config) cursor = connection.cursor() @@ -445,8 +442,7 @@ def init_migrate_postgresql( f"Migrate module with these parameters is already running. Please wait for it to finish before starting a new one." ) - if query_hash not in postgres_dict: - postgres_dict[query_hash] = {} + postgres_dict[query_hash] = {} connection = psycopg2.connect(**config) cursor = connection.cursor() From d9c2f4b944b28741a0188f656306137f0f90f64c Mon Sep 17 00:00:00 2001 From: imilinovic Date: Thu, 8 Jan 2026 13:54:34 +0100 Subject: [PATCH 6/7] cleanup --- python/migrate.py | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/python/migrate.py b/python/migrate.py index ecb2d0e78..30403e3ee 100644 --- a/python/migrate.py +++ b/python/migrate.py @@ -242,11 +242,6 @@ def sql_server( """ global sql_server_dict - if params: - _check_params_type(params, (list, tuple)) - else: - params = [] - if len(config_path) > 0: config = _combine_config(config=config, config_path=config_path) @@ -478,11 +473,6 @@ def postgresql( """ global postgres_dict - if params: - _check_params_type(params, (list, tuple)) - else: - params = [] - if len(config_path) > 0: config = _combine_config(config=config, config_path=config_path) From 482a937055f4379c8dd307b39ef87a843ab9389d Mon Sep 17 00:00:00 2001 From: imilinovic Date: Thu, 8 Jan 2026 14:10:51 +0100 Subject: [PATCH 7/7] cleanup --- python/migrate.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/python/migrate.py b/python/migrate.py index 30403e3ee..580ff4d18 100644 --- a/python/migrate.py +++ b/python/migrate.py @@ -242,6 +242,9 @@ def sql_server( """ global sql_server_dict + if not params: + params = [] + if len(config_path) > 0: config = _combine_config(config=config, config_path=config_path) @@ -473,6 +476,9 @@ def postgresql( """ global postgres_dict + if not params: + params = [] + if len(config_path) > 0: config = _combine_config(config=config, config_path=config_path)