Backend FastAPI

Complete guide for FastAPI backend development, from framework fundamentals to advanced architectural patterns.

When to Use This Skill

• Implementing FastAPI endpoints and API routes
• Setting up FastAPI projects with proper architecture
• Architecting complex FastAPI backend systems
• Applying Clean Architecture, Hexagonal Architecture, or Domain-Driven Design
• Refactoring FastAPI applications for better maintainability
• Implementing dependency injection and security patterns
• Structuring FastAPI services and business logic

MCP Usage Priority

1. context7 - Primary source for FastAPI documentation

   • /fastapi/fastapi

     (845 snippets, Trust: 9.9) - Official repo

   • /tiangolo/fastapi

     (11,584 snippets, Trust: 9.0) - Documentation site
2. web_search - For edge cases or recent updates

FastAPI Quick Reference

Core Concepts

• Async/Await: Use async def for all endpoints (except blocking operations)
• Dependency Injection: Shared logic via

  Depends()

• Pydantic Models: Request/response validation and serialization
• Router Organization: Group endpoints by domain/feature
• Path Operations: Explicit response_model and status codes

Database Integration

• SQLAlchemy: Async sessions with asyncio support
• Alembic: Migration management (auto-generate from models)
• Repository Pattern: Abstract data access layer
• Session Management: Request-scoped sessions via Depends()

Security

• OAuth2/JWT: Token-based authentication
• Dependencies: Security via Depends() chain
• Password Hashing: bcrypt or passlib

Architecture Patterns

Clean Architecture (Uncle Bob)

Clean Architecture provides clear separation of concerns with dependency flowing inward, making FastAPI applications testable, maintainable, and framework-independent at the core.

Layers (dependency flows inward):

• Entities: Core business models
• Use Cases: Application business rules
• Interface Adapters: Controllers, presenters, gateways
• Frameworks & Drivers: UI, database, external services

Key Principles:

• Dependencies point inward
• Inner layers know nothing about outer layers
• Business logic independent of frameworks
• Testable without UI, database, or external services

Implementation Example

# domain/entities/user.py
from dataclasses import dataclass
from datetime import datetime
from typing import Optional

@dataclass
class User:
    """Core user entity - no framework dependencies."""
    id: str
    email: str
    name: str
    created_at: datetime
    is_active: bool = True

    def deactivate(self):
        """Business rule: deactivating user."""
        self.is_active = False

    def can_place_order(self) -> bool:
        """Business rule: active users can order."""
        return self.is_active

# domain/interfaces/user_repository.py
from abc import ABC, abstractmethod
from typing import Optional, List
from domain.entities.user import User

class IUserRepository(ABC):
    """Port: defines contract, no implementation."""

    @abstractmethod
    async def find_by_id(self, user_id: str) -> Optional[User]:
        pass

    @abstractmethod
    async def find_by_email(self, email: str) -> Optional[User]:
        pass

    @abstractmethod
    async def save(self, user: User) -> User:
        pass

    @abstractmethod
    async def delete(self, user_id: str) -> bool:
        pass

# use_cases/create_user.py
from domain.entities.user import User
from domain.interfaces.user_repository import IUserRepository
from dataclasses import dataclass
from datetime import datetime
import uuid

@dataclass
class CreateUserRequest:
    email: str
    name: str

@dataclass
class CreateUserResponse:
    user: User
    success: bool
    error: Optional[str] = None

class CreateUserUseCase:
    """Use case: orchestrates business logic."""

    def __init__(self, user_repository: IUserRepository):
        self.user_repository = user_repository

    async def execute(self, request: CreateUserRequest) -> CreateUserResponse:
        # Business validation
        existing = await self.user_repository.find_by_email(request.email)
        if existing:
            return CreateUserResponse(
                user=None,
                success=False,
                error="Email already exists"
            )

        # Create entity
        user = User(
            id=str(uuid.uuid4()),
            email=request.email,
            name=request.name,
            created_at=datetime.now(),
            is_active=True
        )

        # Persist
        saved_user = await self.user_repository.save(user)

        return CreateUserResponse(
            user=saved_user,
            success=True
        )

# adapters/repositories/postgres_user_repository.py
from domain.interfaces.user_repository import IUserRepository
from domain.entities.user import User
from typing import Optional
import asyncpg

class PostgresUserRepository(IUserRepository):
    """Adapter: PostgreSQL implementation."""

    def __init__(self, pool: asyncpg.Pool):
        self.pool = pool

    async def find_by_id(self, user_id: str) -> Optional[User]:
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow(
                "SELECT * FROM users WHERE id = $1", user_id
            )
            return self._to_entity(row) if row else None

    async def find_by_email(self, email: str) -> Optional[User]:
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow(
                "SELECT * FROM users WHERE email = $1", email
            )
            return self._to_entity(row) if row else None

    async def save(self, user: User) -> User:
        async with self.pool.acquire() as conn:
            await conn.execute(
                """
                INSERT INTO users (id, email, name, created_at, is_active)
                VALUES ($1, $2, $3, $4, $5)
                ON CONFLICT (id) DO UPDATE
                SET email = $2, name = $3, is_active = $5
                """,
                user.id, user.email, user.name, user.created_at, user.is_active
            )
            return user

    async def delete(self, user_id: str) -> bool:
        async with self.pool.acquire() as conn:
            result = await conn.execute(
                "DELETE FROM users WHERE id = $1", user_id
            )
            return result == "DELETE 1"

    def _to_entity(self, row) -> User:
        """Map database row to entity."""
        return User(
            id=row["id"],
            email=row["email"],
            name=row["name"],
            created_at=row["created_at"],
            is_active=row["is_active"]
        )

