Backend Development Skill

Activation: FastAPI, SQLAlchemy, Pydantic, async Python, database operations

Architecture Foundation:

•Concepts: See global software-architecture skill for DDD, CQRS, Hexagonal Architecture
•Python Syntax: See python-cheatsheet skill for quick reference
•GTS Implementation: This document (below)

CRITICAL: Container-First Execution

NEVER run Python commands directly on the host. Always use Docker:
bash
# WRONG - will fail
pytest tests/
ruff check app/

# RIGHT - use Docker
docker compose exec backend pytest tests/
docker compose exec backend ruff check app/
See .claude/rules/container-execution.md for full details.

Stack Overview

Component	Technology	Version Policy
Framework	FastAPI	Latest stable
ORM	SQLAlchemy 2.0	Async with mapped_column
Validation	Pydantic v2	BaseModel schemas
Migrations	Alembic	Async driver
Auth	OAuth2 + JWT	Tone 3000 integration
Queue	TaskIQ	Redis backend
Database	PostgreSQL	Async via asyncpg

Architecture Pattern

code

app/
├── api/           # Route handlers (thin layer)
│   ├── auth.py    # OAuth endpoints
│   ├── shootouts.py
│   ├── jobs.py
│   └── ws.py      # WebSocket handlers
├── core/          # Configuration & shared
│   ├── config.py  # Settings from env
│   ├── security.py
│   ├── database.py
│   └── tone3000.py # API client
├── models/        # SQLAlchemy models
├── schemas/       # Pydantic schemas
├── services/      # Business logic
└── tasks/         # Background jobs

Key Patterns

1. Route Handler (Thin Controller)

python

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

2. SQLAlchemy 2.0 Models

python

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")

3. Pydantic Schemas

python

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

4. Dependency Injection

python

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

5. Background Tasks (TaskIQ)

python

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

Database Operations

Queries

python

from sqlalchemy import select
from sqlalchemy.orm import selectinload

# Get with eager loading
stmt = (
    select(Shootout)
    .where(Shootout.user_id == user_id)
    .options(selectinload(Shootout.jobs))
    .order_by(Shootout.created_at.desc())
)
result = await db.execute(stmt)
shootouts = result.scalars().all()

# Pagination
stmt = (
    select(Shootout)
    .offset(skip)
    .limit(limit)
)

Transactions

Service Layer Pattern: Services use begin_nested() when already in a transaction, otherwise begin():

python

# 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:

python

# 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:

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

Migrations (Alembic)

bash

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

# Apply migrations
docker compose exec backend alembic upgrade head

# Rollback
docker compose exec backend alembic downgrade -1

Error Handling

python

from fastapi import HTTPException, status

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

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

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

# In routes
@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

Code Quality

Tooling Handles (auto-fixed on commit)

Pre-commit auto-fixes formatting, imports, and simple refactors via ruff --fix.

Enabled ruff rules: E, W (PEP 8), F (Pyflakes), I (isort), B (bugbear), UP (pyupgrade), ASYNC, N (naming), C4 (comprehensions), SIM (simplify), RUF, PT (pytest), PERF.

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

Requirement	Enforced By	Notes
Type hints on all functions	mypy strict	Can't add them for you
snake_case functions/variables	ruff N	Can detect but not rename semantically
PascalCase for classes	ruff N	Can detect but not rename semantically
pytest conventions	ruff PT	`test_` prefix, fixtures via conftest.py
ClassVar for mutable class attrs	ruff RUF012	`ClassVar[dict[...]]` for class-level dicts

Commands

bash

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

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

# Type check
docker compose exec backend mypy app/

# Test
docker compose exec backend pytest tests/

# All checks
just check-backend

Resources

See resources/ for detailed guides:

•tone3000-integration.md - OAuth and API patterns
•taskiq-patterns.md - Background job patterns
•websocket-progress.md - Real-time updates

Related Skills

Global (Architecture Concepts)

•software-architecture - DDD, CQRS, Hexagonal Architecture
•service-patterns - Service layer patterns
•repository-patterns - Repository abstraction
•error-handling - Error handling strategies
•web-handlers - HTTP handler patterns

Project (Quick Reference)

•python-cheatsheet - Global concepts → Python syntax mapping

Project (GTS-Specific)

•gts-testing - Testing patterns
•gts-frontend-dev - Astro frontend integration
•docker-infra - Container management