Backend Architecture¶

This document describes the backend architecture patterns used in generated projects.

Layered Architecture¶

The backend follows a clean layered architecture with clear separation of concerns:

Text Only

┌─────────────────────────────────────────────────────────────┐
│                      API Layer (Routes)                      │
│  HTTP endpoints, request validation, response serialization │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                     Service Layer                            │
│      Business logic, orchestration, error handling           │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                   Repository Layer                           │
│          Data access, database queries, CRUD                 │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                      Database                                │
│            PostgreSQL / MongoDB / SQLite                     │
└─────────────────────────────────────────────────────────────┘

Directory Structure¶

Text Only

app/
├── api/                    # API Layer
│   ├── routes/
│   │   └── v1/
│   │       ├── __init__.py
│   │       ├── health.py
│   │       ├── auth.py
│   │       ├── users.py
│   │       └── items.py
│   ├── deps.py             # Dependency injection
│   ├── router.py           # Route aggregation
│   └── exception_handlers.py
│
├── services/               # Service Layer
│   ├── __init__.py
│   ├── user.py
│   └── item.py
│
├── repositories/           # Repository Layer
│   ├── __init__.py
│   ├── base.py             # Generic CRUD operations
│   ├── user.py
│   └── item.py
│
├── schemas/                # Pydantic Models
│   ├── __init__.py
│   ├── base.py
│   ├── user.py
│   └── item.py
│
├── db/                     # Database
│   ├── models/
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── base.py
│   └── session.py
│
└── core/                   # Core Configuration
    ├── config.py
    ├── security.py
    └── exceptions.py

Layer Responsibilities¶

API Layer (`app/api/routes/`)¶

The API layer handles HTTP concerns:

Request validation - Pydantic schemas validate incoming data
Authentication - Dependencies verify JWT/API keys
Response serialization - Format data for clients
Error responses - HTTP status codes and error messages

Python

# app/api/routes/v1/items.py
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession

from app.api.deps import get_db, get_current_user
from app.schemas.item import ItemCreate, ItemResponse
from app.services.item import ItemService

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