# adapters/controllers/user_controller.py
from fastapi import APIRouter, Depends, HTTPException
from use_cases.create_user import CreateUserUseCase, CreateUserRequest
from pydantic import BaseModel

router = APIRouter()

class CreateUserDTO(BaseModel):
    email: str
    name: str

@router.post("/users")
async def create_user(
    dto: CreateUserDTO,
    use_case: CreateUserUseCase = Depends(get_create_user_use_case)
):
    """Controller: handles HTTP concerns only."""
    request = CreateUserRequest(email=dto.email, name=dto.name)
    response = await use_case.execute(request)

    if not response.success:
        raise HTTPException(status_code=400, detail=response.error)

    return {"user": response.user}

Hexagonal Architecture (Ports and Adapters)

Hexagonal Architecture isolates the business logic core from external concerns through ports (interfaces) and adapters (implementations), making it easy to swap implementations and test in isolation.

Components:

• Domain Core: Business logic
• Ports: Interfaces defining interactions
• Adapters: Implementations of ports (database, REST, message queue)

Benefits:

• Swap implementations easily (mock for testing)
• Technology-agnostic core
• Clear separation of concerns

Implementation Example

# Core domain (hexagon center)
class OrderService:
    """Domain service - no infrastructure dependencies."""

    def __init__(
        self,
        order_repository: OrderRepositoryPort,
        payment_gateway: PaymentGatewayPort,
        notification_service: NotificationPort
    ):
        self.orders = order_repository
        self.payments = payment_gateway
        self.notifications = notification_service

    async def place_order(self, order: Order) -> OrderResult:
        # Business logic
        if not order.is_valid():
            return OrderResult(success=False, error="Invalid order")

        # Use ports (interfaces)
        payment = await self.payments.charge(
            amount=order.total,
            customer=order.customer_id
        )

        if not payment.success:
            return OrderResult(success=False, error="Payment failed")

        order.mark_as_paid()
        saved_order = await self.orders.save(order)

        await self.notifications.send(
            to=order.customer_email,
            subject="Order confirmed",
            body=f"Order {order.id} confirmed"
        )

        return OrderResult(success=True, order=saved_order)

# Ports (interfaces)
class OrderRepositoryPort(ABC):
    @abstractmethod
    async def save(self, order: Order) -> Order:
        pass

class PaymentGatewayPort(ABC):
    @abstractmethod
    async def charge(self, amount: Money, customer: str) -> PaymentResult:
        pass

class NotificationPort(ABC):
    @abstractmethod
    async def send(self, to: str, subject: str, body: str):
        pass

# Adapters (implementations)
class StripePaymentAdapter(PaymentGatewayPort):
    """Primary adapter: connects to Stripe API."""

    def __init__(self, api_key: str):
        self.stripe = stripe
        self.stripe.api_key = api_key

    async def charge(self, amount: Money, customer: str) -> PaymentResult:
        try:
            charge = self.stripe.Charge.create(
                amount=amount.cents,
                currency=amount.currency,
                customer=customer
            )
            return PaymentResult(success=True, transaction_id=charge.id)
        except stripe.error.CardError as e:
            return PaymentResult(success=False, error=str(e))

class MockPaymentAdapter(PaymentGatewayPort):
    """Test adapter: no external dependencies."""

    async def charge(self, amount: Money, customer: str) -> PaymentResult:
        return PaymentResult(success=True, transaction_id="mock-123")

Domain-Driven Design (DDD)

Domain-Driven Design focuses on modeling the business domain accurately through strategic and tactical patterns, ensuring the code reflects the real-world business.

Strategic Patterns:

• Bounded Contexts: Separate models for different domains
• Context Mapping: How contexts relate
• Ubiquitous Language: Shared terminology

Tactical Patterns:

• Entities: Objects with identity
• Value Objects: Immutable objects defined by attributes
• Aggregates: Consistency boundaries
• Repositories: Data access abstraction
• Domain Events: Things that happened

Implementation Example

# Value Objects (immutable)
from dataclasses import dataclass
from typing import Optional

@dataclass(frozen=True)
class Email:
    """Value object: validated email."""
    value: str

    def __post_init__(self):
        if "@" not in self.value:
            raise ValueError("Invalid email")

@dataclass(frozen=True)
class Money:
    """Value object: amount with currency."""
    amount: int  # cents
    currency: str

    def add(self, other: "Money") -> "Money":
        if self.currency != other.currency:
            raise ValueError("Currency mismatch")
        return Money(self.amount + other.amount, self.currency)

