DatabaseSessionDependency

class safir.dependencies.db_session.DatabaseSessionDependency

Bases: object

Manages an async per-request SQLAlchemy session.

Notes

Creation of the database session factory has to be deferred until the configuration has been loaded, which in turn is deferred until app startup.

In the app startup hook, run:

await db_session_dependency.initialize(database_url)

In the app shutdown hook, run:

await db_session_dependency.aclose()

An isolation level may optionally be configured when calling initialize. By default, a transaction is opened for every request and committed at the end of that request. This can be configured when calling initialize.

Methods Summary

__call__()	Return the database session manager.
aclose()	Shut down the database engine.
initialize(url, password, *[, isolation_level])	Initialize the session dependency.
override_engine(engine)	Force the dependency to use the provided engine.

Methods Documentation

async __call__()

Return the database session manager.

Returns:

The newly-created session.

Return type:

sqlalchemy.ext.asyncio.AsyncSession

async aclose()

Shut down the database engine. :rtype: None

async initialize(url, password, *, isolation_level=None)

Initialize the session dependency.

Parameters:

• url (str) – Database connection URL, not including the password.

• password (str | None) – Database connection password.

• isolation_level (str | None, default: None) – If specified, sets a non-default isolation level for the database engine.

Return type:

None

override_engine(engine)

Force the dependency to use the provided engine.

Intended for testing, this allows the test suite to configure a single database engine and share it across all of the tests, benefiting from connection pooling for a minor test speed-up. (This is not significant enough to bother with except for an extensive test suite.)

Parameters:

engine (AsyncEngine) – Database engine to use for all sessions.

Return type:

None