Python Cheatsheet

Quick reference for mapping global architecture concepts to Python implementation.

Global → Python Mapping

Service Patterns (Python)

See: gts-backend-dev for full details, service-patterns (global) for concepts

Quick reference:

class ShootoutService:
    """Orchestrates domain logic. Depends on ports, not adapters."""

    def __init__(
        self,
        repo: ShootoutRepository,      # Port (Protocol)
        job_queue: JobQueuePort,       # Port (Protocol)
    ):
        self._repo = repo
        self._job_queue = job_queue

    async def create_shootout(
        self,
        user_id: UUID,
        data: CreateShootoutData,
    ) -> Shootout:
        """Domain orchestration - no infrastructure knowledge."""
        shootout = Shootout.create(user_id=user_id, title=data.title)

        for tone in data.tones:
            shootout.add_tone(tone)  # Domain logic in entity

        await self._repo.save(shootout)
        await self._job_queue.enqueue(RenderShootoutJob(shootout.id))

        return shootout

Dependency injection (composition root):

# bootstrap.py or api/deps.py
def create_shootout_service(session: AsyncSession) -> ShootoutService:
    """Wire adapters to ports at the composition root."""
    return ShootoutService(
        repo=SQLAlchemyShootoutRepository(session),
        job_queue=TaskIQJobQueue(broker),
    )

# FastAPI dependency
async def get_shootout_service(
    session: AsyncSession = Depends(get_session),
) -> ShootoutService:
    return create_shootout_service(session)

Repository Patterns (Python)

See: gts-backend-dev for full details, repository-patterns (global) for concepts

Protocol-based ports (preferred):

from typing import Protocol
from uuid import UUID

class ShootoutRepository(Protocol):
    """Port for shootout persistence."""

    async def get_by_id(self, id: UUID) -> Shootout | None:
        """Retrieve shootout by ID."""
        ...

    async def save(self, shootout: Shootout) -> None:
        """Persist shootout."""
        ...

SQLAlchemy 2.0 adapter:

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

class SQLAlchemyShootoutRepository:
    """Adapter: implements ShootoutRepository port with SQLAlchemy."""

    def __init__(self, session: AsyncSession):
        self._session = session

    async def get_by_id(self, id: UUID) -> Shootout | None:
        stmt = select(ShootoutModel).where(ShootoutModel.id == id)
        result = await self._session.execute(stmt)
        model = result.scalar_one_or_none()
        return ShootoutMapper.to_entity(model) if model else None

    async def save(self, shootout: Shootout) -> None:
        model = ShootoutMapper.to_model(shootout)
        self._session.add(model)
        await self._session.flush()

SQLAlchemy 2.0 models (typed):

from sqlalchemy import ForeignKey, String, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.database import Base

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    tone3000_id: Mapped[int] = mapped_column(unique=True, index=True)
    username: Mapped[str] = mapped_column(String(100))

    # Relationships
    shootouts: Mapped[list["Shootout"]] = relationship(
        back_populates="user",
        cascade="all, delete-orphan"
    )

class Shootout(Base):
    __tablename__ = "shootouts"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    title: Mapped[str] = mapped_column(String(200))
    config_json: Mapped[str] = mapped_column(Text)

    user: Mapped["User"] = relationship(back_populates="shootouts")

Queries — Single Query via joinedload:

CRITICAL: See .claude/rules/query-patterns.md. Never use selectinload — always joinedload.

from sqlalchemy import select
from sqlalchemy.orm import joinedload

# Single entity — full aggregate in ONE query
stmt = (
    select(Gear)
    .where(Gear.id == gear_id)
    .options(
        joinedload(Gear.make),
        joinedload(Gear.source),
        joinedload(Gear.models),
        joinedload(Gear.tags),
    )
)
result = await db.execute(stmt)
gear = result.unique().scalar_one_or_none()  # .unique() REQUIRED

# Paginated list — ID subquery avoids LIMIT on cartesian rows
id_stmt = (
    select(Shootout.id)
    .where(Shootout.user_id == user_id)
    .order_by(Shootout.created_at.desc())
    .offset(skip)
    .limit(limit)
)
stmt = (
    select(Shootout)
    .where(Shootout.id.in_(id_stmt))
    .options(joinedload(Shootout.chains))
    .order_by(Shootout.created_at.desc())
)
result = await db.execute(stmt)
shootouts = result.unique().scalars().all()

# Chained — nested JOINs in one query
stmt = (
    select(User)
    .options(
        joinedload(User.identities).joinedload(UserIdentity.provider)
    )
)

ORM relationship defaults:

# CORRECT — lazy="raise" forces explicit joinedload in repositories
models: Mapped[list[GearModel]] = relationship(
    "GearModel", back_populates="gear",
    cascade="all, delete-orphan",
    lazy="raise",
)