# Entities (with identity)
class Order:
    """Entity: has identity, mutable state."""

    def __init__(self, id: str, customer: Customer):
        self.id = id
        self.customer = customer
        self.items: List[OrderItem] = []
        self.status = OrderStatus.PENDING
        self._events: List[DomainEvent] = []

    def add_item(self, product: Product, quantity: int):
        """Business logic in entity."""
        item = OrderItem(product, quantity)
        self.items.append(item)
        self._events.append(ItemAddedEvent(self.id, item))

    def total(self) -> Money:
        """Calculated property."""
        return sum(item.subtotal() for item in self.items)

    def submit(self):
        """State transition with business rules."""
        if not self.items:
            raise ValueError("Cannot submit empty order")
        if self.status != OrderStatus.PENDING:
            raise ValueError("Order already submitted")

        self.status = OrderStatus.SUBMITTED
        self._events.append(OrderSubmittedEvent(self.id))

# Aggregates (consistency boundary)
class Customer:
    """Aggregate root: controls access to entities."""

    def __init__(self, id: str, email: Email):
        self.id = id
        self.email = email
        self._addresses: List[Address] = []
        self._orders: List[str] = []  # Order IDs, not full objects

    def add_address(self, address: Address):
        """Aggregate enforces invariants."""
        if len(self._addresses) >= 5:
            raise ValueError("Maximum 5 addresses allowed")
        self._addresses.append(address)

    @property
    def primary_address(self) -> Optional[Address]:
        return next((a for a in self._addresses if a.is_primary), None)

# Domain Events
@dataclass
class OrderSubmittedEvent:
    order_id: str
    occurred_at: datetime = field(default_factory=datetime.now)

# Repository (aggregate persistence)
class OrderRepository:
    """Repository: persist/retrieve aggregates."""

    async def find_by_id(self, order_id: str) -> Optional[Order]:
        """Reconstitute aggregate from storage."""
        pass

    async def save(self, order: Order):
        """Persist aggregate and publish events."""
        await self._persist(order)
        await self._publish_events(order._events)
        order._events.clear()

Architecture Best Practices

1. Dependency Rule: Dependencies always point inward
2. Interface Segregation: Small, focused interfaces
3. Business Logic in Domain: Keep frameworks out of core
4. Test Independence: Core testable without infrastructure
5. Bounded Contexts: Clear domain boundaries
6. Ubiquitous Language: Consistent terminology
7. Thin Controllers: Delegate to use cases
8. Rich Domain Models: Behavior with data

Common Pitfalls

• Anemic Domain: Entities with only data, no behavior
• Framework Coupling: Business logic depends on frameworks
• Fat Controllers: Business logic in controllers
• Repository Leakage: Exposing ORM objects
• Missing Abstractions: Concrete dependencies in core
• Over-Engineering: Clean architecture for simple CRUD

MANDATORY: External Validation Before Completion

⚠️ CRITICAL REQUIREMENT - NO EXCEPTIONS ⚠️

ALL API implementations MUST be validated with external tools before claiming completion.

Required External Testing

You MUST test endpoints with real HTTP clients:

• curl

  (command line)

• httpie

  (command line)
• Postman (GUI)
• Thunder Client (VS Code)

Test Coverage Required

✅ Success Cases:

• Valid requests with expected data
• Edge cases (empty, optional fields, different locales)
• Pagination (page 1, page 2, last page)

✅ Error Cases:

• Malformed JSON → 400 Bad Request
• Missing required fields → 400/422
• Invalid data types → 400/422
• Not found scenarios → 404
• Server errors → 500 (if applicable)

✅ Performance:

• Response time measurement
• Verify meets acceptance criteria

Documentation Required

Your implementation report MUST include:

## External Validation

### Success Cases
**Command:**
```bash
curl -X POST http://localhost:8011/api/v1/endpoint \
  -H "Content-Type: application/json" \
  -d '{"key": "value"}'

Response (200 OK):

{...actual response...}

Validation:

• ✅ Status: 200 OK
• ✅ Response matches schema
• ✅ Response time: X.Xs (< target)

Error Cases

[... similar format for error scenarios ...]

Performance Metrics

• Average: X.Xs
• P95: X.Xs
• Target: < Y.Ys ✅ PASS

### Why This Matters

**Prevents:**
- ❌ Frontend discovering backend bugs
- ❌ Multiple iteration loops
- ❌ "Works on my machine" syndrome

**Ensures:**
- ✅ Real HTTP validation
- ✅ API contract verified
- ✅ Error handling works
- ✅ Frontend can trust backend

### Enforcement

Coordinator will **REJECT** your work if:
- No external validation section in report
- Only unit tests shown (insufficient)
- No curl commands with actual responses
- No error case testing

**Full Requirements:** `.sdlc-workflow/guides/backend-validation-requirements.md`

---

## MANDATORY: Read Before Implementation

- [FastAPI Standards](references/guidelines-fastapi.md) - Architecture, patterns, best practices
- [Python Standards](references/guidelines-python.md) - Code quality, typing, idioms