Runtime and providers

class alchemiq.UnitOfWork

Bases: object

One async with = one transaction. Autocommit on clean exit, rollback on error.

Reentrant: a nested UnitOfWork joins the active session and its exit is a no-op; only the outermost commits and closes the session. Use savepoint() for partial rollback within a single transaction.

E.g.:

# basic usage - commits on clean exit
async with UnitOfWork() as uow:
    uow.session.add(MyRow(id=1, name="a"))

# savepoint - inner error rolls back only to the savepoint
async with UnitOfWork() as uow:
    uow.session.add(MyRow(id=3, name="keep"))
    await uow.session.flush()
    async with uow.savepoint():
        uow.session.add(MyRow(id=4, name="drop"))
        raise RuntimeError("inner")  # only the savepoint is rolled back

# reentrant - nested UoW joins the outer session
async with UnitOfWork() as outer:
    async with UnitOfWork() as inner:
        assert inner.session is outer.session  # same session, no-op exit

See also

UnitOfWork.savepoint() - nested savepoint for partial rollback.

async commit()

Flush and commit the current transaction.

Raises:

RuntimeError – if called on a nested (joined) UnitOfWork - only the outermost may commit.

Return type:

None

async rollback()

Abort the current transaction without closing the session.

Raises:

RuntimeError – if called on a nested (joined) UnitOfWork. Raise an exception to abort the outer transaction instead, or use savepoint() for partial rollback.

Return type:

None

savepoint()

Yield a nested savepoint; rolls back only to this point on error.

E.g.:

async with UnitOfWork() as uow:
    uow.session.add(MyRow(id=1, name="keep"))
    await uow.session.flush()
    async with uow.savepoint():
        uow.session.add(MyRow(id=2, name="drop"))
        raise RuntimeError("abort inner")  # only savepoint rolled back
    # id=1 is still pending; commits with the outer transaction

Return type:

AsyncIterator[None]

alchemiq.configure(dsn, *, echo=False, **engine_kwargs)

Create the process-global async engine + sessionmaker. Call once at startup.

Stores the engine and sessionmaker as module-level singletons used by every UnitOfWork and session_scope call. A second call replaces them (useful in tests that cycle databases).

E.g.:

# pytest conftest.py - set up once per session
alchemiq.configure(pg_container.get_connection_url())
await alchemiq.create_all()

# application lifespan
alchemiq.configure(os.environ["DATABASE_URL"], echo=False)

Parameters:

• dsn (str) – async-compatible SQLAlchemy DSN, e.g. postgresql+asyncpg://....

• echo (bool) – when True, the engine logs all SQL statements (useful for debugging).

• engine_kwargs (Any) – additional keyword arguments forwarded to create_async_engine (e.g. pool_size, max_overflow).

Return type:

None

async alchemiq.create_all()

Create all mapped tables in the database (DDL CREATE IF NOT EXISTS).

Return type:

None

async alchemiq.drop_all()

Drop all mapped tables from the database (DDL DROP IF EXISTS). Destructive.

Return type:

None

async alchemiq.dispose()

Dispose the engine and clear state (lifespan shutdown / test teardown).

Return type:

None

alchemiq.configure_password_hashing(scheme)

Set the process-global scheme used to hash new passwords.

The default is scrypt (stdlib, no extra). argon2 requires the [argon2] extra and bcrypt the [bcrypt] extra; the backend is checked eagerly so a misconfiguration fails at startup, not at first hash. Existing stored hashes keep verifying regardless - verify_password() dispatches on each hash’s own format.

Parameters:

scheme (Literal['scrypt', 'argon2', 'bcrypt']) – one of "scrypt", "argon2", "bcrypt".

Raises:

ConfigError – on an unknown scheme or a missing backend extra.

Return type:

None

alchemiq.reset_password_hashing()

Restore the default scrypt scheme. Intended for tests and teardown.

Return type:

None