mysql database managing using mysqlclient by sqlalchemy in python

We will use mysqlclient ( recomended for Python 3 ) here to manage mysql database. Read here how to install and connect to MySQL database using sqlalchemy.

Python to MySQL database connection by SQLAlchemy with error handling using try except blocks

Here is the connection string again , after successful connection we will use the variable my_conn in our examples below.

from sqlalchemy import create_engine
my_conn = create_engine("mysql+mysqldb://userid:pw@localhost/my_db")

usrid : Your MySql login user id
pw : Your MySql login password
localhost : Address of MySQL server, IP address or localhost
my_db : Database name of MySQL where our table is available.

Once above code works without any error ( use your own userid , password and database name ) we will go for collecting records.
In our MySQL database there is a student table with sample data. You can use the link at the end of this tutorial to get SQL dump to create your own table.

Based on the data you are handling you can add the type of charset to your connection.

engine = create_engine("mysql+mysqldb://userid:pw@localhost/db?charset=utf8mb4")

MySQL SELECT Query in Python using fetchall(), fetchmany(),fetchone(),first() methods by SQLAlchemy

fetchmany() : Returning 5 rows of data

query="SELECT * FROM student LIMIT 0,5"
                
my_data=my_conn.execute(query) # SQLAlchemy my_conn result
#my_data=my_data.fetchmany(size=2) # collect 2 rows of data
for row in my_data:
    print(row)

We can reate a list from the result set

my_data=my_conn.execute(query) # SQLAlchemy my_conn result
my_list = [r for r, in my_data] # create a  list

What happens if our SQL query ask for 10 records by using LIMIT and we try to print 15 records by using fetchmany() ?

fetchone()

fetchone() will return one row of data and move the pointer to next row.

query="SELECT * FROM student LIMIT 0,5"
                
my_data=my_conn.execute(query) 
my_row=my_data.fetchone()
#my_data=my_data.fetchmany(size=2)
print(my_row[0] , my_row[1]) # 1 John Deo
#my_data.next()
my_row=my_data.fetchone()
print(my_row[0],my_row[1]) # 2 Max Ruin

fetchall()

Fetch all rows (Returns a list of data with each row as element )

from sqlalchemy.exc import SQLAlchemyError
q="SELECT * FROM student LIMIT 0,10"
try:
  my_cursor=my_conn.execute(q)
  my_data=my_cursor.fetchall()
  for row in my_data:
    print(row)
except SQLAlchemyError as e:
  error=str(e.__dict__['orig'])
  print(error)
except Exception as e:
    print(e)
else:
  print("Total Number of rows : ",my_cursor.rowcount)

first()

Collect the first row of data and then closes the result unconditionally.

from sqlalchemy.exc import SQLAlchemyError
q="SELECT * FROM student LIMIT 0,10"
try:
  my_cursor=my_conn.execute(q)
  my_data=my_cursor.first()
  print(type(my_data))
  print(my_data['name'])
  for row in my_data:
    print(row)
except SQLAlchemyError as e:
  error=str(e.__dict__['orig'])
  print(error)
except Exception as e:
    print(e)
else:
  print("Total Number of rows : ",my_cursor.rowcount)

What happens if we try to use the cursor again after using first()?

  my_data=my_cursor.first()
  print(my_data['name'])
  my_data=my_cursor.fetchone()

This will generate error

This result object is closed.

WHERE Condition

MySQL SELECT Query in Python with WHERE to get matching records by user input & parameterized query

from sqlalchemy import create_engine
my_conn = create_engine("mysql+mysqldb://userid:password@localhost/my_database")


rs=my_conn.execute("SELECT * FROM  student WHERE class='Three'")
my_data= rs.fetchall() # a list 
print(my_data)

Output is here

[(2, 'Max Ruin', 'Three', 85, 'male'), 
(3, 'Arnold', 'Three', 55, 'male'), (27, 'Big Nose', 'Three', 81, 'female')]

We used here SQL WHERE to collect records of class = 'Three' , similarly we can use different SQL queries to get data from MySQL table.

