I have put together an example python scripts for myself to understand how to connect to a MySQL database using pymysql. It works but in Thonny-Assistant it is showing a few errors and I am asking for help to understand these errors to see if they are important.
The example script creates a table in the database, adds random data, updates the data, and then deletes the data and then the table.
My first error is line 43 where I am attempting some error checking. This is line 43:
print("pre_load_data: Connection error pymysql %d: %s" %(e.args[0], e.args[1]))
which gives the error
Argument 'builtins.str' does not match format type 'd'
Then another error is line 61 which produces 2 errors. This is line 61:
with connection:
'Connection' has no attribute '__enter__'
'Connection' has no attribute '__exit__'
Then on line 106, I get many different types of errors which are just assigning a variable with a field from the database.
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
Thank you for reading any any help would be grateful to understand.
Here is my full script...
#!/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)
Can anyone explain the errors and how can I solve them?
For the last error, you can add a type annotation indicating that the cursor only returns dictionaries. Use:
cursor: pymysql.cursors.DictCursor = connection.cursor()
I don't think Thonny is able to infer this automatically from cursorclass=pymysql.cursors.DictCursor
Regarding the error on the with statement, it looks like context manager support was removed from pymysql connections. See Can I use pymysql.connect() with "with" statement?. So just get rid of that with statement, and put connection.close() at the end (or just exit the script, it will close automatically).