Mastering Data Updates with SQLAlchemy Core in 2025

SQLAlchemy is the go-to database toolkit for Python developers looking to work with relational databases in their applications. It provides a set of tools for database access using Python, with the goal of being efficient and easy to use. SQLAlchemy has two main components:

1. Core – The SQL Expression Language, for writing SQL queries and statements in Python code
2. ORM (Object Relational Mapper) – Allows you to map Python classes to database tables to work in terms of objects

In this article, we‘ll focus on SQLAlchemy Core and how you can leverage it to perform data manipulation, especially updates, against your database. We‘ll be using the latest SQLAlchemy 1.4/2.0 release to highlight new features and best practices.

What‘s New in SQLAlchemy Core 1.4/2.0

Released in April 2021, SQLAlchemy 1.4 brought a number of enhancements and new features to both Core and ORM. Some key updates pertinent to Core include:

• Moving the engine and connection into one Engine class for a unified interface
• Making use of the Connection object explicit via Engine.connect()
• A new expire_on_commit parameter to control when connections are returned to the connection pool
• Improved support for Python asyncio through the asyncio and greenlet extensions

Under the hood, SQLAlchemy 2.0 introduces foundational changes to the internals, paving the way for async database support and better future-proofing of the API. While the external API remains largely backward compatible, it‘s recommended to consult the migration guide when upgrading from 1.3 or earlier.

Connecting to a Database

Before you can execute any SQL with SQLAlchemy Core, you need to establish a connection to your target database. This is done by creating an Engine object using the create_engine() function:

from sqlalchemy import create_engine
engine = create_engine("sqlite+pysqlite:///:memory:", echo=True, future=True)

The first argument is the database URL which specifies the database driver (e.g. sqlite, postgresql, mysql) and connection details like host, port, username and database name. Here we‘re using an in-memory SQLite database for demonstration purposes.

The echo parameter enables output logging to help with debugging. The future flag opts into the 2.0 behaviors for forward compatibility.

Defining Tables and Metadata

To start working with tables, we define them via the Table construct, which is tied to a MetaData collection object:

from sqlalchemy import MetaData, Table, Column, Integer, String
metadata = MetaData()
user_table = Table(
"user_account", 
metadata,
Column(‘id‘, Integer, primary_key=True),
Column(‘username‘, String(50), nullable=False),
Column(‘fullname‘, String(100)),
Column(‘email‘, String)
)

Each Column in the table has a name, a data type, and optional specifiers like primary key, foreign key or nullable constraints.

To create the corresponding table in the database, we use metadata.create_all():

with engine.begin() as conn:
    metadata.create_all(conn)

Inserting Rows

With a table defined, we can insert data using an insert() construct:

from sqlalchemy import insert
insert_stmt = insert(user_table).values(
username="johndoe",
fullname="John Doe",
email="[email protected]"
)
with engine.begin() as conn:
conn.execute(insert_stmt)

You can also pass a list of dictionaries to insert multiple rows at once:

with engine.begin() as conn:
    conn.execute(
        user_table.insert(),
        [
            {"username": "janedoe", "fullname": "Jane Doe", "email": "[email protected]"},
            {"username": "smithj", "fullname": "John Smith", "email": "[email protected]"},
        ],
    )

Updating Data

An update statement comprises a target table, new values to set, and an optional WHERE clause that filters the rows to update. It is constructed using update():

from sqlalchemy import update
update_stmt = (
update(user_table)
.where(user_table.c.username == "johndoe")
.values(fullname="John E. Doe")
)
with engine.begin() as conn:
conn.execute(update_stmt)

Advanced UPDATE Techniques

Beyond basic updates, SQLAlchemy Core provides features to tune your update statements for more complex logic or performance:

Correlated Updates

A correlated update issues an UPDATE using rows from another table. This utilizes a correlated subquery as a scalar value:

from sqlalchemy import select
scalar_subq = (
select(address_table.c.email_address)
.where(address_table.c.user_id == user_table.c.id)
.order_by(address_table.c.id)
.limit(1)
.scalar_subquery()
)
update_stmt = update(user_table).values(email=scalar_subq)

UPDATE..FROM

If additional tables are present in the WHERE clause of an UPDATE, most databases require the use of UPDATE..FROM syntax:

update_stmt = (
    update(user_table)  
    .where(user_table.c.id == address_table.c.user_id)
    .where(address_table.c.email_address.startswith("steve"))
    .values(fullname="Steve Austin")
    .execution_options(synchronize_session="fetch")  
)

Returning Updated Rows

Some backends support RETURNING clauses to retrieve the updated state of rows from the database after an UPDATE executes. This can avoid the need for a subsequent SELECT:

update_stmt = (
    update(user_table)
    .where(user_table.c.username == "janedoe")
    .values(email="[email protected]")
    .returning(user_table.c.id, user_table.c.email)
)
with engine.begin() as conn:
for id, email in conn.execute(update_stmt):
print(f"Updated user {id}, new email: {email}")

Bulk Updates and Performance

When updating a large number of rows, there are several ways to optimize for performance, depending on your particular database and use case:

• Use executemany() to batch multiple parameter sets into a single statement execution
• Tune the batch size to balance between round trips and excessive memory usage
• Disable autocommit mode on the connection for the duration of the bulk operation
• Consider using alternative APIs provided by the database driver, such as psycopg2.extras.execute_batch() for PostgreSQL
• Drop or disable indexes (especially unique indexes) prior to the bulk update, then rebuild them after
• Use IN expressions or temporary tables for the UPDATE criteria if the number of rows to update is large

Here‘s an example of a bulk update using executemany() to pass multiple parameter sets:

with engine.begin() as conn:
    conn.execute(
        user_table.update().
        where(user_table.c.username == bindparam("username")).
        values(fullname=bindparam("fullname")),
        [
            {"username": "janedoe", "fullname": "Jane Z. Doe"},
            {"username": "smithj", "fullname": "John Q. Smith"},
        ],
    )

It‘s important to measure before and after any optimizations to ensure they are having the intended effect and not introducing any regressions.

Deleting Rows

To round out the data manipulation features, SQLAlchemy Core provides delete() constructs to remove rows from a table:

from sqlalchemy import delete
delete_stmt = (
delete(user_table)
.where(user_table.c.username == "johndoe")
)
with engine.begin() as conn:
conn.execute(delete_stmt)

Like updates, deletes can act on multiple tables by listing them in the WHERE clause. You can also take advantage of RETURNING, if supported, to get information about the deleted rows.

Conclusion

SQLAlchemy Core provides a robust yet easy to use toolkit for working with relational databases from Python. With the 1.4 and 2.0 releases, it continues to evolve to meet the demands of modern data-driven applications.

In this article, we focused on data manipulation using Core, with a deep dive on updates. Whether you need to insert, update or delete data, SQLAlchemy has you covered with a Pythonic API that integrates closely with the database. We also looked at some performance techniques for bulk updates.

I hope this gives you a solid foundation for mastering SQLAlchemy Core. The official tutorial and documentation are great resources to continue learning. Remember, when in doubt, test and measure to find what works best for your scenario. Happy coding!