Database API

class fastapi_restly.db.FRGlobals

Bases: object

async_database_url: str | None

async_make_session: async_sessionmaker[AsyncSession] | None

database_url: str | None

make_session: sessionmaker[Session] | None

session_generator: Callable[[], AsyncIterator[AsyncSession]] | None

sync_session_generator: Callable[[], Iterator[Session]] | None

fastapi_restly.db.activate_savepoint_only_mode(make_session: async_sessionmaker | sessionmaker) → None

Intended for use in tests. Puts the session factory into savepoint-only mode so that no test data is ever committed to the database. Each test can roll back instantly by closing the session, leaving the database clean for the next test.

This is done with “create_savepoint” mode and a wrapper on engine.connect() that begins the outer transaction before the Session can use it. https://docs.sqlalchemy.org/en/20/orm/session_transaction.html#session-external-transaction

async fastapi_restly.db.async_generate_session() → AsyncIterator[AsyncSession]

FastAPI dependency for async database session.

fastapi_restly.db.async_session() → AsyncIterator[AsyncSession]

Open an async database session for use outside of request context.

Example:

async with fr.async_session() as session:
    result = await session.execute(select(User))

fastapi_restly.db.configure(*, async_database_url: str | None = None, async_engine: AsyncEngine | None = None, async_make_session: async_sessionmaker | None = None, database_url: str | None = None, engine: Engine | None = None, make_session: sessionmaker | None = None, session_generator: Callable[[], AsyncIterator[AsyncSession]] | None = None, sync_session_generator: Callable[[], Iterator[Session]] | None = None) → None

Configure FastAPI-Restly. Call once at startup.

Pass async parameters (async_database_url, async_engine, or async_make_session) to enable async support, sync parameters (database_url, engine, or make_session) for sync support, or both if your application uses both.

Use session_generator / sync_session_generator to plug in a custom session factory instead of the built-in one.

fastapi_restly.db.deactivate_savepoint_only_mode(make_session: async_sessionmaker | sessionmaker) → None

Reverts the effect of activate_savepoint_only_mode. Restores the original engine.connect and disables savepoint-only mode.

fastapi_restly.db.generate_session() → Iterator[Session]

FastAPI dependency for sync database session.

fastapi_restly.db.get_async_engine() → AsyncEngine

Return the async engine registered via configure().

fastapi_restly.db.get_engine() → Engine

Return the sync engine registered via configure().

fastapi_restly.db.get_fr_globals() → FRGlobals

fastapi_restly.db.session() → Iterator[Session]

Open a sync database session for use outside of request context.

Example:

with fr.session() as session:
    result = session.execute(select(User))

fastapi_restly.db.use_fr_globals(globals_obj: FRGlobals) → Iterator[None]