The FastAPI backend is organised in layers that separate concerns: routers handle HTTP routing and dependency injection; services contain business logic and call the database; models define the SQLAlchemy ORM schema; schemas define Pydantic validation for input and output. This layered structure means you can test business logic in the service layer without HTTP, swap the database in tests, and change the API shape without touching the business logic.
Backend Directory Structure
app/
├── main.py # App factory, middleware, lifespan
├── config.py # Pydantic Settings, validated on import
├── database.py # SQLAlchemy engine, session factory
│
├── models/ # SQLAlchemy ORM models
│ ├── user.py # User, RefreshToken
│ ├── post.py # Post, Tag, PostTag (association)
│ ├── comment.py # Comment
│ └── notification.py # Notification
│
├── schemas/ # Pydantic request/response schemas
│ ├── auth.py # LoginRequest, TokenResponse, RegisterRequest
│ ├── post.py # PostCreate, PostUpdate, PostResponse, PaginatedPostsResponse
│ ├── comment.py # CommentCreate, CommentResponse
│ ├── user.py # UserResponse, UserUpdate
│ └── common.py # PaginatedResponse generic base
│
├── routers/ # FastAPI APIRouter instances
│ ├── auth.py # /api/auth/*
│ ├── posts.py # /api/posts/*
│ ├── comments.py # /api/comments/* (nested under posts)
│ ├── users.py # /api/users/*
│ ├── tags.py # /api/tags/*
│ ├── notifications.py # /api/notifications/*
│ ├── ws.py # /ws/* WebSocket endpoints
│ └── health.py # /api/health
│
├── services/ # Business logic (no HTTP knowledge)
│ ├── auth.py # hash_password, verify_password, create_tokens, verify_token
│ ├── posts.py # create_post, update_post, delete_post, get_paginated_posts
│ ├── comments.py # create_comment, delete_comment
│ └── users.py # update_avatar, update_profile
│
├── dependencies/ # FastAPI dependency functions
│ ├── auth.py # get_current_user, require_owner, require_admin
│ └── db.py # get_db (session generator)
│
├── middleware/ # Custom ASGI middleware
│ ├── rate_limit.py # slowapi Limiter instance
│ ├── request_id.py # RequestIdMiddleware
│ └── metrics.py # MetricsMiddleware
│
└── ws/ # WebSocket infrastructure
├── connection_manager.py # ConnectionManager (per-user)
└── room_manager.py # RoomManager (per-post room)Note: The service layer contains pure Python functions that take a database session and domain objects as arguments and return domain objects. They have no knowledge of HTTP, FastAPI, or request/response shapes. This makes them independently testable:
result = create_post(db, user_id=1, data=PostCreate(title="Test", ...)) — no mock HTTP request needed. The router is responsible only for extracting data from the request, calling the service, and returning a response.Tip: The
main.py app factory pattern (a function that creates and configures the FastAPI app) makes testing easier — tests can call create_app() with test-specific settings instead of importing the global app object. It also makes it easy to add application-level configuration (different middleware for development vs production) without polluting the module-level scope.Warning: As the application grows, resist the temptation to add business logic directly in routers. A router handler that is longer than 20 lines is a warning sign — extract the logic into the service layer. Router handlers should read like a summary: “validate input, call service, return response.” If the handler has complex conditional logic, it belongs in the service where it can be unit tested without the HTTP layer.
main.py — App Factory
# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from app.config import settings
from app.database import engine, Base
from app.middleware.request_id import RequestIdMiddleware
from app.middleware.metrics import MetricsMiddleware
from app.routers import auth, posts, comments, users, tags, notifications, ws, health
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup: nothing here — migrations run in entrypoint.sh
yield
# Shutdown: close connection pools, flush metrics
engine.dispose()
def create_app() -> FastAPI:
app = FastAPI(
title = "Blog API",
version = "1.0.0",
lifespan = lifespan,
docs_url = "/api/docs" if not settings.is_production else None,
redoc_url = None,
)
# Middleware (applied in reverse order — last added runs first)
app.add_middleware(RequestIdMiddleware)
app.add_middleware(MetricsMiddleware)
app.add_middleware(
CORSMiddleware,
allow_origins = settings.cors_origins_list,
allow_credentials = True,
allow_methods = ["*"],
allow_headers = ["*"],
)
# Routers
app.include_router(health.router)
app.include_router(auth.router, prefix="/api")
app.include_router(posts.router, prefix="/api")
app.include_router(comments.router, prefix="/api")
app.include_router(users.router, prefix="/api")
app.include_router(tags.router, prefix="/api")
app.include_router(notifications.router, prefix="/api")
app.include_router(ws.router)
return app
app = create_app()