Database Patterns
PostgreSQL + Alembic + async SQLAlchemy. Migrations gated by Dark Factory's pre-upgrade Job. Repository pattern. Async by default.
The problem
Migrations break prod. Async session management is repeatedly botched.
The two reliable ways to break prod: (1) ship a migration that locks a table for 20 minutes, or (2) leak a synchronous DB session into an async event loop. Both happen on every team eventually.
The third reliable footgun: SQLAlchemy's reserved metadata attribute
silently shadowing your column. The fourth: forgetting to review the auto-generated
Alembic migration before merging. All of these are well-known; none are caught by
default tooling.
How it works
Schema-first, migration-gated, async by default — with sane patterns.
Sage scaffolds with SQLAlchemy 2.0 typed models, async sessions throughout, and a repository pattern that keeps query logic out of route handlers. Alembic is wired with autogenerate; the cycle-doc gate requires a migration review section in the cycle doc before code lands.
Migrations are gated at deploy time too: a pre-upgrade Helm Job
runs alembic upgrade head before the rollout proceeds. A failed
migration aborts the rollout — prior pods keep serving. No more "deploy succeeded
but DB is broken" outages.
- PostgreSQL + SQLAlchemy 2.0 typed models (Mapped[T] + mapped_column)
- Alembic migrations with autogenerate; cycle-doc gate requires migration review per cycle
- Pre-upgrade Helm Job runs alembic upgrade head before pods roll forward — failed migration aborts
- Async sessions throughout: AsyncSession, async with for transactions, no sync session leaks
- Repository pattern — query logic in repos, not in routes; testable, mockable, reusable
- extra_metadata convention (metadata is reserved in SQLAlchemy); JSONB columns for flexible state
# backend/app/models.py — SQLAlchemy 2.0 typed models
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import String, ForeignKey, DateTime
class Conversation(Base):
__tablename__ = "conversations"
id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
org_id: Mapped[UUID] = mapped_column(ForeignKey("organizations.id"))
title: Mapped[str] = mapped_column(String(200))
created_at: Mapped[datetime] = mapped_column(DateTime, default=utcnow)
extra_metadata: Mapped[dict] = mapped_column(JSONB, default=dict)
# NOTE: metadata is reserved in SQLAlchemy — always use extra_metadata.
# backend/app/repositories/conversation.py — async repo pattern
class ConversationRepo:
def __init__(self, session: AsyncSession):
self.session = session
async def list_for_org(self, org_id: UUID, limit: int = 50) -> list[Conversation]:
stmt = (
select(Conversation)
.where(Conversation.org_id == org_id)
.order_by(Conversation.created_at.desc())
.limit(limit)
)
return list((await self.session.scalars(stmt)).all())
# Migration: alembic revision --autogenerate + reviewed in DF cycle doc
# Pre-upgrade Helm Job runs alembic upgrade head before pods roll forward;
# failed migration aborts the rollout — prior pods keep serving.Get Started
Stop shipping migrations that break prod.
Pre-upgrade Helm Job runs alembic upgrade head before pods roll. Failed migration aborts the rollout. Async sessions throughout. Repository pattern. Sane defaults.