Python中避免SQL冗余更新的问题排查与解决

需要通过记录表的增量更新追踪操作，但Python编写的SQL同步脚本在test_status无变化时仍执行更新操作。日志显示多条新旧test_status相同的更新记录，虽已添加状态相等则跳过的判断逻辑，但问题依旧。此前已解决插入部分的类似问题，更新部分的冗余执行问题仍存在。

异常日志 #

Updated record for accession_id: KCH2400337322, test_type: 35, new test_status: 3, old test_status: 3
Updated record for accession_id: KCH2400337323, test_type: 35, new test_status: 3, old test_status: 3
Updated record for accession_id: KCH2400337324, test_type: 35, new test_status: 3, old test_status: 3
Updated record for accession_id: KCH2400337325, test_type: 35, new test_status: 5, old test_status: 5
Updated record for accession_id: KCH2400337327, test_type: 35, new test_status: 5, old test_status: 5
Updated record for accession_id: KCH2400337328, test_type: 40, new test_status: 3, old test_status: 3
Updated record for accession_id: KCH2400337329, test_type: 35, new test_status: 3, old test_status: 3
Updated record for accession_id: KCH2400337329, test_type: 40, new test_status: 2, old test_status: 2

同步脚本代码（updateEntry.py） #

import mysql.connector
from helper import getDepartmentIdHelper, getTestTypeID
from config import testType1, testType2, testType3, testType4, interval

intvl = interval  # 100
department_id = getDepartmentIdHelper()  # 3
test_type_id1 = getTestTypeID(testType1)  # 35
test_type_id2 = getTestTypeID(testType2)  # 39
test_type_id3 = getTestTypeID(testType3)  # 40
test_type_id4 = getTestTypeID(testType4)  # 41

def updateEntries():
    try:
        # Connect to iBlissDB
        iblis_connection = mysql.connector.connect(
            host="127.0.0.1",
            port="3306",
            user="root",
            password="root",
            database="tests"
        )

        # Connect to srsDB
        srs_connection = mysql.connector.connect(
            host="127.0.0.1",
            port="3306",
            user="root",
            password="root",
            database="Haematology"
        )

        iblis_cursor = iblis_connection.cursor(dictionary=True)
        srs_cursor = srs_connection.cursor(dictionary=True)

        # iblis_query to fetch the required data
        iblis_query = """
        WITH RankedTests AS (
            SELECT 
                specimens.accession_number AS accession_id,
                tests.test_type_id AS test_type,
                tests.test_status_id AS test_status,
                ROW_NUMBER() OVER (
                    PARTITION BY specimens.accession_number, tests.test_type_id 
                    ORDER BY tests.time_created DESC
                ) AS rn
            FROM 
                specimens
            INNER JOIN 
                tests ON specimens.id = tests.specimen_id
            WHERE 
                specimens.specimen_type_id = %s
                AND tests.test_status_id NOT IN (1, 6, 7, 8)
                AND tests.time_created >= NOW() - INTERVAL %s DAY
                AND tests.test_type_id IN (%s, %s, %s, %s)
        )
        SELECT
            accession_id,
            test_type,
            test_status
        FROM
            RankedTests
        WHERE
            rn = 1;
        """

        # Execute the query
        iblis_cursor.execute(iblis_query, (department_id, intvl, test_type_id1, test_type_id2, test_type_id3, test_type_id4))
        iblis_results = iblis_cursor.fetchall()

        # Insert the results into srsDB if they don't already exist
        for result in iblis_results:
            accession_id = result['accession_id']
            test_type = result['test_type']
            test_status = result['test_status']

            # Check if accession_id with the same test_type already exists in the srsDB tests table
            srs_cursor.execute("SELECT test_status FROM tests WHERE accession_id = %s AND test_type = %s", (accession_id, test_type))
            existing_record = srs_cursor.fetchone()

            if not existing_record:
                # Insert the record into srsDB
                srs_insert_query = """
                INSERT INTO tests (accession_id, test_type, test_status)
                VALUES (%s, %s, %s)
                """
                srs_cursor.execute(srs_insert_query, (accession_id, test_type, test_status))
                srs_connection.commit()
                print(f"Inserted new record for accession_id: {accession_id}, test_type: {test_type}, test_status: {test_status}")

            elif existing_record['test_status'] == test_status:
                print(f"No update needed for accession_id: {accession_id}, test_type: {test_type}, test_status: {test_status}")
                continue

            elif existing_record['test_status'] != test_status:
                # Update the status if it is different
                srs_update_query = """
                UPDATE tests
                SET test_status = %s
                WHERE accession_id = %s AND test_type = %s
                """
                srs_cursor.execute(srs_update_query, (test_status, accession_id, test_type))
                srs_connection.commit()
                print(f"Updated record for accession_id: {accession_id}, test_type: {test_type}, new test_status: {test_status}, old test_status: {existing_record['test_status']}")

    except mysql.connector.Error as err:
        print(f"Error: {err}")

    finally:
        # Close all connections and cursors
        if 'iblis_cursor' in locals():
            iblis_cursor.close()
        if 'iblis_connection' in locals():
            iblis_connection.close()
        if 'srs_cursor' in locals():
            srs_cursor.close()
        if 'srs_connection' in locals():
            srs_connection.close()