# BANNED — fires separate query
models: Mapped[list[GearModel]] = relationship(..., lazy="selectin")

Error Handling (Python)

See: error-handling (global) for concepts

Custom exceptions:

class AppError(Exception):
    """Base application error."""

class NotFoundError(AppError):
    """Resource not found."""

class UnauthorizedError(AppError):
    """User not authorised."""

FastAPI error handling:

from fastapi import HTTPException, status

@router.get("/{id}")
async def get_item(id: int, db: AsyncSession = Depends(get_db)):
    item = await db.get(Item, id)
    if not item:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item {id} not found"
        )
    return item

Web Handlers (Python)

See: gts-backend-dev for full details, web-handlers (global) for concepts

FastAPI route pattern:

from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

router = APIRouter(prefix="/shootouts", tags=["shootouts"])

@router.get("/{shootout_id}", response_model=ShootoutRead)
async def get_shootout(
    shootout_id: int,
    db: Annotated[AsyncSession, Depends(get_db)],
    current_user: Annotated[User, Depends(get_current_user)],
) -> Shootout:
    """Get a shootout by ID."""
    shootout = await db.get(Shootout, shootout_id)
    if not shootout or shootout.user_id != current_user.id:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Shootout not found"
        )
    return shootout

Pydantic schemas:

from pydantic import BaseModel, ConfigDict

class ShootoutBase(BaseModel):
    title: str
    config_json: str

class ShootoutCreate(ShootoutBase):
    pass

class ShootoutRead(ShootoutBase):
    model_config = ConfigDict(from_attributes=True)

    id: int
    user_id: int

Transaction Patterns (CRITICAL)

See: gts-backend-dev Section “Transactions” for full details

Service layer transaction ownership:

# In services (gear_item_service.py, signal_chain_service.py, etc.)
context_manager = (
    session.begin_nested() if session.in_transaction() else session.begin()
)
async with context_manager:
    # ... do work, add(), flush() ...
# Savepoint released (if nested) or committed (if begin)

The Contract:

• in_transaction()=False → Service owns commit via begin()
• in_transaction()=True → Caller must commit (service uses savepoint)

Common Pitfall: Pre-check queries start implicit transactions:

# API endpoint
@router.post("/items")
async def create_item(db: DbSession, ...):
    # This query starts an implicit transaction!
    existing = await db.execute(select(Item).where(...))

    # Service sees in_transaction()=True, uses begin_nested()
    item = await service.create_item(...)

    # BUG: Without this, changes are never committed!
    await db.commit()  # ← REQUIRED when API queried before service call

    return item

Rules:

1. If API does ANY query before calling a write service → must await db.commit() after
2. If API only calls service (no pre-queries) → service handles commit via begin()
3. Service-to-service calls → outer service owns commit, inner uses savepoint

Dependency Injection (FastAPI)

from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import async_session_maker

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Annotated[AsyncSession, Depends(get_db)],
) -> User:
    """Decode JWT and return current user."""
    payload = decode_token(token)
    user = await db.execute(
        select(User).where(User.tone3000_id == payload["sub"])
    )
    user = user.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

Background Tasks (TaskIQ)

from taskiq import TaskiqScheduler
from taskiq_redis import ListQueueBroker, RedisAsyncResultBackend

broker = ListQueueBroker(url="redis://redis:6379")
broker.with_result_backend(RedisAsyncResultBackend(redis_url="redis://redis:6379"))

@broker.task
async def process_shootout(job_id: str, shootout_id: int) -> str:
    """Process a shootout through the pipeline."""
    await update_job_status(job_id, "running", progress=0)

    # Process...
    await update_job_status(job_id, "running", progress=50)

    await update_job_status(job_id, "completed", progress=100)
    return output_path

Migrations (Alembic)

# Create migration
docker compose exec webapp alembic revision --autogenerate -m "add jobs table"

# Apply migrations
docker compose exec webapp alembic upgrade head

# Rollback
docker compose exec webapp alembic downgrade -1

Code Quality

Commands

# Lint (check only)
docker compose exec webapp ruff check app/

# Lint (auto-fix)
docker compose exec webapp ruff check --fix app/

# Type check
docker compose exec webapp mypy app/

# Test
docker compose exec webapp pytest tests/

# All checks
just check-backend

You Must Get Right (can’t auto-fix)

Related Skills

Project-Specific

• gts-backend-dev – Full FastAPI/SQLAlchemy documentation
• gts-frontend-dev – Astro frontend patterns
• gts-testing – Testing patterns specific to GTS

Global (Architecture Concepts)

• service-patterns – Service layer concepts
• repository-patterns – Repository abstraction concepts
• error-handling – Error handling concepts
• web-handlers – HTTP handler concepts