Refactor SQL query handling in DbReplicator and MySQLApi for improved security and clarity

- Updated DbReplicator to pass raw primary key values to mysql_api, eliminating manual quote handling for parameterized queries.
- Enhanced MySQLApi to use parameterized queries for pagination, preventing SQL injection and improving query safety.
- Added detailed logging for query execution and parameters to aid in debugging and error handling.

Author: jaredmdobson

mysql_ch_replicator/db_replicator_initial.py

@@ -167,14 +167,9 @@ def perform_initial_replication_table(self, table_name):
 
         while True:
 
+            # Pass raw primary key values to mysql_api - it will handle proper SQL parameterization
+            # No need to manually add quotes - parameterized queries handle this safely
             query_start_values = max_primary_key
-            if query_start_values is not None:
-                for i in range(len(query_start_values)):
-                    key_type = primary_key_types[i]
-                    value = query_start_values[i]
-                    if 'int' not in key_type.lower():
-                        value = f"'{value}'"
-                        query_start_values[i] = value
 
             records = self.replicator.mysql_api.get_records(
                 table_name=table_name,

mysql_ch_replicator/mysql_api.py

@@ -92,10 +92,14 @@ def get_records(
             order_by_str = ",".join(order_by_escaped)
 
             where = ""
+            query_params = []
+
             if start_value is not None:
-                # Build the start_value condition for pagination
-                start_value_str = ",".join(map(str, start_value))
-                where = f"WHERE ({order_by_str}) > ({start_value_str}) "
+                # Build the start_value condition for pagination using parameterized query
+                # This prevents SQL injection and handles special characters properly
+                placeholders = ",".join(["%s"] * len(start_value))
+                where = f"WHERE ({order_by_str}) > ({placeholders}) "
+                query_params.extend(start_value)
 
             # Add partitioning filter for parallel processing (e.g., sharded crawling)
             if (
@@ -116,10 +120,23 @@ def get_records(
             # Construct final query
             query = f"SELECT * FROM `{table_name}` {where}ORDER BY {order_by_str} LIMIT {limit}"
 
+            # Log query details for debugging
             logger.debug(f"Executing query: {query}")
+            if query_params:
+                logger.debug(f"Query parameters: {query_params}")
 
-            # Execute the query
-            cursor.execute(query)
-            res = cursor.fetchall()
-            records = [x for x in res]
-            return records
+            # Execute the query with proper parameterization
+            try:
+                if query_params:
+                    cursor.execute(query, tuple(query_params))
+                else:
+                    cursor.execute(query)
+                res = cursor.fetchall()
+                records = [x for x in res]
+                return records
+            except Exception as e:
+                logger.error(f"Query execution failed: {query}")
+                if query_params:
+                    logger.error(f"Query parameters: {query_params}")
+                logger.error(f"Error details: {e}")
+                raise