使用pymysql操作MySQL的三类错误解析与修复请求

错误1：格式化字符串类型不匹配（第43行） #

原因 #

pymysql.Error的args[0]并非始终为整数类型，部分错误场景下错误码会以字符串形式返回。原代码用%d（整数格式化符）格式化字符串类型的错误码，导致类型不匹配报错。

解决方法 #

将格式化符%d改为%s，统一处理整数和字符串类型的错误码：

print("pre_load_data: Connection error pymysql %s: %s" %(e.args[0], e.args[1]))

错误2：Connection对象不支持上下文管理器（第61行） #

原因 #

旧版本pymysql的Connection类未实现__enter__和__exit__方法，无法直接用于with语句（上下文管理器）。

解决方法 #

有两种可行方案：

1. 升级pymysql版本：升级至>=0.7.11版本，该版本开始支持Connection作为上下文管理器使用。
2. 移除外层with connection:语句：内层已通过with connection.cursor() as cursor:管理游标，可直接删除外层with，并在脚本末尾添加connection.close()确保连接关闭。

修改后的代码片段：

# 移除原有的with connection:语句，直接使用游标上下文
with connection.cursor() as cursor:
    # 后续操作保持不变

错误3：Tuple类型无法用字符串索引（第106行） #

原因 #

Thonny的类型检查识别到result的类型为Union[Tuple[Any,...],Dict[str, Any], None]，即可能是元组、字典或None。若查询失败或返回非字典结果，使用字符串索引会触发错误；同时finally块无论try是否成功都会执行，若查询异常，result可能未被正确赋值为字典。

解决方法 #

在finally块中先判断result的有效性，确保是字典类型后再进行索引操作：

finally:
    print("# SELECT record direct values using cursor.fetchone(): clean up operations")
    # 先判断result存在且为字典类型
    if result and isinstance(result, dict):
        first_name_variable = result.get('FIRST_NAME')
        print("Extract data from result into variable: ", first_name_variable)
    else:
        print("No valid result to extract data from")

修改后的完整脚本 #

#!/usr/bin/python3
# Python Database Master File with Error Checking
import pymysql
# Just used for timing and random stuff
import sys
import time
import random
import string

def get_random_string(length):
    # choose from all lowercase letter
    letters = string.ascii_lowercase
    result_str = ''.join(random.choice(letters) for i in range(length))
    print("Random string of length", length, "is:", result_str)
    return result_str

start = time.time()
print("starting @ ",start)
runallcode = 0 # Change to 1 to delete records, trucate and drop
sqlinsert = []
FirstName = get_random_string(4)
for y in range (1,3):
    presqlinsert = (
        FirstName,
        get_random_string(10),
        y,
        "M",
        random.randint(0,50)
        )
    sqlinsert.append(presqlinsert)
print("sqlinsert @ ",sqlinsert)

# Open database connection
try:
    # Connect to the database
    connection = pymysql.connect(host='192.168.0.2',
        user='weby',
        password='password',
        database='datadb',
        charset='utf8mb4',
        cursorclass=pymysql.cursors.DictCursor)
except pymysql.Error as e:
    print("pre_load_data: Connection error pymysql %s: %s" %(e.args[0], e.args[1]))
    sys.exit(1)

# Simple SQL just to get Version Number
# prepare a cursor object using cursor() method
cursor = connection.cursor()
# execute SQL query using execute() method.
cursor.execute("SELECT VERSION()")
# Fetch a single row using fetchone() method.
data = cursor.fetchone()
print ("Database version : %s " % data)

# Check if EMPLOYEE table exists, if so drop it
# prepare a cursor object using cursor() method
cursor = connection.cursor()
# Drop table if it already exist using execute() method.
cursor.execute("DROP TABLE IF EXISTS EMPLOYEE")

