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:password@localhost/my_database")
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.
Tutorial on MySQL SELECT Query
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)
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()?
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)
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.
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
The variable my_conn is our connection string, same is created before ( see above ) . We used try except error handling here.
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.