我为自己编写了一个示例 python 脚本，以了解如何使用
pymysql示例脚本在数据库中创建一个表，添加随机数据，更新数据，然后删除数据，然后删除表。
我的第一个错误是第 43 行，我正在尝试进行一些错误检查。这是第 43 行：
print("pre_load_data: Connection error pymysql %d: %s" %(e.args[0], e.args[1]))
这给出了错误
Argument 'builtins.str' does not match format type 'd'
另一个错误是第 61 行，它产生了 2 个错误。这是第 61 行：
with connection:
'Connection' has no attribute '__enter__'
'Connection' has no attribute '__exit__'
然后在第 106 行，我收到许多不同类型的错误，这些错误只是用数据库中的字段分配变量。
first_name_variable = result['FIRST_NAME']
No overload variant of '__getitem__ of 'tuple' matches argument type 'str'
Possible overload variants
def __getitem__(self, int) -> Any
def __getitem__(self, slice) -> Tuple[Any,...]
Value of type 'Union[Tuple[Any,...],Dict[str, Ant], None]' is not indexable
感谢您阅读任何帮助，我们将不胜感激。
这是我的完整脚本...
#!/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 %d: %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 %d: %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 %d: %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 %d: %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 %d: %s" %(e.args[0], e.args[1]))
finally:
print("# SELECT record direct values using cursor.fetchone(): clean up operations")
first_name_variable = result['FIRST_NAME']
print("Extract data from result into variable: ", first_name_variable)
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 %d: %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 %d: %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 %d: %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 %d: %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 %d: %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 %d: %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 %d: %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 %d: %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 %d: %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 %d: %s" %(e.args[0], e.args[1]))
finally:
print("# DROP table: clean up operations")
end = time.time()
print("ending @ ",end - start)
任何人都可以解释这些错误以及如何解决它们吗？
对于最后一个错误，可以添加类型注释，表示游标只返回字典。用途：
cursor: pymysql.cursors.DictCursor = connection.cursor()
我认为 Thonny 无法自动从
cursorclass=pymysql.cursors.DictCursor