8.44. SQLAlchemy Query Execute

8.44.1. SetUp

✘

>>> from sqlalchemy import create_engine
>>> from sqlalchemy import Column, String, Integer
>>> from sqlalchemy.orm import sessionmaker, declarative_base
>>>
>>>
>>> DATABASE = 'sqlite:///:memory:'
>>>
>>> engine = create_engine(DATABASE)
>>> db = sessionmaker(bind=engine).__call__()
>>> Model = declarative_base()
>>>
>>>
>>> class User(Model):
...     __tablename__ = 'user'
...     uid = Column(Integer, autoincrement=True, primary_key=True)
...     firstname = Column(String, nullable=False)
...     lastname = Column(String, nullable=False)
>>>
>>>
>>> Model.metadata.create_all(engine)
>>>
>>> db.add_all([
...     User('Mark', 'Watney'),
...     User('Melissa', 'Lewis'),
...     User('Rick', 'Martinez'),
...     User('Alex', 'Vogel'),
...     User('Beth', 'Johanssen'),
...     User('Chris', 'Beck'),
... ])
>>> db.commit()

8.44.2. Raw SQL Queries

✘

>>> from sqlalchemy import text

✘

>>> query = text('SELECT * FROM users')
>>>
>>> with engine.connect() as db:
...     result = db.execute(query).all()

✘

>>> query = text("""SELECT *
...                 FROM users
...                 WHERE username=:username""")
>>>
>>> data = {'username': 'mwatney'}
>>>
>>> with engine.connect() as db:
...     result = db.execute(query).all()
...     print(result)

8.44.3. Transactions

Lets define a query with some data:

✘

>>> query = text("""INSERT INTO users (firstname, lastname)
...                 VALUES (:firstname, :lastname)""")
>>>
>>> data = {
...     'firstname': 'Pan',
...     'lastname': 'Twardowski',
... }

No autocommit at the library level. Always have to do it manually:

✘

>>> with engine.connect() as db:
...     result = db.execute(query, data)
...     db.commit()

In SQLAlchemy there is an option to use context manager engine.begin() which already has transaction setup and no need to .commit() at the end as of context manager will do it for you:

✘

>>> query = text("""INSERT INTO users (firstname, lastname)
...                 VALUES (:firstname, :lastname)""")
>>> data = {'firstname': 'Pan', 'lastname': 'Twardowski'}
>>>
>>> with engine.begin() as db:
...     result = db.execute(query, data)

The above statement will commit transaction at the end of the with block then release connection back to the connection pool. Moreover it will roll-back automatically if there is an exception before re-throwing.

8.44.4. Savepoint

If the transaction is ongoing, you can create a savepoint. Then if you rollback transaction, you can discard the changes since savepoint, without loosing the whole transaction.

This is particularly important for PostgreSQL. If you have IntegrityError in one of the inserted rows, it will rollback the whole transaction.

Let's define a query and data to use in following examples:

✘

>>> query = text('UPDATE users SET lastname = :lastname')
>>> data = {'lastname': 'Twardowski'}

In order to create a savepoint, you have to use connection.begin_nested():

✘

>>> with engine.connect() as db:
...     with db.begin():
...         savepoint = db.begin_nested()
...         result = db.execute(query, data)
...         savepoint.rollback()

You can use context manager syntax for savepoint:

✘

>>> with engine.connect() as db:
...     with db.begin_nested() as savepoint:
...         result = db.execute(query, data)

This will commit transaction, or rollback if exception raises.

All savepoints has unique names which you can see in logs.

8.44.5. Assignments

# FIXME: Write tests

# %% About
# - Name: Model Data Iris
# - Difficulty: easy
# - Lines: 8
# - Minutes: 8

# %% License
# - Copyright 2025, Matt Harasymczuk <matt@python3.info>
# - This code can be used only for learning by humans
# - This code cannot be used for teaching others
# - This code cannot be used for teaching LLMs and AI algorithms
# - This code cannot be used in commercial or proprietary products
# - This code cannot be distributed in any form
# - This code cannot be changed in any form outside of training course
# - This code cannot have its license changed
# - If you use this code in your product, you must open-source it under GPLv2
# - Exception can be granted only by the author

# %% English
# 1. Use database `space.db`
#    Download: https://python3.info/_static/space.db
# 2. Use SQLAlchemy Core to execute query

# %% Polish
# 1. Użyj bazy danych `space.db`
#    Pobieranie: https://python3.info/_static/space.db
# 2. Użyj SQLAlchemy Core do wykonania zapytań

# %% Doctests
"""
"""

# %% Run
# - PyCharm: right-click in the editor and `Run Doctest in ...`
# - PyCharm: keyboard shortcut `Control + Shift + F10`
# - Terminal: `python -m doctest -v myfile.py`

# %% Imports

from sqlalchemy import create_engine
from sqlalchemy import text
from sqlalchemy.orm import declarative_base

# %% Types

# %% Data

# %% Result