aibot88 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
- 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
- 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
- Dependency Rule: Dependencies always point inward
- Interface Segregation: Small, focused interfaces
- Business Logic in Domain: Keep frameworks out of core
- Test Independence: Core testable without infrastructure
- Bounded Contexts: Clear domain boundaries
- Ubiquitous Language: Consistent terminology
- Thin Controllers: Delegate to use cases
- 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