UXDL Docs
Backend

PostgreSQL + SQLAlchemy

Python — SQLAlchemy 2.0 models, Alembic migrations, and async sessions.

SQLAlchemy 2.0 is the standard ORM for Python (FastAPI) services. We use the async engine with asyncpg and Alembic for migrations — the Python counterpart to Drizzle / Sequelize on the Node side.

Install

bash
uv add "sqlalchemy[asyncio]" asyncpg alembic

Models

python
# app/models/base.py
from sqlalchemy.orm import DeclarativeBase
 
 
class Base(DeclarativeBase):
    pass
python
# app/models/user.py
import uuid
from datetime import datetime
 
from sqlalchemy import String, func
from sqlalchemy.orm import Mapped, mapped_column
 
from app.models.base import Base
 
 
class User(Base):
    __tablename__ = "users"
 
    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    cognito_sub: Mapped[str] = mapped_column(String(255), unique=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
python
# app/models/project.py
import uuid
from datetime import datetime
 
from sqlalchemy import ForeignKey, String, func
from sqlalchemy.orm import Mapped, mapped_column
 
from app.models.base import Base
 
 
class Project(Base):
    __tablename__ = "projects"
 
    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
    name: Mapped[str] = mapped_column(String(255))
    owner_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

Async engine & session

python
# app/db/session.py
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 
from app.core.config import settings
 
engine = create_async_engine(settings.database_url, pool_size=10, echo=False)
async_session = async_sessionmaker(engine, expire_on_commit=False)

Expose the session as a FastAPI dependency so routes and services get a scoped, auto-closed session:

python
# app/api/deps.py (continued)
from collections.abc import AsyncGenerator
from typing import Annotated
 
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
 
from app.db.session import async_session
 
 
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        yield session
 
 
DbSession = Annotated[AsyncSession, Depends(get_db)]

Queries

python
# app/services/projects.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
 
from app.models.project import Project
from app.schemas.projects import ProjectCreate
 
 
async def list_by_owner(db: AsyncSession, owner_id: str) -> list[Project]:
    result = await db.execute(select(Project).where(Project.owner_id == owner_id))
    return list(result.scalars().all())
 
 
async def create(db: AsyncSession, owner_id: str, data: ProjectCreate) -> Project:
    project = Project(name=data.name, owner_id=owner_id)
    db.add(project)
    await db.commit()
    await db.refresh(project)
    return project

The session is injected from the route, keeping services free of FastAPI imports:

python
# app/api/v1/projects.py
@router.get("", response_model=list[ProjectOut])
async def list_projects(db: DbSession, user: CurrentUser):
    return await projects_service.list_by_owner(db, user.sub)

Migrations (Alembic)

bash
uv run alembic init -t async app/db/migrations

Point Alembic at the metadata and URL:

python
# app/db/migrations/env.py (key lines)
from app.core.config import settings
from app.models.base import Base
# import all models so they register on Base.metadata
import app.models.user  # noqa: F401
import app.models.project  # noqa: F401
 
target_metadata = Base.metadata
config.set_main_option("sqlalchemy.url", settings.database_url)
bash
uv run alembic revision --autogenerate -m "create users and projects"
uv run alembic upgrade head      # apply
uv run alembic downgrade -1      # rollback one

Scripts

bash
# Common commands (run via uv)
uv run alembic revision --autogenerate -m "<message>"   # generate migration
uv run alembic upgrade head                             # apply migrations
uv run alembic downgrade -1                             # rollback

Official documentation