Python FastAPI Service Layout

First PublishedMay 7, 2026ByAtif Alam

This page is a folder and responsibility map for a typical service, not a step-by-step tutorial. Pair it with your team’s templates or onboarding guides. For pyproject.toml, virtual environments, and dependency installs, see Environment setup. For a client-side stdlib walkthrough (fetch JSON, normalize records), see Read from API and process data.

Production-grade structure for both public-facing APIs and internal microservices. Same skeleton scales from one service to many.

Folder structure

service/
├── pyproject.toml
├── Dockerfile
├── docker-compose.yml          # local dev: app + postgres + redis + jaeger
├── Makefile                    # make test / lint / run / migrate
├── .env.example
├── alembic/                    # DB migrations
│   └── versions/
├── openapi/
│   └── openapi.yaml            # generated from code or hand-authored
├── deploy/
│   ├── helm/                   # k8s chart
│   └── k8s/                    # raw manifests if no helm
├── tests/
│   ├── conftest.py             # fixtures: client, db, auth
│   ├── unit/
│   ├── integration/
│   └── contract/               # validates handlers against openapi.yaml
└── src/
    └── app/
        ├── __init__.py
        ├── main.py             # FastAPI() + lifespan + middleware wiring
        ├── config.py           # Pydantic Settings, env-driven
        ├── deps.py             # shared dependencies (db session, current user)
        │
        ├── api/                # HTTP layer — thin
        │   ├── __init__.py
        │   ├── v1/
        │   │   ├── __init__.py
        │   │   ├── router.py   # aggregates v1 routers
        │   │   ├── users.py
        │   │   ├── orders.py
        │   │   └── health.py
        │   └── internal/       # service-to-service endpoints
        │       └── router.py
        │
        ├── schemas/            # Pydantic request/response DTOs
        │   ├── user.py
        │   └── order.py
        │
        ├── domain/             # business logic — no FastAPI, no SQLAlchemy
        │   ├── models.py       # dataclasses or domain entities
        │   ├── users.py        # UserService
        │   └── orders.py       # OrderService
        │
        ├── repositories/       # data access — SQLAlchemy lives here
        │   ├── base.py
        │   ├── users.py
        │   └── orders.py
        │
        ├── db/
        │   ├── session.py      # engine, async_sessionmaker
        │   └── models.py       # SQLAlchemy ORM models
        │
        ├── middleware/
        │   ├── request_id.py
        │   ├── logging.py
        │   ├── error_handler.py
        │   ├── rate_limit.py
        │   └── auth.py
        │
        ├── observability/
        │   ├── logging.py      # structlog config
        │   ├── metrics.py      # prometheus_client
        │   └── tracing.py      # OpenTelemetry setup
        │
        ├── errors.py           # AppError, ValidationError, NotFound, ...
        └── clients/            # outbound: other services, S3, Redis
            ├── http.py         # httpx client w/ retries, timeouts
            └── cache.py

Why this layering: api → domain → repositories → db. Dependencies point inward only. The domain layer never imports FastAPI or SQLAlchemy, which means you can unit-test business logic without spinning up a test client or database.

main.py — wiring

from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.config import settings
from app.api.v1.router import router as v1_router
from app.api.internal.router import router as internal_router
from app.middleware.request_id import RequestIDMiddleware
from app.middleware.logging import LoggingMiddleware
from app.middleware.error_handler import register_exception_handlers
from app.observability.tracing import setup_tracing
from app.observability.metrics import setup_metrics
from app.db.session import engine

@asynccontextmanager
async def lifespan(app: FastAPI):
    setup_tracing(app)
    setup_metrics(app)
    yield
    await engine.dispose()

app = FastAPI(
    title=settings.service_name,
    version=settings.version,
    lifespan=lifespan,
    docs_url="/docs" if settings.env != "prod" else None,  # hide swagger in prod
)

# Order matters: outermost first
app.add_middleware(RequestIDMiddleware)
app.add_middleware(LoggingMiddleware)
# CORS, auth, rate limit added here per environment

register_exception_handlers(app)

app.include_router(v1_router, prefix="/v1")
app.include_router(internal_router, prefix="/internal")  # mTLS-only at ingress

Middleware chain (outside → inside)

Request
  │
  ├─ RequestIDMiddleware     # generate/propagate X-Request-ID
  ├─ TracingMiddleware       # OTel span, attach trace_id
  ├─ LoggingMiddleware       # structured access log w/ duration
  ├─ AuthMiddleware          # validates JWT, sets request.state.principal
  ├─ RateLimitMiddleware     # per-principal token bucket (Redis)
  ├─ Router → handler
  │     └─ Depends(get_db), Depends(get_current_user), ...
  └─ ExceptionHandler        # converts AppError → JSON response

Error handling pattern

One error type per failure mode in the domain, mapped centrally to HTTP.

app/errors.py

class AppError(Exception):
    code: str = "INTERNAL_ERROR"
    http_status: int = 500
    def __init__(self, message: str, details: dict | None = None):
        self.message = message
        self.details = details or {}

class NotFoundError(AppError):
    code = "NOT_FOUND"
    http_status = 404

class ValidationFailed(AppError):
    code = "VALIDATION_FAILED"
    http_status = 422

class ConflictError(AppError):
    code = "CONFLICT"
    http_status = 409

class UnauthorizedError(AppError):
    code = "UNAUTHORIZED"
    http_status = 401

class ForbiddenError(AppError):
    code = "FORBIDDEN"
    http_status = 403

class RateLimited(AppError):
    code = "RATE_LIMITED"
    http_status = 429

app/middleware/error_handler.py

import structlog
from fastapi import Request
from fastapi.responses import JSONResponse
from app.errors import AppError

log = structlog.get_logger()

def register_exception_handlers(app):
    @app.exception_handler(AppError)
    async def app_error_handler(request: Request, exc: AppError):
        return JSONResponse(
            status_code=exc.http_status,
            content={
                "error": {
                    "code": exc.code,
                    "message": exc.message,
                    "details": exc.details,
                    "trace_id": getattr(request.state, "trace_id", None),
                }
            },
        )

    @app.exception_handler(Exception)
    async def unhandled(request: Request, exc: Exception):
        log.exception("unhandled_error", path=request.url.path)
        return JSONResponse(
            status_code=500,
            content={
                "error": {
                    "code": "INTERNAL_ERROR",
                    "message": "Internal server error",
                    "trace_id": getattr(request.state, "trace_id", None),
                }
            },
        )

Domain raises typed errors; handlers stay clean:

app/api/v1/users.py

@router.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: UUID, svc: UserService = Depends(get_user_service)):
    return await svc.get(user_id)   # raises NotFoundError if missing

Public vs internal split

• Public (/v1/*): OAuth2/JWT auth, strict rate limits, full input validation, conservative response shapes (no internal IDs/state)
• Internal (/internal/*): mTLS at ingress (Istio/Linkerd), looser rate limits, can expose internal fields, separate router so it’s easy to gate at the gateway
• Same codebase, same observability, different middleware stacks — apply auth middleware conditionally based on path prefix

Key dependencies

[project]
dependencies = [
    "fastapi>=0.110",
    "pydantic>=2.5",
    "pydantic-settings",
    "uvicorn[standard]",
    "sqlalchemy[asyncio]>=2.0",
    "asyncpg",
    "alembic",
    "httpx",
    "redis",
    "structlog",
    "prometheus-client",
    "opentelemetry-api",
    "opentelemetry-sdk",
    "opentelemetry-instrumentation-fastapi",
    "opentelemetry-instrumentation-sqlalchemy",
    "python-jose[cryptography]",  # JWT
]