sqlalchemy-boltons 5.0.0

Utilities that should've been inside SQLAlchemy but aren't

sqlalchemy_boltons

SQLAlchemy is great. However, it doesn't have everything built-in. Some important things are missing, and need to be "bolted on".

(Name inspired from boltons. Not affiliated.)

sqlite

SQLAlchemy doesn't automatically fix pysqlite's broken transaction handling. This module implements the usual fix for that well-known broken behaviour, and also adds extra features on top of that.

You can customize, on a per-engine or per-connection basis:

• begin: The type of transaction to be started, such as "BEGIN DEFERRED" (the default) or "BEGIN IMMEDIATE" (or "BEGIN CONCURRENT" someday maybe).
• foreign_keys: The foreign-key enforcement setting. Can be IMMEDIATE, DEFERRED, or OFF.
• journal: The journal mode such as DELETE or WAL.

Here's a minimal example:

from sqlalchemy.orm import sessionmaker
from sqlalchemy_boltons import sqlite as sq

engine = sq.create_engine_sqlite("file.db", create_engine_args={"echo": True})

# Configure the engine to use a plain "BEGIN" to start transactions and
# and to use deferred enforcement of foreign keys (recommended!)
engine = sq.Options.new(
    timeout=0.5,
    begin="DEFERRED",
    foreign_keys="DEFERRED",
    recursive_triggers=True,
    trusted_schema=False,  # set to True if the db file is from a trusted source
    schemas={"main": sq.SchemaOptions.new(journal="WAL", synchronous="NORMAL")},
).apply(engine)

# Make a separate engine for write transactions using "BEGIN IMMEDIATE"
# for eager locking.
engine_w = sq.Options.apply_lambda(
    engine, lambda options: options.evolve(begin="IMMEDIATE")
)

# Construct a sessionmaker for each engine.
SessionR = sessionmaker(engine)
SessionW = sessionmaker(engine_w)

# read-only transaction
with SessionR() as session:
    session.execute(select(...))

# lock the database eagerly for writing
with SessionW() as session:
    session.execute(update(...))

orm

orm.RelationshipComparator

Somedays you really wish you could write something like:

# Find every parent that has at least one child.
Parent1 = aliased(Parent)
select(Parent1).where(
    exists().select_from(Child).where(Child.parent == Parent1)  # this doesn't work
)

But it doesn't work. You get an exception. You try various other things but it just won't produce the right subquery. You find tons of people online saying you must resign yourself to writing the explicit filtering condition, writing out .where(Child.parent_id == Parent1.id). Like a caveman. Why even bother using an ORM at this point? If you're lucky, you find yourself reading mail archives from a decade ago with somewhat of a solution, but it's not obvious how to make it work with table aliases. Despair is setting in.

But lucky you, you can now use this library:

from sqlalchemy_boltons.orm import RelationshipComparator as Rel

# Find every parent that has at least one child.
Parent1 = aliased(Parent)
select(Parent1).where(
    exists().select_from(Child).where(Rel(Child.parent) == Parent1)
)  #                                  ^^^^            ^

Hope is restored. This even works on self-referential tables!

orm.IdKey

IdKey holds the primary key information for an ORM instance. This is useful for passing a reference to an ORM object across different sessions and/or threads. This is implemented in terms of Session.identity_key and Session.get, so you don't have to mess around with Session.merge.

with Session() as s1:
    instance = s1.execute(sa.select(MyClass).where(...)).one()
    key = IdKey.from_instance(instance)

# ...later, maybe in a different thread:
with Session() as s2:
    instance = key.get_one(s2)  # raises sqlalchemy.exc.NoResultFound if object doesn't exist anymore

core

core.bytes_startswith(sql_expr, prefix: bytes)

This function constructs a SQL expression equivalent to sql_expr.startswith(prefix) but using comparison operators instead. The intended use case is efficiently searching a BLOB-like column that has an index on it.

temporary_table

Create a temporary table so you can write things like:

import sqlalchemy as sa
from sqlalchemy_boltons.temporary import CacheTemporaryMetaData

tmp_table_def = sa.Table("ids", sa.MetaData(), sa.Column("id", sa.Integer, primary_key=True))
tmp_cache = CacheTemporaryMetaData(tmp_table_def)

...

my_ids = [1, 23, 4, 5, 67, 89]

with Session() as s:
    with tmp_cache(s, "") as tmp:
        s.execute(sa.insert(tmp), [{"id": x} for x in my_ids])
        s.flush()
        results = s.execute(sa.select(SomeOtherTable).join(tmp, tmp.c.id == SomeTable.id)).all()

Do you need a more complicated temporary schema with multiple tables?

import dataclasses as dc
import sqlalchemy as sa
import sqlalchemy.orm as sao
from sqlalchemy_boltons.temporary import CacheTemporaryMetaData, MetaDataMaker


@dc.dataclass
class MySchema:
    MappedObject: type
    table1: sa.Table
    table2: sa.Table


def _make_tmp(maker: MetaDataMaker):
    class Base(sao.DeclarativeBase):
        metadata = maker.metadata

    @Base.registry.mapped_as_dataclass(init=False)
    class MappedObject:
        __tablename__ = maker.table_name("mapped")
        mapped_id: sao.Mapped[int] = sao.mapped_column(primary_key=True)

    t1 = maker.table("table1", sa.Column("integer_id", sa.Integer()))
    t2 = maker.table(
        "table2",
        sa.Column("string", sa.String()),
        sa.Column("fk_t1", sa.ForeignKey(f"{t1.name}.integer_id")),
        sa.Column("fk_mapped", sa.ForeignKey(f"{maker.table_name("mapped")}.mapped_id")),
    )

    return MySchema(MappedObject=MappedObject, table1=t1, table2=t2)


tmp_cache = CacheTemporaryMetaData(_make_tmp)

...

with Session() as s:
    with tmp_cache(s, "") as tmp:  # tmp is an instance of MySchema
        s.add(tmp.MappedObject(mapped_id=5))
        s.flush()

reset

PostgreSQL and MS SQL both require additional things to be done to fully clean up before reusing a connection. A mere "ROLLBACK" is insufficient. By default, SQLAlchemy doesn't do those things.

from sqlalchemy_boltons.reset import install_reset_auto

engine = create_engine(..., pool_reset_on_return=False)
install_reset_auto(engine)