hawkapi-sqlalchemy

SQLAlchemy integration for HawkAPI. Async sessions, multi-database routing (primary/replica/shards), Alembic helpers, and pytest fixtures.

Install

pip install hawkapi-sqlalchemy                     # SQLite included
pip install 'hawkapi-sqlalchemy[postgres]'         # + asyncpg
pip install 'hawkapi-sqlalchemy[mysql]'            # + aiomysql

Quickstart

from hawkapi import Depends, HawkAPI
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import Mapped, mapped_column

from hawkapi_sqlalchemy import Base, TimestampMixin, get_session, init_database


class User(Base, TimestampMixin):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    email: Mapped[str] = mapped_column(unique=True)


app = HawkAPI()
init_database(app, url="postgresql+asyncpg://user:pw@localhost/app")


@app.post("/users")
async def create(email: str, sess: AsyncSession = Depends(get_session)):
    sess.add(User(email=email))
    await sess.flush()
    return {"ok": True}

The get_session dependency opens a fresh session per request, commits on success, and rolls back on exception — no boilerplate.

Multiple databases

from hawkapi_sqlalchemy import DatabaseConfig, init_database, session_for

init_database(
    app,
    databases={
        "primary": DatabaseConfig(url="postgresql+asyncpg://…/primary"),
        "replica": DatabaseConfig(url="postgresql+asyncpg://…/replica"),
        "analytics": DatabaseConfig(url="postgresql+asyncpg://…/analytics"),
    },
)

# DI helpers:
from hawkapi_sqlalchemy import get_session, get_replica_session

get_analytics = session_for("analytics", commit=False)


@app.get("/report")
async def report(sess: AsyncSession = Depends(get_analytics)):
    ...

get_replica_session falls back to primary if no replica is registered, so you can switch on without changing handlers.

Mixins

from hawkapi_sqlalchemy import Base, TimestampMixin, UUIDMixin


class Doc(Base, UUIDMixin, TimestampMixin):
    __tablename__ = "docs"
    title: Mapped[str] = mapped_column()

• TimestampMixin — created_at / updated_at with DB-side defaults + Python onupdate.
• UUIDMixin — string id column with a uuid4() default.
• Prefer DataclassBase over Base to get SQLAlchemy 2.0's dataclass-style declarative.

Alembic

In your alembic/env.py:

from hawkapi_sqlalchemy.alembic import run_migrations
from myapp.db import Base, settings  # your Base + URL

run_migrations(target_metadata=Base.metadata, url=settings.database_url)

That's it — handles both online (live connection) and offline (--sql) modes; uses NullPool for migrations; enables render_as_batch=True automatically for SQLite.

Healthchecks

from hawkapi_sqlalchemy import all_healthy


@app.get("/healthz")
async def healthz():
    return await all_healthy(app.state.db)

Returns {"primary": True, "replica": True, ...}.

Testing

import pytest
from hawkapi_sqlalchemy import Base, temporary_database


@pytest.fixture
async def db():
    async with temporary_database(Base.metadata) as database:
        yield database


async def test_something(db):
    async with db.session() as sess:
        ...

temporary_database creates an in-memory SQLite engine, calls Base.metadata.create_all, yields, then drops the schema and disposes.

DatabaseConfig

DatabaseConfig(
    url="postgresql+asyncpg://…",
    echo=False,
    pool_size=5,
    max_overflow=10,
    pool_timeout=30.0,
    pool_recycle=3600,
    pool_pre_ping=True,
    connect_args={"server_settings": {"jit": "off"}},
    engine_kwargs={...},     # forwarded to create_async_engine
    session_kwargs={...},    # forwarded to async_sessionmaker
)

Development

git clone https://github.com/Hawk-API/hawkapi-sqlalchemy.git
cd hawkapi-sqlalchemy
uv sync --extra dev
uv run pytest -q
uv run ruff check . && uv run ruff format --check .
uv run pyright src/

License

MIT.