@router.post("/", response_model=ItemResponse)
async def create_item(
    item_in: ItemCreate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Create a new item."""
    service = ItemService(db)
    item = await service.create(item_in)
    return item

Key principles: - Routes are thin - delegate business logic to services - Use dependency injection for database sessions and auth - Keep request/response schemas separate from internal models

Service Layer (`app/services/`)¶

Services contain business logic and orchestration:

Business rules - Validation beyond schema constraints
Error handling - Domain-specific exceptions
Orchestration - Coordinate multiple repositories
External services - API calls, email, etc.

Python

# app/services/item.py
from uuid import UUID

from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import NotFoundError
from app.db.models.item import Item
from app.repositories import item_repo
from app.schemas.item import ItemCreate, ItemUpdate


class ItemService:
    """Service for item-related business logic."""

    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_by_id(self, item_id: UUID) -> Item:
        """Get item by ID."""
        item = await item_repo.get_by_id(self.db, item_id)
        if not item:
            raise NotFoundError(
                message="Item not found",
                details={"item_id": str(item_id)},
            )
        return item

    async def create(self, item_in: ItemCreate) -> Item:
        """Create a new item."""
        # Business validation could go here
        return await item_repo.create(
            self.db,
            title=item_in.title,
            description=item_in.description,
        )

    async def update(self, item_id: UUID, item_in: ItemUpdate) -> Item:
        """Update an item."""
        item = await self.get_by_id(item_id)
        update_data = item_in.model_dump(exclude_unset=True)
        return await item_repo.update(self.db, db_item=item, update_data=update_data)

Key principles: - Services are stateless (except for db session) - Raise domain exceptions, not HTTP exceptions - Services can call other services for complex operations

Repository Layer (`app/repositories/`)¶

Repositories handle data access:

CRUD operations - Create, Read, Update, Delete
Query building - Complex database queries
Data mapping - ORM model interactions

Python

# app/repositories/item.py
from uuid import UUID

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

from app.db.models.item import Item


class ItemRepository:
    """Repository for Item database operations."""

    async def get_by_id(self, db: AsyncSession, item_id: UUID) -> Item | None:
        """Get item by ID."""
        return await db.get(Item, item_id)

    async def get_multi(
        self,
        db: AsyncSession,
        *,
        skip: int = 0,
        limit: int = 100,
        active_only: bool = False,
    ) -> list[Item]:
        """Get multiple items with pagination."""
        query = select(Item)
        if active_only:
            query = query.where(Item.is_active == True)
        query = query.offset(skip).limit(limit)
        result = await db.execute(query)
        return list(result.scalars().all())

    async def create(
        self,
        db: AsyncSession,
        *,
        title: str,
        description: str | None = None,
    ) -> Item:
        """Create a new item."""
        item = Item(title=title, description=description)
        db.add(item)
        await db.flush()
        await db.refresh(item)
        return item


# Singleton instance
item_repo = ItemRepository()

Key principles: - Repositories are model-specific - Use db.flush() instead of db.commit() - let the caller manage transactions - Return ORM models, not dicts

Base Repository¶

The template provides a generic base repository for common CRUD operations:

Python

# app/repositories/base.py
from typing import Any, Generic, TypeVar

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

from app.db.base import Base

ModelType = TypeVar("ModelType", bound=Base)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)


class BaseRepository(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
    """Base class for repository operations."""

    def __init__(self, model: type[ModelType]):
        self.model = model

    async def get(self, db: AsyncSession, id: Any) -> ModelType | None:
        """Get a single record by ID."""
        return await db.get(self.model, id)

    async def get_multi(
        self,
        db: AsyncSession,
        *,
        skip: int = 0,
        limit: int = 100,
    ) -> list[ModelType]:
        """Get multiple records with pagination."""
        result = await db.execute(
            select(self.model).offset(skip).limit(limit)
        )
        return list(result.scalars().all())

    async def create(
        self,
        db: AsyncSession,
        *,
        obj_in: CreateSchemaType,
    ) -> ModelType:
        """Create a new record."""
        obj_in_data = obj_in.model_dump()
        db_obj = self.model(**obj_in_data)
        db.add(db_obj)
        await db.flush()
        await db.refresh(db_obj)
        return db_obj

Schemas (Pydantic Models)¶

Schemas define request/response structures:

Python

# app/schemas/item.py
from uuid import UUID
from datetime import datetime

from pydantic import BaseModel, Field


class ItemBase(BaseModel):
    """Shared properties."""
    title: str = Field(..., min_length=1, max_length=100)
    description: str | None = None


class ItemCreate(ItemBase):
    """Properties to receive on creation."""
    pass


class ItemUpdate(BaseModel):
    """Properties to receive on update."""
    title: str | None = Field(None, min_length=1, max_length=100)
    description: str | None = None


class ItemResponse(ItemBase):
    """Properties to return to client."""
    id: UUID
    is_active: bool
    created_at: datetime
    updated_at: datetime

    model_config = {"from_attributes": True}

Key principles: - Separate schemas for Create, Update, and Response - Use from_attributes = True for ORM model conversion - Apply validation constraints (min_length, max_length, etc.)

Database Models¶

SQLAlchemy models define the database schema:

Python

# app/db/models/item.py
from uuid import uuid4
from datetime import datetime

from sqlalchemy import String, Text, Boolean, DateTime
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy.sql import func

from app.db.base import Base


class Item(Base):
    """Item model."""

    __tablename__ = "items"

    id: Mapped[UUID] = mapped_column(
        UUID(as_uuid=True),
        primary_key=True,
        default=uuid4,
    )
    title: Mapped[str] = mapped_column(String(100), nullable=False)
    description: Mapped[str | None] = mapped_column(Text, nullable=True)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=func.now(),
    )

Dependency Injection¶

FastAPI dependencies provide database sessions and authentication:

Python

# app/api/deps.py
from collections.abc import AsyncGenerator
from typing import Annotated

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.security import verify_token
from app.db.session import async_session_maker
from app.services.user import UserService

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """Get database session."""
    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)],
):
    """Get current authenticated user."""
    payload = verify_token(token)
    if payload is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
        )

    user_service = UserService(db)
    user = await user_service.get_by_id(payload["sub"])
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="User not found",
        )
    return user


# Type aliases for cleaner route signatures
DB = Annotated[AsyncSession, Depends(get_db)]
CurrentUser = Annotated["User", Depends(get_current_user)]

Usage in routes:

Python

@router.get("/me")
async def get_me(current_user: CurrentUser):
    return current_user


@router.get("/items")
async def list_items(db: DB):
    service = ItemService(db)
    return await service.get_multi()

Exception Handling¶

Custom exceptions provide consistent error responses:

Python

# app/core/exceptions.py
from typing import Any


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

    def __init__(
        self,
        message: str,
        status_code: int = 500,
        details: dict[str, Any] | None = None,
    ):
        self.message = message
        self.status_code = status_code
        self.details = details or {}
        super().__init__(self.message)


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

    def __init__(self, message: str = "Not found", details: dict | None = None):
        super().__init__(message, status_code=404, details=details)


class AlreadyExistsError(AppError):
    """Resource already exists."""

    def __init__(self, message: str = "Already exists", details: dict | None = None):
        super().__init__(message, status_code=409, details=details)


class ValidationError(AppError):
    """Validation failed."""

    def __init__(self, message: str = "Validation error", details: dict | None = None):
        super().__init__(message, status_code=422, details=details)


class UnauthorizedError(AppError):
    """Authentication required."""

    def __init__(self, message: str = "Unauthorized", details: dict | None = None):
        super().__init__(message, status_code=401, details=details)


class ForbiddenError(AppError):
    """Permission denied."""

    def __init__(self, message: str = "Forbidden", details: dict | None = None):
        super().__init__(message, status_code=403, details=details)

Exception handlers convert these to HTTP responses:

Python

# app/api/exception_handlers.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

from app.core.exceptions import AppError


def register_exception_handlers(app: FastAPI) -> None:
    """Register custom exception handlers."""

    @app.exception_handler(AppError)
    async def app_error_handler(request: Request, exc: AppError):
        return JSONResponse(
            status_code=exc.status_code,
            content={
                "error": exc.message,
                "details": exc.details,
            },
        )

Testing¶

The layered architecture makes testing straightforward:

Python

# tests/test_services.py
import pytest
from unittest.mock import AsyncMock, patch

from app.services.item import ItemService
from app.schemas.item import ItemCreate


@pytest.mark.asyncio
async def test_create_item():
    """Test item creation."""
    mock_db = AsyncMock()

    with patch("app.services.item.item_repo") as mock_repo:
        mock_repo.create.return_value = Item(
            id="123",
            title="Test",
            description="Test item",
        )

        service = ItemService(mock_db)
        item = await service.create(ItemCreate(title="Test", description="Test item"))

        assert item.title == "Test"
        mock_repo.create.assert_called_once()

Best Practices¶

Keep routes thin - Delegate to services
Services handle business logic - Validation, orchestration
Repositories handle data - Queries, CRUD
Use dependency injection - For testability
Raise domain exceptions - Not HTTP exceptions in services
Use transactions appropriately - flush() in repos, commit() in deps
Separate schemas - Create, Update, Response
Type everything - Pydantic + Python type hints

Backend Architecture¶

Layered Architecture¶

Directory Structure¶

Layer Responsibilities¶

API Layer (app/api/routes/)¶

Service Layer (app/services/)¶

Repository Layer (app/repositories/)¶

Base Repository¶

Schemas (Pydantic Models)¶

Database Models¶

Dependency Injection¶

Exception Handling¶

Testing¶

Best Practices¶

API Layer (`app/api/routes/`)¶

Service Layer (`app/services/`)¶

Repository Layer (`app/repositories/`)¶