Backend Engineering

Default Stack (Ask First)

Before applying anything, ask:

"Can I use the Supernova backend stack? Python 3.12 + FastAPI + SQLAlchemy 2.0 async + PostgreSQL. Or do you have an existing backend stack I should match?"

If an existing requirements.txt, pyproject.toml, or setup.py is detected, read it and match the existing stack.

Progressive Disclosure

• Load references/fastapi-patterns.md when setting up dependency injection, middleware, or lifespan events.
• Load references/sqlalchemy-async.md when implementing async DB sessions or complex ORM queries.

SOP: Backend Implementation

Step 1 - Project Scaffold (New Projects Only)

For greenfield projects, structure the project as follows. Do not deviate:

src/
├── main.py              # FastAPI app instance, lifespan, router inclusion
├── config.py            # pydantic-settings BaseSettings
├── database.py          # Async engine + session factory
├── models/              # SQLAlchemy ORM models (one file per domain entity)
├── schemas/             # Pydantic request/response models
├── repositories/        # Data access layer (all DB queries live here)
├── services/            # Business logic (calls repositories, never raw DB)
├── api/
│   └── v1/              # FastAPI routers (thin - delegate to services)
├── middleware/          # Custom ASGI middleware
└── tests/               # pytest-asyncio test files mirroring src/ structure

pyproject.toml is the only dependency file. Use uv as the package manager.

Step 2 - Dependency Management

# pyproject.toml
[project]
requires-python = ">=3.12"
dependencies = [
    "fastapi>=0.111.0",
    "uvicorn[standard]>=0.29.0",
    "sqlalchemy[asyncio]>=2.0.0",
    "asyncpg>=0.29.0",
    "pydantic>=2.7.0",
    "pydantic-settings>=2.2.0",
    "alembic>=1.13.0",
    "structlog>=24.1.0",
]

[project.optional-dependencies]
dev = ["pytest>=8.0", "pytest-asyncio>=0.23", "httpx>=0.27", "ruff>=0.4"]

Install with: uv pip install -e ".[dev]"

Step 3 - Application Layering (Strict Separation)

The request lifecycle must flow exactly as follows. Do not skip layers:

HTTP Request
  -> FastAPI Router (validates request schema via Pydantic)
  -> Service Layer (applies business logic, raises domain exceptions)
  -> Repository Layer (executes SQLAlchemy queries)
  -> Database

Router rules: Routers only receive validated Pydantic inputs and return Pydantic outputs. No SQL, no business logic.

Service rules: Services call repositories and other services. They raise HTTPException or custom domain exceptions. Never return raw ORM models - always map to schemas.

Repository rules: Repositories are the only place that imports AsyncSession. Every method is async def. Return ORM model instances, not dicts.

Step 4 - Pydantic v2 Model Patterns

from pydantic import BaseModel, EmailStr, ConfigDict
from uuid import UUID
from datetime import datetime

class UserBase(BaseModel):
    email: EmailStr
    display_name: str

class UserCreate(UserBase):
    password: str  # plain text in; hashed in service

class UserOut(UserBase):
    model_config = ConfigDict(from_attributes=True)
    id: UUID
    created_at: datetime

Always define three Pydantic models per entity: Base (shared fields), Create (input), Out (response). Never expose ORM models directly.

Step 5 - Async SQLAlchemy Session Pattern

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

engine = create_async_engine(settings.DATABASE_URL, echo=settings.DEBUG)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

Inject with db: AsyncSession = Depends(get_db) in services via the router. Never create sessions manually outside of this factory.

Step 6 - Structured Logging

import structlog

log = structlog.get_logger()

async def create_user(user_in: UserCreate, db: AsyncSession) -> UserOut:
    log.info("creating_user", email=user_in.email)
    # ...
    log.info("user_created", user_id=str(user.id))
    return UserOut.model_validate(user)

Log with structured key=value pairs, never f-strings in log calls. Include user_id, request_id, and relevant entity IDs in every log line.

Step 7 - Error Handling

# In service layer
from fastapi import HTTPException

async def get_user_by_id(user_id: UUID, db: AsyncSession) -> UserOut:
    user = await user_repo.get_by_id(db, user_id)
    if not user:
        raise HTTPException(status_code=404, detail=f"User {user_id} not found")
    return UserOut.model_validate(user)

Define a global exception handler in main.py for unhandled exceptions: log the full traceback with structlog, return a generic 500 response. Never expose stack traces to API consumers.

Step 8 - Environment Configuration

# config.py
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
    DATABASE_URL: str
    SECRET_KEY: str
    DEBUG: bool = False
    ENVIRONMENT: str = "development"

settings = Settings()

.env is git-ignored. Provide .env.example with all keys and no values.

Step 9 - Testing Pattern

# tests/test_user_service.py
import pytest
from httpx import AsyncClient, ASGITransport
from src.main import app

@pytest.mark.asyncio
async def test_create_user():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.post("/api/v1/users", json={"email": "test@test.com", ...})
    assert response.status_code == 201
    assert "id" in response.json()["data"]

Test via the HTTP layer, not by calling services directly. This ensures routers, validation, and middleware are also tested.