Using parameters in queries to get data from table

We will pass parameters to our query to get matching data from table.

sql="SELECT * FROM student WHERE class=%s and id > %s"
rs=my_conn.execute(sql, 'Four',20)
print("Rows collected  = ",rs.rowcount)
print(rs.fetchall())

We used rowcount() to know number of records returned by MySQL database table
Output is here

Rows collected  =  2
[(21, 'Babby John', 'Four', 69, 'female'), (31, 'Marry Toeey', 'Four', 88, 'male')]

Adding records

We used SQL insert command

rs=my_conn.execute("INSERT INTO  student (`id` ,`name` ,`class` ,`mark` ,`sex`) \
                  VALUES ('36',  'King',  'Five',  '45',  'male')")
print("Rows Added  = ",rs.rowcount)

Output is here

Rows Added  =  1

More on inserting records to MySQL table

Updating records

We used update command

rs=my_conn.execute("UPDATE student SET class='Four' Where class='Five'")
print("Rows updated = ",rs.rowcount)

Output

Rows updated =  11

Deleting records

We used Delete command

from sqlalchemy import create_engine
my_conn = create_engine("mysql+mysqldb://root:test@localhost/my_tutorial")

sql="DELETE  FROM student WHERE class=%s and id < %s"
rs=my_conn.execute(sql, 'Four',20)
print("Rows Deleted  = ",rs.rowcount)

Output

Rows Deleted  =  2

lastrowid

We will keep one auto increment id field ( change the present ID field ) in our student table. Once we add one record , MySQL will add next incremental unique id to the ID field. This ID which is generated while adding the record can be collected by using lastrowid.

more on Auto Incremented Field.

rs=my_conn.execute("INSERT INTO  student (`name` ,`class` ,`mark` ,`sex`) \
                  VALUES ( 'King2',  'Five',  '45',  'male')")
print("Last insert ID  = ",rs.lastrowid)
print("Rows Added  = ",rs.rowcount)

Output is here

Last insert ID  =  40
Rows Added  =  1

Using parameters

query="INSERT INTO  `my_tutorial`.`student` (`name` ,`class` ,`mark` ,`sex`) \
            VALUES(%s,%s,%s,%s)"
my_data=('King','Five',45,'male')
rs=my_conn.execute(query,my_data)
print("Last insert ID  = ",rs.lastrowid)
print("Rows Added  = ",rs.rowcount)

Output

Last insert ID  =  42
Rows Added  =  1

You must use Auto increment id field to get the value using lastrowid. Use the correct table structure given at student sql dump page to create your table with Auto Increment ID

SQLAlchemy With Error handling

Try except code block to catch SQLALchemy MySQL connection error if login details are wrong

The variable my_conn is our connection string, same is created before ( see above ) . We used try except error handling here.
Checking only the connection string using SQLAlchemyError

from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy import create_engine
my_conn = create_engine("mysql+mysqldb://userid:pw@localhost/my_db")
try:
   my_conn.connect() 
   print("Successfully connected")
except SQLAlchemyError as e:
  error=str(e.__dict__['orig'])
  print(error)

Using at execute() stage

from sqlalchemy.exc import SQLAlchemyError
q="SELECT * FROM student LIMIT 0,10"
try:
  rs=my_conn.execute(q)
except SQLAlchemyError as e:
  error=str(e.__dict__['orig'])
  print(error)
else:
  print("Total Number of rows : ",rs.rowcount)

Exercise
In above code you have seen how to use SQLAlchemy library and manage MySQL database.
Here is a list of Queries you can execute using the above learning and display the outcome.

List of Queries

Python functions for MySQL Collect records Python- Tutorials

Subscribe to our YouTube Channel here

Python sqlalchemy to manage MySQL database

fetchmany() : Returning 5 rows of data

fetchone()

fetchall()

first()

WHERE Condition

Using parameters in queries to get data from table

Adding records

Updating records

Deleting records

lastrowid

SQLAlchemy With Error handling

Subscribe