server.web.db.engine

Database engine management using FastAPI lifespan pattern.

This module provides database engine creation and lifecycle management without global singletons. The engine is stored in FastAPI’s app.state and managed through the db_lifespan context manager.

Attributes

log

Functions

`create_engine_from_config`(...)	Create a SQLAlchemy engine from the current configuration.
`db_lifespan`(→ AsyncIterator[None])	Manage database engine lifecycle via FastAPI lifespan.
`is_mysql_dialect`(→ bool)	Check if the dialect is MySQL.
`is_sqlite_dialect`(→ bool)	Check if the dialect is SQLite.

Module Contents

server.web.db.engine.log

server.web.db.engine.create_engine_from_config() → tuple[sqlalchemy.engine.Engine, str, Dict[str, Any]]

Create a SQLAlchemy engine from the current configuration.

Returns:: Tuple of (engine, dialect_name, config_info) where config_info contains the pool settings and connect args used for debugging.

async server.web.db.engine.db_lifespan(app: fastapi.FastAPI) → AsyncIterator[None]

Manage database engine lifecycle via FastAPI lifespan.

Creates the database engine and sessionmaker on startup, stores them in app.state, and properly disposes of the engine on shutdown.

Usage:

app = FastAPI(lifespan=db_lifespan)

# In route handlers:
def get_db(request: Request):
    SessionLocal = request.app.state.db_sessionmaker
    ...

server.web.db.engine.is_mysql_dialect(dialect: str) → bool: Check if the dialect is MySQL.

server.web.db.engine.is_sqlite_dialect(dialect: str) → bool: Check if the dialect is SQLite.