qfa.adapters.db#

Database engine, session factory, SQL schema, and auth adapter.

The llm_calls table declared here backs the usage repository (qfa.adapters.usage_repository); the tenants and keys tables back SQLAlchemyAuthAdapter. The engine + session helpers are the composition root the API layer uses to wire repositories at startup.

Functions

`create_async_engine_from_settings`(settings)	Create an async engine from app DB settings.
`create_session_factory`(engine)	Create a session factory bound to the given engine.
`resolve_database_url`(settings)	Resolve an SQLAlchemy database URL from DB settings.

Classes

SQLAlchemyAuthAdapter(session_factory)

Auth adapter backed by SQLAlchemy and PostgreSQL.

qfa.adapters.db.resolve_database_url(settings: DatabaseSettings) → str[source]#

Resolve an SQLAlchemy database URL from DB settings.

If settings.url is provided, it is returned unchanged. Otherwise the URL is assembled from host/user/port/name and auth mode.

qfa.adapters.db.create_async_engine_from_settings(settings: DatabaseSettings) → AsyncEngine[source]#

Create an async engine from app DB settings.

In entra mode, a fresh AAD access token is injected on each new physical connection via SQLAlchemy’s do_connect hook.

qfa.adapters.db.create_session_factory(engine: AsyncEngine) → async_sessionmaker[AsyncSession][source]#

Create a session factory bound to the given engine.

Parameters:: engine (AsyncEngine) – The async engine to bind sessions to.
Returns:: A factory for creating async sessions.
Return type:: async_sessionmaker[AsyncSession]

class qfa.adapters.db.SQLAlchemyAuthAdapter(session_factory: Callable[[...], AsyncSession])[source]#

Bases: AuthLookupPort, AuthManagementPort

Auth adapter backed by SQLAlchemy and PostgreSQL.

async validate_api_key(provided_key: str) → TenantApiKey | None[source]#: Validate an API key against hashed keys stored in DB.

async get_auth_keys(tenant_id: str | None = None) → list[AuthKeyInfo][source]#: Get API key metadata for one tenant or all tenants.

async add_tenant(tenant_name: str, allows_superusers: bool = False) → str[source]#: Create and persist a tenant record, returning its id.

async delete_tenant(tenant_id: str) → None[source]#: Delete a tenant and all related keys.

async get_tenants() → list[TenantInfo][source]#: Return metadata for all tenants.

async add_key(key_name: str, tenant_id: str, is_superuser: bool = False) → KeyCreationResponse[source]#: Persist a new API key for a tenant.

async delete_key(key_id: str) → None[source]#: Delete an API key by id.