Fastapi Backend Guidelines

FastAPI Backend Development Guidelines

Purpose

Comprehensive guide for modern FastAPI development with async Python, emphasizing Domain-Driven Design, layered architecture (Router → Service → Repository), SQLModel ORM, and async best practices.

When to Use This Skill

• Creating new API routes or endpoints
• Building domain services and business logic
• Implementing repositories for data access
• Setting up database models with SQLModel
• Async/await patterns and error handling
• Organizing backend code with DDD
• Pydantic validation and DTOs
• Python async best practices

Quick Start

New API Route Checklist

Creating an API endpoint? Follow this checklist:

• Define route in backend/api/v1/routers/{domain}.py
• Use FastAPI dependency injection for session
• Use get_read_session_dependency for reads, get_write_session_dependency for writes
• Call service layer (don't access repository directly)
• Use Pydantic DTOs for request/response
• Handle errors with custom exceptions
• Add proper HTTP status codes
• Use async/await throughout
• Document with docstrings
• Use type hints on all parameters

New Domain Feature Checklist

Creating a new domain? Set up this structure:

• Create backend/domain/{domain}/ directory
• Create model.py - SQLModel database models with ULID ID generation
• Create repository.py - Data access layer extending BaseRepository
• Create service.py - Business logic layer
• Create DTOs in backend/dtos/{domain}.py
• Create router in backend/api/v1/routers/{domain}.py
• Register router in main.py
• Follow async patterns throughout
• Add enums to backend/domain/user/enums.py if needed

Project Structure Quick Reference

Your YGS backend structure:

backend/
  backend/
    main.py                  # FastAPI app creation with lifespan

    api/
      v1/
        routers/             # API route handlers
          admin.py           # Dashboard, members, matching
          auth.py            # Login, signup, OAuth
          match.py           # Match weeks, history
          user.py            # User management
          upload.py          # S3 presigned URLs

    domain/                  # Domain-Driven Design
      user/
        model.py             # User, UserProfile, UserLifestyle, etc.
        repository.py        # UserRepository, UserDataLoader
        service.py           # UserService
        enums.py             # All domain enums
      auth/
        service.py           # AuthService (JWT, Firebase, Kakao)
        repository.py        # AuthRepository
      admin/
        model.py             # ConsultSchedule
        service.py           # AdminService
        repository.py        # AdminRepository
        matching_service.py  # MatchingService (scoring algorithm)
      match/
        model.py             # MatchWeek, MatchHistory, MatchFeedback
        service.py           # MatchService
        repository.py        # MatchRepository
      llm/
        matching_service.py  # LLM-enhanced matching
      shared/
        base_repository.py   # Generic BaseRepository

    dtos/                    # Pydantic DTOs
      admin.py               # Dashboard, member DTOs
      auth.py                # Login, signup, OAuth DTOs
      match.py               # Match week, history DTOs
      user.py                # User, profile DTOs
      llm_match.py           # LLM matching DTOs

    db/
      orm.py                 # Read/Write session management

    core/
      config.py              # Pydantic Settings configuration

    middleware/              # Middleware
      error_handler.py       # ErrorHandlerMiddleware
      admin_auth.py          # Admin authentication

    utils/                   # Utilities
      s3.py                  # S3 presigned URLs
      s3_private.py          # Private user data S3
      firebase.py            # Firebase verification
      password.py            # bcrypt hashing
      excel.py               # Excel export

    error/                   # Custom exceptions
      __init__.py            # AppException, NotFoundError, etc.

Common Imports Cheatsheet

# FastAPI
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from fastapi.responses import StreamingResponse

# SQLModel & SQLAlchemy
from sqlmodel import select, or_, and_, col
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy import func, desc
from sqlalchemy.orm import selectinload

# Database
from backend.db.orm import get_write_session_dependency, get_read_session_dependency

# Pydantic
from pydantic import BaseModel, Field, field_validator, EmailStr

# Your domain
from backend.domain.user.model import User, UserProfile
from backend.domain.user.service import UserService
from backend.dtos.user import UserResponse, UserCreateRequest
from backend.error import NotFoundError, ForbiddenError, UnauthorizedError

# Type hints
from typing import List, Optional, Dict, Any

# ID generation
from ulid import ULID

Topic Guides

🏗️ Layered Architecture

Three-Layer Pattern:

1. Router Layer: API endpoints, request validation, response formatting
2. Service Layer: Business logic, orchestration, domain rules
3. Repository Layer: Data access, queries, database operations

Key Concepts:

• Routers call Services (never Repositories directly)
• Services orchestrate business logic
• Repositories handle all database operations
• Each layer has clear responsibilities
• Async/await throughout the stack
• Read/Write session separation

📖 Complete Guide: resources/layered-architecture.md

🛣️ API Routes & Routers

PRIMARY PATTERN: FastAPI Routers

• Create routers in backend/api/v1/routers/
• Use dependency injection for sessions
• Use get_read_session_dependency for GET requests
• Use get_write_session_dependency for POST/PATCH/DELETE
• Follow REST conventions
• Use appropriate HTTP methods and status codes
• Async route handlers

Router Structure:

from fastapi import APIRouter, Depends
from sqlmodel.ext.asyncio.session import AsyncSession
from backend.db.orm import get_read_session_dependency, get_write_session_dependency

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

@router.get("/{user_id}")
async def get_user(
    user_id: str,
    session: AsyncSession = Depends(get_read_session_dependency),
) -> UserResponse:
    service = UserService(session)
    return await service.get_user(user_id)

@router.post("", status_code=201)
async def create_user(
    request: UserCreateRequest,
    session: AsyncSession = Depends(get_write_session_dependency),
) -> UserResponse:
    service = UserService(session)
    return await service.create_user(request)

📖 Complete Guide: resources/api-routes.md

🗄️ Database & ORM

SQLModel + SQLAlchemy:

• SQLModel for models (combines SQLAlchemy + Pydantic)
• Async sessions with asyncpg driver
• Read/Write session separation with caching
• Repository pattern for all queries
• ULID-based ID generation with prefixes

Model Pattern:

from sqlmodel import SQLModel, Field, Column, DateTime, Text
from datetime import datetime, timezone
from ulid import ULID

def generate_user_id() -> str:
    """Generate user ID with prefix."""
    return f"usr_{ULID()}"

class User(SQLModel, table=True):
    __tablename__ = "user"

    id: str = Field(
        default_factory=generate_user_id,
        primary_key=True,
        max_length=30,
    )
    phone: str = Field(sa_column=Column(Text, nullable=False, unique=True))
    name: str = Field(sa_column=Column(Text, nullable=False))
    gender: GenderEnum = Field(sa_column=Column(Text, nullable=False))

    # Soft delete pattern
    deleted_at: Optional[datetime] = Field(
        sa_column=Column(DateTime(timezone=True), nullable=True),
        default=None,
    )

    # Timestamps
    created_at: datetime = Field(
        sa_column=Column(DateTime(timezone=True), nullable=False),
        default_factory=lambda: datetime.now(tz=timezone.utc),
    )
    updated_at: datetime = Field(
        sa_column=Column(DateTime(timezone=True), nullable=False),
        default_factory=lambda: datetime.now(tz=timezone.utc),
    )

📖 Complete Guide: resources/database-orm.md

📦 Domain-Driven Design

Domain Organization:

• Each domain in backend/domain/{name}/
• Contains: model.py, repository.py, service.py
• Clear separation of concerns
• Business logic in services
• Data access in repositories

Your Domains:

• user: User management (User, UserProfile, UserLifestyle, UserPreference, etc.)
• auth: Authentication (JWT, Firebase, Kakao OAuth)
• admin: Admin dashboard, member management, consultations
• match: Match weeks, history, feedback
• llm: LLM-enhanced matching with Gemini
• shared: BaseRepository, common utilities

📖 Complete Guide: resources/domain-driven-design.md

🔄 Service Layer

Service Pattern:

• Business logic orchestration
• Domain rule enforcement
• Calls repositories for data
• Returns DTOs, not models directly
• Transaction management
• Uses asyncio.gather for parallel queries

Service Structure:

class UserService:
    def __init__(self, session: AsyncSession):
        self.session = session
        self._user_repo = UserRepository(session)
        self._profile_repo = UserProfileRepository(session)
        self._data_loader = UserDataLoader(session)

    async def get_user_detail(self, user_id: str) -> UserDetailResponse:
        # Use UserDataLoader for N+1 prevention
        user_with_relations = await self._data_loader.load_user_with_relations(
            user_id,
            load_profile=True,
            load_photos=True,
        )
        if not user_with_relations:
            raise NotFoundError(f"User {user_id} not found")
        return self._to_detail_response(user_with_relations)

📖 Complete Guide: resources/service-layer.md

💾 Repository Pattern

Repository Pattern:

• Encapsulates data access
• Extends BaseRepository for CRUD
• Domain-specific queries
• Returns domain models
• All queries are async
• Soft delete support

Repository Structure:

from backend.domain.shared.base_repository import BaseRepository

class UserRepository(BaseRepository[User]):
    def __init__(self, session: AsyncSession):
        super().__init__(session, User)

    async def find_by_phone(self, phone: str) -> Optional[User]:
        stmt = select(User).where(
            User.phone == phone,
            User.deleted_at.is_(None),
        )
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()

UserDataLoader Pattern (N+1 Prevention):

@dataclass
class UserWithRelations:
    user: User
    profile: Optional[UserProfile] = None
    lifestyle: Optional[UserLifestyle] = None
    photos: List[UserPhoto] = field(default_factory=list)

class UserDataLoader:
    async def load_user_with_relations(
        self,
        user_id: str,
        load_profile: bool = False,
        load_photos: bool = False,
    ) -> Optional[UserWithRelations]:
        # Build queries based on flags
        queries = [self._load_user(user_id)]
        if load_profile:
            queries.append(self._load_profile(user_id))
        if load_photos:
            queries.append(self._load_photos(user_id))

        # Execute all queries in parallel
        results = await asyncio.gather(*queries)
        # ... combine results

📖 Complete Guide: resources/repository-pattern.md

📝 DTOs & Validation

Pydantic DTOs:

• Request/Response data transfer objects
• Validation with Pydantic
• Separate from domain models
• Located in backend/dtos/
• Use field_validator for enum validation

DTO Pattern:

from pydantic import BaseModel, Field, field_validator
from backend.domain.user.enums import GenderEnum, UserStatusEnum

class AdminBasicInfoUpdateRequest(BaseModel):
    """Request DTO for updating basic user info (admin only)."""

    name: Optional[str] = Field(None, description="User name")
    status: Optional[str] = Field(None, description="User status")
    is_admin: Optional[bool] = Field(None, description="Admin flag")
    birth_year: Optional[int] = Field(None, ge=1940, le=2010, description="Birth year")

    model_config = {"extra": "forbid"}  # Reject unknown fields

    @field_validator("status")
    @classmethod
    def validate_status(cls, v: Optional[str]) -> Optional[str]:
        if v is not None:
            valid_values = [e.value for e in UserStatusEnum]
            if v not in valid_values:
                raise ValueError(f"Invalid status: {v}. Valid: {valid_values}")
        return v

📖 Complete Guide: resources/dtos-validation.md

⚡ Async/Await Patterns

Async Best Practices:

• Use async/await throughout
• Async database sessions
• Proper session cleanup
• Avoid blocking operations
• Use asyncio.gather for parallel queries

Async Patterns:

# Parallel queries with asyncio.gather
async def get_dashboard_data(self) -> dict:
    # Run all queries in parallel
    total, monthly, weekly, today = await asyncio.gather(
        self._get_total_count(),
        self._get_monthly_count(),
        self._get_weekly_count(),
        self._get_today_count(),
    )
    return {
        "total_members": total,
        "monthly_members": monthly,
        "weekly_members": weekly,
        "today_members": today,
    }

📖 Complete Guide: resources/async-patterns.md

🚨 Error Handling

Error Handling Strategy:

• Custom exception classes in backend/error/
• HTTP exception mapping via ErrorHandlerMiddleware
• Middleware for error handling
• Consistent error responses

Error Pattern:

# backend/error/__init__.py
class AppException(Exception):
    def __init__(self, message: str):
        self.message = message
        super().__init__(self.message)

class NotFoundError(AppException):
    pass

class ForbiddenError(AppException):
    pass

class UnauthorizedError(AppException):
    pass

# In service
if not user:
    raise NotFoundError(f"User {user_id} not found")

# ErrorHandlerMiddleware handles conversion to HTTP response
# NotFoundError → 404, ForbiddenError → 403, etc.

📖 Complete Guide: resources/error-handling.md

📚 Complete Examples

Full working examples:

• Complete domain (model + repository + service + router)
• CRUD operations with async
• Complex queries with SQLModel
• Firebase/Kakao authentication patterns
• S3 presigned URL generation
• Pagination and filtering
• N+1 prevention with UserDataLoader

📖 Complete Guide: resources/complete-examples.md

Navigation Guide

Core Principles

1. Layered Architecture: Router → Service → Repository (never skip layers)
2. Domain-Driven Design: Organize by domain, not by type
3. Async Everything: Use async/await throughout the stack
4. Repository Pattern: All data access through repositories
5. Service Layer: Business logic in services, not routers or repositories
6. DTOs for API: Use Pydantic DTOs for request/response
7. Type Hints: Explicit types on all functions and parameters
8. Error Handling: Custom exceptions, middleware for HTTP mapping
9. Read/Write Split: Separate sessions for read and write operations
10. Dependency Injection: Use FastAPI's Depends() for sessions
11. ULID IDs: Use ULID with entity prefixes (usr_, mw_, mh_, etc.)
12. Soft Delete: Use deleted_at timestamp instead of hard deletes
13. N+1 Prevention: Use asyncio.gather and DataLoader patterns

Quick Reference: New Domain Template

# backend/domain/myfeature/model.py
from sqlmodel import SQLModel, Field, Column, DateTime, Text
from datetime import datetime, timezone
from typing import Optional
from ulid import ULID

def generate_myfeature_id() -> str:
    return f"mf_{ULID()}"

class MyFeature(SQLModel, table=True):
    __tablename__ = "my_feature"

    id: str = Field(
        default_factory=generate_myfeature_id,
        primary_key=True,
        max_length=30,
    )
    name: str = Field(sa_column=Column(Text, nullable=False))

    created_at: datetime = Field(
        sa_column=Column(DateTime(timezone=True), nullable=False),
        default_factory=lambda: datetime.now(tz=timezone.utc),
    )
    deleted_at: Optional[datetime] = Field(
        sa_column=Column(DateTime(timezone=True), nullable=True),
        default=None,
    )

# backend/domain/myfeature/repository.py
from backend.domain.shared.base_repository import BaseRepository
from sqlmodel.ext.asyncio.session import AsyncSession

class MyFeatureRepository(BaseRepository[MyFeature]):
    def __init__(self, session: AsyncSession):
        super().__init__(session, MyFeature)

    async def find_by_name(self, name: str) -> Optional[MyFeature]:
        stmt = select(MyFeature).where(
            MyFeature.name == name,
            MyFeature.deleted_at.is_(None),
        )
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()

# backend/domain/myfeature/service.py
from sqlmodel.ext.asyncio.session import AsyncSession
from backend.error import NotFoundError

class MyFeatureService:
    def __init__(self, session: AsyncSession):
        self.session = session
        self._repository = MyFeatureRepository(session)

    async def get_feature(self, id: str) -> MyFeatureResponse:
        feature = await self._repository.get_by_id(id)
        if not feature:
            raise NotFoundError(f"Feature {id} not found")
        return MyFeatureResponse.model_validate(feature)

# backend/dtos/myfeature.py
from pydantic import BaseModel, Field
from datetime import datetime

class MyFeatureResponse(BaseModel):
    id: str
    name: str
    created_at: datetime

class MyFeatureCreateRequest(BaseModel):
    name: str = Field(..., min_length=1, max_length=255)

# backend/api/v1/routers/myfeature.py
from fastapi import APIRouter, Depends, HTTPException
from sqlmodel.ext.asyncio.session import AsyncSession
from backend.db.orm import get_read_session_dependency, get_write_session_dependency

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

@router.get("/{id}")
async def get_feature(
    id: str,
    session: AsyncSession = Depends(get_read_session_dependency),
) -> MyFeatureResponse:
    service = MyFeatureService(session)
    return await service.get_feature(id)

@router.post("", status_code=201)
async def create_feature(
    request: MyFeatureCreateRequest,
    session: AsyncSession = Depends(get_write_session_dependency),
) -> MyFeatureResponse:
    service = MyFeatureService(session)
    return await service.create_feature(request)

Related Skills

• nextjs-frontend-guidelines: Frontend patterns that consume this API
• error-tracking: Error tracking with Sentry (backend integration)
• pytest-backend-testing: Testing patterns for FastAPI backends

Skill Status: Modular structure with progressive loading for optimal context management