Backend Architecture

Learn about Starterbase's backend architecture, the 3-file module pattern, and how everything fits together.

Overview

The backend is built with FastAPI and follows a clean, modular architecture that separates concerns and makes the codebase easy to navigate and extend.

Tech Stack

Technology	Purpose
FastAPI	Modern Python web framework with automatic OpenAPI docs
SQLAlchemy 2.0	Async ORM with `Mapped` type annotations
Pydantic	Data validation and serialization
Alembic	Database migrations
asyncpg	Async PostgreSQL driver
bcrypt	Password hashing
PyJWT	JWT token handling
structlog	Structured logging
pytest	Testing framework

Project Structure

backend/
├── app/
│   ├── api/
│   │   └── v1/              # API version 1
│   │       ├── router.py    # Main router aggregator
│   │       ├── auth/        # Authentication module
│   │       ├── billing/     # Stripe billing module
│   │       ├── organizations/ # Organization management
│   │       ├── settings/    # User settings
│   │       ├── health/      # Health check endpoint
│   │       ├── mcp/         # MCP server module
│   │       └── admin/       # Admin panel module
│   │           ├── analytics/
│   │           ├── users/
│   │           ├── roles/
│   │           ├── organizations/
│   │           ├── billing/
│   │           ├── subscriptions/
│   │           ├── products/
│   │           ├── invitations/
│   │           └── settings/
│   ├── controllers/         # Shared business logic
│   │   ├── auth.py          # JWT, password hashing, token blacklist
│   │   ├── email.py         # Email sending (Resend)
│   │   ├── tokens.py        # Token utilities
│   │   └── totp.py          # 2FA TOTP operations
│   ├── core/                # Core functionality
│   │   ├── config.py        # Configuration (Pydantic Settings)
│   │   ├── database.py      # Database connection and session
│   │   ├── exceptions.py    # Custom HTTP exceptions
│   │   ├── middleware.py    # Correlation ID, request logging
│   │   ├── logging.py       # Structured logging (structlog)
│   │   ├── redis_client.py  # Redis connection
│   │   ├── pagination.py    # Pagination utilities
│   │   ├── filtering.py     # Filter parsing utilities
│   │   ├── datetime.py      # Timezone-aware datetime helpers
│   │   └── responses.py     # Standard API responses
│   ├── models/              # SQLAlchemy models
│   │   ├── base.py          # Base model class
│   │   ├── user.py          # User model
│   │   ├── role.py          # Role model
│   │   ├── organization.py  # Organization model
│   │   ├── subscription.py  # Subscription model
│   │   ├── plan.py          # Billing plan model
│   │   ├── product.py       # One-time product model
│   │   ├── purchase.py      # Purchase record model
│   │   ├── invitation.py    # User invitation model
│   │   ├── password_reset.py # Password reset token model
│   │   ├── trusted_device.py # 2FA trusted device model
│   │   └── enums/           # Enum types
│   └── main.py              # FastAPI application entry point
├── alembic/                 # Database migrations
│   ├── versions/            # Migration files
│   └── env.py               # Alembic configuration
├── tests/                   # Test suite
├── pyproject.toml           # Python dependencies (uv)
└── .env                     # Environment variables

The 3-File Module Pattern

Every API module follows a consistent 3-file pattern:

app/api/v1/<module>/
├── router.py       # FastAPI routes (HTTP layer)
├── deps.py         # Dependencies + business logic
└── schemas.py      # Pydantic models (validation)

File Responsibilities

`schemas.py` - Data Validation

Defines Pydantic models for request/response validation:

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

class UserCreate(BaseModel):
    """Request schema for creating a user."""
    email: EmailStr
    password: str
    first_name: str | None = None

class UserResponse(BaseModel):
    """Response schema for user data."""
    id: UUID
    email: str
    first_name: str | None
    created_at: datetime

    model_config = {"from_attributes": True}

Guidelines:

Use descriptive names: UserCreate, UserUpdate, UserResponse
Use from_attributes = True for ORM model conversion
Leverage Pydantic validators (EmailStr, HttpUrl, etc.)

`deps.py` - Business Logic

Contains business logic with direct database access:

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

from app.core.exceptions import BadRequestException, NotFoundException
from app.models.user import User
from .schemas import UserCreate

async def create_user(db: AsyncSession, data: UserCreate) -> User:
    """Create a new user."""
    # Check if email already exists
    stmt = select(User).where(User.email == data.email)
    result = await db.execute(stmt)
    if result.scalar_one_or_none():
        raise BadRequestException("Email already registered")

    # Create user
    user = User(
        email=data.email,
        password_hash=hash_password(data.password),
        first_name=data.first_name,
    )
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

async def get_user_by_id(db: AsyncSession, user_id: UUID) -> User | None:
    """Get user by ID."""
    return await db.get(User, user_id)

Guidelines:

