Python MySQL SQLAlchemy 2.0+: Connection, Queries, CRUD & Error Handling

Interactive Google Colab Practice Notebook

Practice SQLAlchemy 2.0+ connection engines, context managers, text() constructs, CRUD operations, and error handling directly inside Google Colab.

View on GitHub

We will use SQLAlchemy (using the recommended mysqlclient or pymysql driver for Python 3) to manage our MySQL database. SQLAlchemy provides a powerful, high-level Object Relational Mapper and Core SQL toolkit for robust database management.

Connecting to MySQL Using SQLAlchemy 2.0+

To establish a connection engine, we use create_engine() with a database connection string. In modern SQLAlchemy 2.0+, queries are executed using Python context managers (`with` blocks) to automatically manage connection lifecycles and transactions.

from sqlalchemy import create_engine, text

# Create connection engine
engine = create_engine("mysql+pymysql://userid:password@localhost/my_database?charset=utf8mb4")

# Execute a query using a connection context manager
with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM student LIMIT 0, 5"))
    for row in result:
        print(row)

Connection string parameters:

mysql+pymysql / mysql+mysqldb: The dialect and DBAPI driver.
userid: Your MySQL login username.
password: Your MySQL login password.
localhost: Database server address or IP.
my_database: Target database name.
?charset=utf8mb4: Character set configuration for full Unicode support.

Fetching Records: fetchall(), fetchmany(), fetchone(), and first()

Once connected, you can collect records from result objects using various cursor-like retrieval methods:

with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM student LIMIT 0, 5"))
    
    # fetchall(): Returns all rows as a list
    rows = result.fetchall()
    for row in rows:
        print(row)

Parameterized Queries with WHERE Conditions

To prevent SQL injection attacks, always pass parameters as separate dictionary or tuple arguments to execute():

with engine.connect() as conn:
    query = "SELECT * FROM student WHERE class=:class_name AND mark > :min_mark"
    result = conn.execute(text(query), {"class_name": "Four", "min_mark": 50})
    
    print("Rows collected =", result.rowcount)
    for row in result:
        print(row)

Inserting, Updating, and Deleting Records (Transactions)

For data-modifying queries (INSERT, UPDATE, DELETE), use engine.begin(). This automatically manages the transaction, committing changes upon successful completion or rolling back if an exception occurs.

# Inserting a record
with engine.begin() as conn:
    query = "INSERT INTO student (name, class, mark, gender) VALUES (:name, :class, :mark, :gender)"
    result = conn.execute(text(query), {"name": "King", "class": "Five", 45, "gender": "male"})
    print("Rows Added =", result.rowcount)
    print("Last Insert ID =", result.inserted_primary_key)

Error Handling with SQLAlchemyError

Catch database connection and execution exceptions gracefully using SQLAlchemyError inside try-except blocks:

from sqlalchemy.exc import SQLAlchemyError

try:
    with engine.connect() as conn:
        result = conn.execute(text("SELECT * FROM non_existent_table"))
except SQLAlchemyError as e:
    print("Database Error:", str(e.__dict__.get('orig', e)))

Frequently Asked Questions & Common Errors

In SQLAlchemy 2.0+, executing queries directly on the Engine object is deprecated. You must open a connection using with engine.connect() as conn: or with engine.begin() as conn: and execute queries on the connection object.

SQLAlchemy 2.0+ requires raw SQL strings to be explicitly wrapped in the text() construct before execution (e.g., conn.execute(text("SELECT * FROM student"))).

Use engine.connect() for read-only queries (SELECT). Use engine.begin() for write operations (INSERT, UPDATE, DELETE) because it automatically commits transactions upon block exit and rolls back if errors occur.

You can inspect the inserted_primary_key attribute on the result object returned by conn.execute() when inserting into a table with an auto-incrementing primary key.

Related Tutorials & Next Steps

Python MySQL Insert Record Guide
Learn how to insert single and multiple records using MySQL Connector.
Python MySQL Connection Guide
Explore database connection fundamentals and connection pooling.
Python MySQL Error Handling
Master try-except exception blocks and error code trapping.
Student Table SQL Dump File
Download the complete student.sql database dump file featuring sample records with the gender column schema.

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

Subhendu Mohapatra

Author & Instructor at plus2net

I write and maintain practical tutorials on Python, PHP, SQL, JavaScript, HTML, jQuery, and web development at plus2net. The tutorials focus on clear explanations, working examples, and code that readers can test and adapt while learning.

About the author Watch on YouTube

Connect: