qfa.adapters.db#

SQLAlchemy-based usage repository for LLM call tracking.

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

SqlAlchemyUsageRepository(session_factory)

Usage repository 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.SqlAlchemyUsageRepository(session_factory: Callable[[...], AsyncSession])[source]#

Bases: UsageRepositoryPort

Usage repository backed by SQLAlchemy and PostgreSQL.

Parameters:: session_factory (Callable[..., AsyncSession]) – Factory for creating async database sessions.

async record_call(record: LLMCallRecord) → None[source]#: Insert a single LLM call attempt record.

async get_usage_stats(tenant_id: str, from_: datetime | None = None, to: datetime | None = None) → UsageStats[source]#

Aggregate stats for a single tenant within an optional time window.

Cost, distributions, and token totals scope to status='ok' rows. total_calls and failed_calls count all rows in the window.

async get_all_usage_stats(from_: datetime | None = None, to: datetime | None = None) → list[UsageStats][source]#

Per-tenant stats plus a grand total entry (tenant_id=None).

Tenants are returned alphabetically by tenant_id; the grand-total entry is appended last. Implemented as a single GROUPING SETS query (Postgres-only) so cost is O(1) round-trips regardless of tenant count.