Database¶

Async SQLAlchemy 2.x integration with asyncpg. Provides a declarative base, engine management, and a FastAPI dependency for session injection.

BluefoxBase¶

The shared declarative base for all SQLAlchemy models:

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column
from bluefox_core import BluefoxBase

class User(BluefoxBase):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(255))

get_session¶

FastAPI dependency that yields an AsyncSession:

from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from bluefox_core import get_session

router = APIRouter()

@router.get("/users")
async def list_users(session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User))
    return result.scalars().all()

If no database is configured, get_session raises HTTP 503.

DatabaseManager¶

Manages the async engine lifecycle. You typically don't interact with it directly — the app factory handles it.

from bluefox_core.database import DatabaseManager

manager = DatabaseManager(settings)
await manager.startup()   # verifies connectivity with SELECT 1
await manager.shutdown()   # disposes the engine

The manager is stored on app.state.db and its startup() / shutdown() methods are called automatically in the app lifespan.

URL normalization¶

If DATABASE_URL starts with postgresql://, it is automatically converted to postgresql+asyncpg:// for asyncpg compatibility. This means you can use the same DATABASE_URL format as tools like psql.

Running without a database¶

When DATABASE_URL is empty, the app starts normally — no engine is created, app.state.db.engine is None, and the /readiness endpoint skips the database check.