app.database module#

Database engines and session factories (sync and async) for the app.

This module centralizes SQLAlchemy setup. It exposes reusable, process-wide engine singletons and companion session factories for both synchronous and asynchronous usage. When targeting SQLite, it applies pragmatic defaults to improve concurrency (WAL) and resilience (busy timeout) via connection event hooks. All ORM models inherit from the exported Base.

See Also#

app.models

Declarative ORM models (bound to app.database.Base).

app.schemas

Pydantic schemas mirroring ORM shapes.

app.main

FastAPI app wiring request-scoped DB sessions.

app.config

Source of settings.DATABASE_URL.

Notes#

  • Primary role: provide ready-to-use SQLAlchemy engines and session factories and a declarative Base for model definitions.

  • Key dependencies: app.config.settings.DATABASE_URL must be defined; SQLAlchemy (sync/async) drivers must be available for the chosen URL.

  • Invariants: engines are module-level singletons; SQLite engines use check_same_thread=False and have PRAGMA journal_mode=WAL and busy_timeout set at connect time.

Examples#

>>> # Ensure schema exists (tables are created if missing)
>>> from app.database import ensure_database_schema
>>> ensure_database_schema()

>>> # Sync usage
>>> from app.database import SessionLocal
>>> with SessionLocal() as session:
...     pass

>>> # Async usage
>>> from app.database import AsyncSessionLocal
>>> async def get_count() -> int:
...     async with AsyncSessionLocal() as session:
...         return 0
class app.database.Base(*args: Any, **kwargs: Any)[source]#

Bases: DeclarativeBase

Declarative base for all ORM models.

All SQLAlchemy ORM models in this project must inherit from this base. The metadata bound to this base is used by ensure_database_schema() to create database tables.

Notes

  • Exposed so that app.models can subclass it for all entities.

  • Keeps a single metadata registry for consistent schema management.

app.database.ensure_database_schema() → None[source]#

Create missing tables based on ORM metadata.

Delegates to SQLAlchemy’s Base.metadata.create_all using the configured synchronous engine. This is idempotent and safe to call during startup or migrations bootstrap.

Raises:
SQLAlchemyError

If the schema creation step fails (for example due to missing privileges or an unavailable database).

See also

app.models

Entities whose tables are created from the metadata.

app.main

Application startup that may call this function.

Examples

>>> from app.database import ensure_database_schema
>>> ensure_database_schema()