updateEntries()

核心问题是数据类型不匹配：

• iBlissDB中test_status_id为整数类型，查询返回的test_status是Python整数
• srsDB中test_status字段可能是字符串类型（如CHAR/VARCHAR），查询返回的existing_record['test_status']是Python字符串
• 整数与字符串直接用==比较时结果为False，导致代码错误进入更新分支，执行不必要的UPDATE操作

方案1：Python内统一类型后比较 #

修改循环内的状态判断逻辑，将查询到的现有状态转换为整数后再比较：

# ... 原有代码省略 ...
for result in iblis_results:
    accession_id = result['accession_id']
    test_type = result['test_type']
    test_status = result['test_status']

    # Check if accession_id with the same test_type already exists in the srsDB tests table
    srs_cursor.execute("SELECT test_status FROM tests WHERE accession_id = %s AND test_type = %s", (accession_id, test_type))
    existing_record = srs_cursor.fetchone()

    if not existing_record:
        # 插入逻辑保持不变
        srs_insert_query = """
        INSERT INTO tests (accession_id, test_type, test_status)
        VALUES (%s, %s, %s)
        """
        srs_cursor.execute(srs_insert_query, (accession_id, test_type, test_status))
        srs_connection.commit()
        print(f"Inserted new record for accession_id: {accession_id}, test_type: {test_type}, test_status: {test_status}")
    else:
        # 统一转换为整数类型后比较
        existing_test_status = int(existing_record['test_status'])
        if existing_test_status == test_status:
            print(f"No update needed for accession_id: {accession_id}, test_type: {test_type}, test_status: {test_status}")
            continue
        else:
            # Update the status if it is different
            srs_update_query = """
            UPDATE tests
            SET test_status = %s
            WHERE accession_id = %s AND test_type = %s
            """
            srs_cursor.execute(srs_update_query, (test_status, accession_id, test_type))
            srs_connection.commit()
            print(f"Updated record for accession_id: {accession_id}, test_type: {test_type}, new test_status: {test_status}, old test_status: {existing_test_status}")
# ... 后续代码省略 ...

方案2：SQL查询时直接转换类型 #

修改srsDB的查询语句，将test_status转换为整数后返回，避免Python内手动转换：

# 将原有查询语句替换为：
srs_cursor.execute("SELECT CAST(test_status AS UNSIGNED) AS test_status FROM tests WHERE accession_id = %s AND test_type = %s", (accession_id, test_type))

方案3：根源统一数据库字段类型 #

检查并修改srsDB中tests表的test_status字段类型，使其与iBlissDB中的test_status_id类型一致（如均设为INT），彻底避免类型不匹配问题。

内容的提问来源于stack exchange，提问作者89Tain