# 移除外层with connection:语句
with connection.cursor() as cursor:
    try:
        # Create a new table called EMPLOYEE
        sql = "CREATE TABLE EMPLOYEE (id INT NOT NULL AUTO_INCREMENT,FIRST_NAME CHAR(20) NOT NULL,LAST_NAME CHAR(20),AGE INT,SEX CHAR(1),INCOME FLOAT,PRIMARY KEY (id))"
        cursor.execute(sql)
        # connection is not autocommit by default. So you must commit to save your changes.
        connection.commit()
    except pymysql.Error as e:
        print("# Create a new table: error pymysql %s: %s" %(e.args[0], e.args[1]))
    finally:
        print("# Create a new table: clean up operations")

    try:
        # ALTER TABLE to add index
        sql = "ALTER TABLE EMPLOYEE ADD INDEX (id);"
        cursor.execute(sql)
        # connection is not autocommit by default. So you must commit to save your changes.
        connection.commit()
    except pymysql.Error as e:
        print("# ALTER TABLE to add index: error pymysql %s: %s" %(e.args[0], e.args[1]))
    finally:
        print("# ALTER TABLE to add index: clean up operations")

    try:
        # Create a new record direct values
        sql = "INSERT INTO EMPLOYEE(FIRST_NAME, LAST_NAME, AGE, SEX, INCOME)  VALUES ('Thirsty','Lasty',69,'M',999)"
        cursor.execute(sql)
        # connection is not autocommit by default. So you must commit to save your changes.
        connection.commit()
    except pymysql.Error as e:
        print("# Create a new record direct values: error pymysql %s: %s" %(e.args[0], e.args[1]))
    finally:
        print("# Create a new record direct values: clean up operations")

    try:
        # SELECT record direct values using cursor.fetchone()
        sql = "SELECT * FROM EMPLOYEE WHERE id = 1"
        cursor.execute(sql)
        result = cursor.fetchone()
        print(result)
    except pymysql.Error as e:
        print("# SELECT record direct values using cursor.fetchone(): error pymysql %s: %s" %(e.args[0], e.args[1]))
    finally:
        print("# SELECT record direct values using cursor.fetchone(): clean up operations")
        # 增加类型判断
        if result and isinstance(result, dict):
            first_name_variable = result.get('FIRST_NAME')
            print("Extract data from result into variable: ", first_name_variable)
        else:
            print("No valid result to extract data from")

    try:
        # Create a new record using variables
        sql = "INSERT INTO EMPLOYEE(FIRST_NAME, LAST_NAME, AGE, SEX, INCOME)  VALUES (%s, %s, %s, %s, %s)"
        cursor.execute(sql, ('Qweyac', 'PterMohan', 40, 'M', 5450))
        # connection is not autocommit by default. So you must commit to save your changes.
        connection.commit()
    except pymysql.Error as e:
        print("# Create a new record using variables: error pymysql %s: %s" %(e.args[0], e.args[1]))
    finally:
        print("# Create a new record using variables: clean up operations")

    try:
        # SELECT record direct values using cursor.fetchall()
        sql = "SELECT * FROM EMPLOYEE"
        cursor.execute(sql)
        result = cursor.fetchall()
        print(result)
    except pymysql.Error as e:
        print("# SELECT record direct values using cursor.fetchall(): error pymysql %s: %s" %(e.args[0], e.args[1]))
    finally:
        print("# SELECT record direct values using cursor.fetchall(): clean up operations")

    try:
        # Create a new record using array and cursor.executemany
        sql = "INSERT INTO EMPLOYEE(FIRST_NAME, LAST_NAME, AGE, SEX, INCOME)  VALUES (%s, %s, %s, %s, %s)"
        cursor.executemany(sql,sqlinsert)
        # connection is not autocommit by default. So you must commit to save your changes.
        connection.commit()
    except pymysql.Error as e:
        print("# Create a new record using array and cursor.executemany: error pymysql %s: %s" %(e.args[0], e.args[1]))
    finally:
        print("# Create a new record using array and cursor.executemany: clean up operations")
        sqlinsert.clear()

    try:
        # SELECT record direct values using cursor.fetchall()
        sql = "SELECT * FROM EMPLOYEE"
        cursor.execute(sql)
        result = cursor.fetchall()
        print(result)
    except pymysql.Error as e:
        print("# SELECT record direct values using cursor.fetchall(): error pymysql %s: %s" %(e.args[0], e.args[1]))
    finally:
        print("# SELECT record direct values using cursor.fetchall(): clean up operations")


    try:
        # UPDATE record using variables
        sql = "UPDATE EMPLOYEE SET FIRST_NAME = %s, LAST_NAME = %s WHERE id = 2"
        cursor.execute(sql, ('Peter', 'Brown'))
        # connection is not autocommit by default. So you must commit to save your changes.
        connection.commit()
    except pymysql.Error as e:
        print("# UPDATE record using variables: error pymysql %s: %s" %(e.args[0], e.args[1]))
    finally:
        print("# UPDATE record using variables: clean up operations")


    try:
        # SELECT record direct values using cursor.fetchone()
        sql = "SELECT * FROM EMPLOYEE WHERE id = 2"
        cursor.execute(sql)
        result = cursor.fetchone()
        print(result)
    except pymysql.Error as e:
        print("# SELECT record direct values using cursor.fetchone(): error pymysql %s: %s" %(e.args[0], e.args[1]))
    finally:
        print("# SELECT record direct values using cursor.fetchone(): clean up operations")

    try:
        # SELECT record direct values using cursor.fetchall()
        sql = "SELECT * FROM EMPLOYEE"
        cursor.execute(sql)
        result = cursor.fetchall()
        print(result)
    except pymysql.Error as e:
        print("# SELECT record direct values using cursor.fetchall(): error pymysql %s: %s" %(e.args[0], e.args[1]))
    finally:
        print("# SELECT record direct values using cursor.fetchall(): clean up operations")
        print("Total number of rows in table: ", cursor.rowcount)
        print("\nPrinting each row")
        for row in result:
            for key,value in row.items():
                print('The key is %s'%key,'The value is %s'%value)

    if runallcode == 1:
        try:
            # DELETE FROM table called
            sql = "DELETE FROM EMPLOYEE WHERE id = %s"
            deletesql = 1
            cursor.execute(sql,deletesql)
            # connection is not autocommit by default. So you must commit to save your changes.
            connection.commit()
        except pymysql.Error as e:
            print("# Delete From table: error pymysql %s: %s" %(e.args[0], e.args[1]))
        finally:
            print("# Delete  From table: clean up operations")

        try:
            # Trucate table called EMPLOYEE
            sql = "TRUNCATE TABLE EMPLOYEE"
            cursor.execute(sql)
            # connection is not autocommit by default. So you must commit to save your changes.
            connection.commit()
        except pymysql.Error as e:
            print("# Truncate table: error pymysql %s: %s" %(e.args[0], e.args[1]))
        finally:
            print("# Truncate table: clean up operations")

        try:
            # DROP table called EMPLOYEE
            sql = "DROP TABLE EMPLOYEE"
            cursor.execute(sql)
            # connection is not autocommit by default. So you must commit to save your changes.
            connection.commit()
        except pymysql.Error as e:
            print("# DROP table: error pymysql %s: %s" %(e.args[0], e.args[1]))
        finally:
            print("# DROP table: clean up operations")

# 手动关闭连接
connection.close()
end = time.time()
print("ending @ ",end - start)

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