PyMySQL执行INSERT无报错但未插入数据的技术问题排查

执行PyMySQL插入脚本时无报错，但数据并未写入数据库。在GCP Workbench中执行相同SQL语句可正常插入，调试发现cursor的rowcount为-1。尝试过调整参数格式、开关autocommit、修改datetime格式等方法均无效。此外通过Pycharm的cloud-sql-proxy删除行时会报错：

Unexpected update count received (Actual: 0, Expected: 1). All changes will be rolled back. com.intellij.database.console.JdbcEngineUtils$UnsafeUpdateRolledBackException: Unexpected update count received (Actual: 0, Expected: 1). All changes will be rolled back. in JdbcEngine$UpdateOperation$UnexpectedUpdatesGuard.reportChangesWillBeRolledBack(JdbcEngine.java:1927)

但在Workbench执行TRUNCATE TABLE creative_tester;可正常清空表。

原问题代码 #

插入脚本 #

def update_saturated_content(content_selection):
    conn = open_connection()
    conn.autocommit(True)
    with conn.cursor() as cursor:
        for content in content_selection:
            text = [content[0]["text"]]
            row = tuple([content[0]["list"], content[0]["link"], content[0]["promo"], content[1], float(0.2),
                         datetime.strptime(datetime.now().isoformat(), '%Y-%m-%dT%H:%M:%S.%f'), 1], )
            insert_query = ('''INSERT INTO creative_tester (''' +
                            '''list, ''' +
                            '''link, ''' +
                            '''promo, ''' +
                            '''text, ''' +
                            '''weighting, ''' +
                            '''date_added, ''' +
                            '''in_use) ''' +
                            '''VALUES (%s, %s, %s, %s, %s, %s, %s); ''')
            cursor.execute = (insert_query, row)
            conn.commit()
        cursor.close()
        conn.close()
        return

数据库连接函数 #

def open_connection():
    unix_socket = '/cloudsql/{}'.format(db_connection_name)
    try:
        if os.environ.get('FLASK_ENV') == 'development':
            conn = pymysql.connect(user=db_user, password=db_password, host="127.0.0.1", port=3306, db=db_name,
                                   cursorclass=pymysql.cursors.DictCursor)
        elif os.environ.get('GAE_ENV') == 'standard':
            conn = pymysql.connect(user=db_user, password=db_password, unix_socket=unix_socket, db=db_name,
                                   cursorclass=pymysql.cursors.DictCursor)
    except pymysql.MySQLError as e:
        print(e)

    return conn

数据表结构 #

CREATE TABLE `creative_tester` (
  `list` varchar(255) DEFAULT NULL,
  `link` varchar(255) DEFAULT NULL,
  `promo` varchar(255) DEFAULT NULL,
  `text` varchar(1027) DEFAULT NULL,
  `weighting` float DEFAULT NULL,
  `date_added` datetime DEFAULT NULL,
  `in_use` tinyint DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3

问题解决 #

核心错误定位 #

原代码中**cursor.execute = (insert_query, row)是赋值操作，而非调用execute方法**，导致SQL语句根本没有被执行。正确写法应为cursor.execute(insert_query, row)。

优化后的最终代码 #

def update_saturated_content(content_selection):
    conn = open_connection()
    conn.begin()
    conn.autocommit(True)
    creative_update = []
    with conn.cursor() as cursor:
        for content in content_selection:
            text = [content[0]["text"]]
            creative_update.append(tuple([content[0]["list"], content[0]["link"], content[0]["promo"], content[1], float(0.2),
                         datetime.now().isoformat(), 1], ))
            query = ('UPDATE creative_tester '
                     'SET in_use = 0 WHERE text LIKE (%s) '
                     'AND in_use = 1; ')
            cursor.execute(query, tuple(text, ))
        insert_query = ('INSERT INTO creative_tester ('
                        'list, '
                        'link, '
                        'promo, '
                        'text, '
                        'weighting, '
                        'date_added, '
                        'in_use) '
                        'VALUES (%s, %s, %s, %s, %s, %s, %s)')
        cursor.executemany(insert_query, creative_update)
        conn.commit()
        cursor.close()
        conn.close()
        return

优化说明 #

1. 使用executemany批量插入，替代循环内单次插入，提升执行效率
2. 简化datetime处理：直接用datetime.now().isoformat()生成符合要求的时间字符串，避免多余的strptime转换
3. 先执行UPDATE逻辑再批量插入，保证数据状态一致性

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