0

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?

2
  • Your print() line is correct according to this answer. So I think there's a problem with the type annotation for pymysql.Error.args. Commented Aug 28, 2023 at 20:13
  • Just use %s instead of %d – or better yet, f-strings. Commented Aug 28, 2023 at 20:33

1 Answer 1

1

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).

Sign up to request clarification or add additional context in comments.

Comments

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.