db

azure_bootstrap.db

Relational database access layer — SQLAlchemy 2 engine/session factory.

Modules:

Name	Description
migrations	Alembic migration conventions — env-driven harness.
outbox	Transactional outbox — reliable send with idempotency and claim/drain.

Functions:

Name	Description
create_engine_from_env	Build a SQLAlchemy engine from DATABASE_URL (or custom env var).
get_engine	Lazy singleton engine.
get_db	FastAPI-style session dependency — yields and always closes.
db_health	Return {status, latency_ms} for health probes.
postgres_rls_statements	Idempotent RLS DDL for multi-tenant Postgres tables.

create_engine_from_env

create_engine_from_env(*, dsn_env: str = 'DATABASE_URL', **kwargs: Any) -> Any

Build a SQLAlchemy engine from DATABASE_URL (or custom env var).

Source code in azure_bootstrap/db/__init__.py

def create_engine_from_env(*, dsn_env: str = "DATABASE_URL", **kwargs: Any) -> Any:
    """Build a SQLAlchemy engine from ``DATABASE_URL`` (or custom env var)."""
    from sqlalchemy import create_engine  # type: ignore[import-untyped]

    dsn = require_env(dsn_env)
    defaults: dict[str, Any] = {"pool_pre_ping": True, "future": True}
    defaults.update(kwargs)
    if not str(dsn).lower().startswith("sqlite"):
        defaults["pool_size"] = int(defaults.get("pool_size", 5))
        defaults["max_overflow"] = int(defaults.get("max_overflow", 10))
    return create_engine(dsn, **defaults)

get_engine

get_engine(*, dsn_env: str = 'DATABASE_URL', **kwargs: Any) -> Any

Lazy singleton engine.

Source code in azure_bootstrap/db/__init__.py

def get_engine(*, dsn_env: str = "DATABASE_URL", **kwargs: Any) -> Any:
    """Lazy singleton engine."""
    global _engine, _sessionmaker
    with _lock:
        if _engine is None:
            _engine = create_engine_from_env(dsn_env=dsn_env, **kwargs)
            from sqlalchemy.orm import sessionmaker  # type: ignore[import-untyped]

            _sessionmaker = sessionmaker(bind=_engine, expire_on_commit=False)
        return _engine

get_db

get_db(*, dsn_env: str = 'DATABASE_URL') -> Generator[Any, None, None]

FastAPI-style session dependency — yields and always closes.

Source code in azure_bootstrap/db/__init__.py

def get_db(*, dsn_env: str = "DATABASE_URL") -> Generator[Any, None, None]:
    """FastAPI-style session dependency — yields and always closes."""
    Session = get_sessionmaker(dsn_env=dsn_env)
    session = Session()
    try:
        yield session
    finally:
        session.close()

db_health

db_health(*, dsn_env: str = 'DATABASE_URL') -> dict[str, Any]

Return {status, latency_ms} for health probes.

Source code in azure_bootstrap/db/__init__.py

def db_health(*, dsn_env: str = "DATABASE_URL") -> dict[str, Any]:
    """Return ``{status, latency_ms}`` for health probes."""
    from sqlalchemy import text  # type: ignore[import-untyped]

    start = time.perf_counter()
    try:
        engine = get_engine(dsn_env=dsn_env)
        with engine.connect() as conn:
            conn.execute(text("SELECT 1"))
        latency = (time.perf_counter() - start) * 1000
        return {"status": "ok", "latency_ms": round(latency, 2)}
    except Exception as exc:
        return {"status": "error", "latency_ms": None, "error": str(exc)}

postgres_rls_statements

postgres_rls_statements(table: str, tenant_col: str = 'tenant_id') -> list[str]

Idempotent RLS DDL for multi-tenant Postgres tables.

Source code in azure_bootstrap/db/__init__.py

def postgres_rls_statements(table: str, tenant_col: str = "tenant_id") -> list[str]:
    """Idempotent RLS DDL for multi-tenant Postgres tables."""
    return [
        f"ALTER TABLE {table} ENABLE ROW LEVEL SECURITY;",
        f"ALTER TABLE {table} FORCE ROW LEVEL SECURITY;",
        f"""CREATE POLICY IF NOT EXISTS tenant_isolation ON {table}
            USING ({tenant_col} = current_setting('app.current_tenant_id', true));""",
    ]