Write standalone async functions (no classes)
Use direct SQLAlchemy queries (no repository layer)
Use custom exceptions from app.core.exceptions
Keep functions focused and single-purpose

`router.py` - HTTP Layer

Defines FastAPI routes and delegates to deps functions:

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

from app.core.database import get_db
from app.core.exceptions import NotFoundException
from . import deps
from .schemas import UserCreate, UserResponse

router = APIRouter()

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    data: UserCreate,
    db: Annotated[AsyncSession, Depends(get_db)],
):
    """Create a new user."""
    return await deps.create_user(db, data)

@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: UUID,
    db: Annotated[AsyncSession, Depends(get_db)],
):
    """Get user by ID."""
    user = await deps.get_user_by_id(db, user_id)
    if not user:
        raise NotFoundException("User not found")
    return user

Guidelines:

Handle HTTP concerns only (routing, status codes)
Delegate all business logic to deps functions
Use Annotated with Depends for dependencies
Specify response_model for automatic validation

Why This Pattern?

Clear separation - HTTP, logic, and validation are isolated
Easy testing - Test deps functions without HTTP overhead
Reusable logic - Use deps functions from multiple routes
Type safety - Full type hints with Pydantic and SQLAlchemy
Maintainable - Predictable structure makes navigation easy

Shared Logic: Controllers

For logic used across multiple modules, use the controllers/ directory:

# app/controllers/auth.py
import bcrypt
import jwt
from uuid import UUID, uuid4
from datetime import timedelta

from app.core.config import settings
from app.core.datetime import now

def hash_password(password: str) -> str:
    """Hash a password using bcrypt."""
    salt = bcrypt.gensalt()
    return bcrypt.hashpw(password.encode(), salt).decode()

def verify_password(password: str, hashed: str) -> bool:
    """Verify a password against its hash."""
    return bcrypt.checkpw(password.encode(), hashed.encode())

def create_access_token(user_id: UUID, email: str) -> str:
    """Create a JWT access token."""
    expire = now() + timedelta(minutes=settings.jwt_access_token_expire_minutes)
    payload = {
        "sub": str(user_id),
        "email": email,
        "exp": expire,
        "iat": now(),
        "jti": str(uuid4()),
        "type": "access",
    }
    return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)

Use controllers for:

Authentication utilities (JWT, password hashing)
Email sending
Token management
TOTP/2FA operations
Cross-cutting concerns

Database Operations

Async SQLAlchemy 2.0

All database operations use async/await with modern Mapped syntax:

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

async def get_users(db: AsyncSession, skip: int = 0, limit: int = 10):
    """Get paginated users."""
    stmt = select(User).offset(skip).limit(limit)
    result = await db.execute(stmt)
    return result.scalars().all()

Session Management

Database sessions are managed via dependency injection:

from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db

@router.get("/users")
async def list_users(
    db: Annotated[AsyncSession, Depends(get_db)],
):
    # db session is automatically created and closed
    return await get_users(db)

Model Definitions

Models use SQLAlchemy 2.0 Mapped annotations:

from datetime import datetime
from uuid import UUID, uuid4
from sqlalchemy import String, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.models.base import Base
from app.core.datetime import now

class User(Base):
    __tablename__ = "users"

    id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    first_name: Mapped[str | None] = mapped_column(String(100), nullable=True)
    role_id: Mapped[UUID | None] = mapped_column(ForeignKey("roles.id"), nullable=True)
    created_at: Mapped[datetime] = mapped_column(default=now)

    # Relationships
    role: Mapped["Role | None"] = relationship("Role", lazy="joined")

Loading Relationships

Use selectinload for eager loading:

from sqlalchemy.orm import selectinload

stmt = select(User).options(selectinload(User.role)).where(User.id == user_id)
result = await db.execute(stmt)
user = result.scalar_one_or_none()

Configuration

Configuration is managed via Pydantic Settings with lowercase variable names:

# app/core/config.py
from pydantic import ConfigDict
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    model_config = ConfigDict(extra="ignore", env_file=".env")

    # Application
    app_name: str = "Starterbase"
    environment: str = "development"
    debug: bool = True
    frontend_url: str = "http://localhost:3000"

    # Database
    db_url: str = "postgresql+asyncpg://user:pass@localhost/db"

    # Redis
    redis_url: str = "redis://localhost:6379"

    # JWT
    jwt_secret: str = "change-this-in-production"
    jwt_algorithm: str = "HS256"
    jwt_access_token_expire_minutes: int = 15
    jwt_refresh_token_expire_days: int = 7

    # Stripe
    stripe_secret_key: str = ""
    stripe_publishable_key: str = ""

    @property
    def stripe_is_configured(self) -> bool:
        return bool(self.stripe_secret_key and self.stripe_publishable_key)

settings = Settings()

Access settings anywhere:

from app.core.config import settings

print(settings.db_url)
if settings.stripe_is_configured:
    # Use Stripe API

Middleware

CorrelationIdMiddleware

Adds X-Correlation-ID header to all requests for distributed tracing:

class CorrelationIdMiddleware:
    """Adds correlation IDs to requests for tracing."""

    HEADER_NAME = "X-Correlation-ID"

    async def __call__(self, scope, receive, send):
        # Extract or generate correlation ID
        correlation_id = extract_header(scope) or str(uuid.uuid4())

        # Bind to structlog context
        structlog.contextvars.bind_contextvars(correlation_id=correlation_id)

        # Add to response headers
        async def send_with_id(message):
            if message["type"] == "http.response.start":
                headers = list(message.get("headers", []))
                headers.append((self.HEADER_NAME.encode(), correlation_id.encode()))
                message["headers"] = headers
            await send(message)

        await self.app(scope, receive, send_with_id)

RequestLoggingMiddleware

Logs all requests/responses with timing:

class RequestLoggingMiddleware(BaseHTTPMiddleware):
    """Logs request/response details with timing."""

    async def dispatch(self, request, call_next):
        start_time = time.perf_counter_ns()
        response = await call_next(request)
        duration_ms = (time.perf_counter_ns() - start_time) / 1_000_000

        self.logger.info(
            f"{request.method} {request.url.path} {response.status_code}",
            status_code=response.status_code,
            duration_ms=round(duration_ms, 2),
        )

        response.headers["X-Process-Time-Ms"] = str(round(duration_ms, 2))
        return response

Exception Handling

Custom exceptions extend HTTPException for automatic handling:

# app/core/exceptions.py
from fastapi import HTTPException, status

class BadRequestException(HTTPException):
    def __init__(self, detail: str = "Bad request"):
        super().__init__(status_code=status.HTTP_400_BAD_REQUEST, detail=detail)

class UnauthorizedException(HTTPException):
    def __init__(self, detail: str = "Unauthorized"):
        super().__init__(status_code=status.HTTP_401_UNAUTHORIZED, detail=detail)

class ForbiddenException(HTTPException):
    def __init__(self, detail: str = "Forbidden"):
        super().__init__(status_code=status.HTTP_403_FORBIDDEN, detail=detail)

class NotFoundException(HTTPException):
    def __init__(self, detail: str = "Not found"):
        super().__init__(status_code=status.HTTP_404_NOT_FOUND, detail=detail)

class ConflictException(HTTPException):
    def __init__(self, detail: str = "Conflict"):
        super().__init__(status_code=status.HTTP_409_CONFLICT, detail=detail)

Usage:

from app.core.exceptions import NotFoundException, BadRequestException

async def get_user(db: AsyncSession, user_id: UUID) -> User:
    user = await db.get(User, user_id)
    if not user:
        raise NotFoundException("User not found")
    return user

No custom exception handlers needed - FastAPI handles HTTPException automatically.

Application Startup

The application uses FastAPI's lifespan for startup/shutdown:

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    logger.info("Application starting")
    if settings.debug:
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        await create_default_roles()

    await get_redis()  # Initialize Redis

    yield

    # Shutdown
    await close_redis()
    await engine.dispose()

app = FastAPI(title=settings.app_name, lifespan=lifespan)

# Middleware (order matters)
app.add_middleware(CORSMiddleware, ...)
app.add_middleware(RequestLoggingMiddleware)
app.add_middleware(CorrelationIdMiddleware)

app.include_router(api_router, prefix="/api/v1")

Next Steps

Authentication - Learn about the JWT auth system
Billing - Stripe integration
Landing Components - Build your landing pages

Get Started

Core Guides

Landing Components

UI Components

Backend Architecture

Overview

Tech Stack

Project Structure

The 3-File Module Pattern

File Responsibilities

`schemas.py` - Data Validation

`deps.py` - Business Logic

`router.py` - HTTP Layer

Why This Pattern?

Shared Logic: Controllers

Database Operations

Async SQLAlchemy 2.0

Session Management

Model Definitions

Loading Relationships

Configuration

Middleware

CorrelationIdMiddleware

RequestLoggingMiddleware

Exception Handling

Application Startup

Next Steps

Get Started

Core Guides

Landing Components

UI Components

Backend Architecture

Overview

Tech Stack

Project Structure

The 3-File Module Pattern

File Responsibilities

schemas.py - Data Validation

deps.py - Business Logic

router.py - HTTP Layer

Why This Pattern?

Shared Logic: Controllers

Database Operations

Async SQLAlchemy 2.0

Session Management

Model Definitions

Loading Relationships

Configuration

Middleware

CorrelationIdMiddleware

RequestLoggingMiddleware

Exception Handling

Application Startup

Next Steps

`schemas.py` - Data Validation

`deps.py` - Business Logic

`router.py` - HTTP Layer