FastAPI 架构指南：用这份模版打造可扩展又安全的系统（附实战经验）

原创 AI研究生 AI大模型观察站 2025-10-27 08:16

我在生产环境中将这套结构打磨用于支持 500+ 并发用户；它扩展轻松、维护不累、生产力全速输出。

如果你只想直接在项目中使用该模板的 FastAPI 结构，请查看这个 GitHub repo，内含完整项目结构：
https://github.com/brianobot/fastAPI_project_structure

简介

过去一年里我几乎每天都在使用 FastAPI。在此过程中，我总结出了一套标准项目结构：既能让我保持对改动与新增功能的积极性，又能很好地扩展。在这篇文章里，我会拆解我的 FastAPI “Go-To architecture” 的每个部分——这是我在新项目启动或旧项目重构时优先优化的那一套。不管你是想更快迭代想法、梳理日益增长的代码库，还是强化应用安全性，这套结构都能帮你自信构建，并像专业人士一样进化你的项目。

项目结构

---

App 目录

App 目录是应用的引擎与核心，包含支撑大多数后端基础设施用例所需的全部模块与代码。下面我会解释该目录中的每个子目录与模块，并说明它们的使用理由，指出常见问题，以及为提升代码质量与开发者体验所做的改进。

init
.py：在每个 Python 包中放置一个 init
.py 是标准做法，它用于将文件夹声明为常规 Python Package。

api_router.py：我用它作为应用中统一管理全部 routers 的中心模块，这样可以方便实现版本控制（如 /v1/users、/v2/users），并让代码库保持清晰易读。

# apps/api\_router.pyfrom fastapi import APIRouterapi\_v1 = APIRouter(prefix="/v1")
api\_v2 = APIRouter(prefix="/v2")# include routes to a root route
# from app.routers import routers as user\_routers
# api\_v1.include\_router(users\_routers)

logger.py：Logging 是每个软件应用的关键部分，它能提供运行时可见性，并在问题与 bug 出现时帮助定位与解决（它们终究会出现）。与其被繁杂配置淹没，这个模板在我几乎所有项目中都非常好用；你也可以拓展它，比如接入外部 handler，将日志发送到可视化面板或分析工具。这里的配置通过 TimedRotatingFileHandler 按天归档日志，让调试更高效。

# apps/logger.pyimport logging
import jsonfrom logging.handlers import TimedRotatingFileHandlerlogger = logging.getLogger()classJsonFormatter(logging.Formatter):defformat(self, record):log\_record = {"timestamp": self.formatTime(record, self.datefmt),"level": record.levelname,"module": record.module,"funcName": record.funcName,"lineno": record.lineno,"message": record.getMessage(),}return json.dumps(log\_record)file\_handler = TimedRotatingFileHandler("logs/app.log", when="midnight", interval=1 / 86400, backupCount=7
)file\_handler.setFormatter(JsonFormatter())logger.handlers = [file\_handler]
logger.setLevel(logging.INFO)

main.py：主模块是应用的入口点，应用的大部分安全策略与规则都在这里设置。

# apps/mains.pyfrom contextlib import asynccontextmanager
from datetime import datetime, UTC
from fastapi import FastAPIfrom starlette.middleware.trustedhost import TrustedHostMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import RedirectResponse
from fastapi.responses import JSONResponse
from fastapi.requests import Request
from fastapi import HTTPException
from fastapi import FastAPIfrom slowapi import Limiter
from slowapi.util import get\_remote\_addressfrom redis import asyncio as aioredisfrom app.logger import logger
from app.api\_router import api
from app.settings import Settings
from app.middlewares import log\_request\_middlewaresettings = Settings()@asynccontextmanager
asyncdeflifespan(app: FastAPI):yielddefinitiate\_app():app = FastAPI(title="FastAPI Sample Project",summary="API for FastAPI Sample Project",lifespan=lifespan,)origins = [# Add allowed origins here]app.add\_middleware(CORSMiddleware,allow\_origins=origins,allow\_credentials=True,allow\_methods=["\*"],allow\_headers=["\*"],)# tweak this to see the most efficient sizeapp.add\_middleware(GZipMiddleware, minimum\_size=100)app.add\_middleware(TrustedHostMiddleware,allowed\_hosts=[# Add allowed hosts here, this controls host that the application can be hosted on],)app.add\_middleware(BaseHTTPMiddleware, dispatch=log\_request\_middleware)# this handles rate limiting protecting the API from abuse or Denial of Service Attackslimiter = Limiter(key\_func=get\_remote\_address)app.state.limiter = limiterapp.include\_router(api)return appapp = initiate\_app()@app.exception\_handler(HTTPException)
asyncdefhttp\_exception\_handler(request: Request, exc: HTTPException):return JSONResponse(status\_code=exc.status\_code,content={"detail": exc.detail,"path": request.url.path,"timestamp": datetime.now(UTC).isoformat(),},)@app.exception\_handler(Exception)
asyncdefglobal\_exception\_handler(request: Request, exc: Exception):logger.error(f"Unexpected error: {str(exc)}")return JSONResponse(status\_code=500,content={"detail": "An unexpected error occurred","path": request.url.path,},)@app.get("/", tags=["Root"])
asyncdefroot():# this redirects root requests to the docs pagereturn RedirectResponse("/docs")

settings.py：我发现使用 pydantic 的 BaseSettings 处理环境变量既简单又干净。你可以为环境变量定义类型，这些类型会在应用启动时验证；对于可选环境变量，还可以提供默认值。

apps/settings.py

from pydantic_settings import BaseSettings
from pydantic import ConfigDict, AnyUrl

classSettings(BaseSettings):
model_config = ConfigDict(env_file=".env")

example_secret: str = "example secret value"JWT_SECRET: str# required environment variable 
JWT_ALGORITHM: str = "HS256"# optional environement variable with default value

dependencies.py：我用这个模块来集中管理自定义的路由依赖。像 get_db 这类常用依赖（例如获取与数据库交互的 Session）都可以在这里声明。单独的依赖模块能让代码组织更清晰、更模块化（不同路由通常都会依赖这些依赖项），测试也更方便，因为逻辑可以被隔离验证。

middlewares.py：我用这个模块来集中管理应用的自定义 middlewares。为依赖模块所述的模块化优势，这里同样适用。

目录

Routers 目录：我倾向于为每个逻辑路由分组创建独立模块，使开发更愉悦、更简单。例如，用一个 auth.py 模块来放与用户认证和用户资料管理相关的全部路由；用 products.py 放与商品管理相关的路由；用 admin.py 放与管理员 API 相关的全部路由。我尽量让路由函数不超过 2 行代码：路由函数只负责声明路由、定义依赖与请求参数；每个路由的业务逻辑放在对应的 service 函数中。

# apps/routers/auth.pyfrom typing import Annotatedfrom pydantic import EmailStr
from fastapi.routing import APIRouter
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi import Depends, Body, BackgroundTasksfrom app.dependencies import get\_db
from app.schemas import auth as auth\_schemas
from app.services import auth as auth\_servicesrouter = APIRouter(prefix="/auth", tags=["Authentication"])EmailBody = Annotated[EmailStr, Body(embed=True)]
DBDep = Annotated[AsyncSession, Depends(get\_db)]@router.post("/signup", response\_model=auth\_schemas.UserModel)
asyncdefsignup(db: DBDep,bg\_task: BackgroundTasks,request\_data: auth\_schemas.UserSignUpData,
):returnawait auth\_services.signup\_user(request\_data, db, bg\_task)

这样的路由函数看起来更清爽，业务逻辑由 service 函数处理；路由本身只关注依赖、请求体或查询参数等要求。

Schema 目录：我把所有 pydantic models 放在 schema 模块。与 Routers 目录类似，Schema 目录中包含全部 schemas 模块。通常我会让每个路由模块与一个对应的 schema 模块和一个 service 模块相匹配。这样应用的每个组件都能以非常逻辑化、可预期的方式归档。例如 auth.py 的一个 schema 模型如下：

# apps/schemas/auth.py
from uuid import UUID
from datetime import datetime
from typing import Annotatedfrom pydantic import BaseModel, EmailStr, FieldclassUserSignUpData(BaseModel):password: Annotated[str, Field(min\_length=8)]email: Annotated[EmailStr, Field(max\_length=254)]classUserModel(BaseModel):id: UUIDemail: EmailStrdate\_created: datetimedate\_updated: datetime

Service 目录：Service 目录存放实际的业务逻辑模块，比如调用第三方 API、对数据库发起查询等。将它们这样分离可以把路由对 API 的声明性需求与具体实现进行关注点分离。与 Routers 和 Schemas 目录一样，我会让它们在组件维度上一一对应。以下以 auth 组件为例：

# apps/services/auth.pyfrom datetime import timedeltafrom fastapi import HTTPException
from passlib.context import CryptContext
from fastapi.security import OAuth2PasswordBearerfrom sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSessionfrom app.settings import Settings
from app.models import User as UserDB
from app.schemas import auth as auth\_schemasettings = Settings()JWT\_SECRET = settings.JWT\_SECRET
JWT\_ALGORITHM = settings.JWT\_ALGORITHMACCESS\_TOKEN\_LIFESPAN = timedelta(days=2)
REFRESH\_TOKEN\_LIFESPAN = timedelta(days=5)pwd\_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2\_scheme = OAuth2PasswordBearer(tokenUrl="v1/auth/token")defverify\_password(plain\_password: str, hashed\_password: str):return pwd\_context.verify(plain\_password, hashed\_password)defget\_password\_hash(password: str):return pwd\_context.hash(password)asyncdefget\_user(email: str, session: AsyncSession) -> UserDB | None:stmt = select(UserDB).where(UserDB.email == email)result = await session.execute(stmt)return result.scalar\_one\_or\_none()asyncdefcreate\_user(user\_data: auth\_schema.UserSignUpData,session: AsyncSession,
):result = await session.execute(select(UserDB).where(UserDB.email == user\_data.email))if result.scalar\_one\_or\_none():raise HTTPException(status\_code=400, detail="Email already registered")result = await session.execute(select(UserDB).where(UserDB.email == user\_data.email))if result.scalar\_one\_or\_none():raise HTTPException(status\_code=400, detail="Username already used")hashed\_password = get\_password\_hash(user\_data.password)new\_user = UserDB(email=user\_data.email,password=hashed\_password,username=user\_data.email,)session.add(new\_user)await session.commit()await session.refresh(new\_user)return new\_userasyncdefsignup\_user(data: auth\_schema.UserSignUpData,session: AsyncSession,
):returnawait create\_user(data, session)

除了核心的 Python 后端部分，我也很推荐使用 Makefile 这类工具，来简化常见命令行工具的调用，比如启动 FastAPI server、运行 pytest 等。

# Makefilerun-local:fastapi dev app/main.pytest-local:pytest -s --covcoverage-report:coverage reportcoverage-html:coverage report && coverage html

要使用 make，需要先在你的电脑上安装它。

在项目根目录放置好这个文件后，每当我想启动 FastAPI server，只需运行：

make run-local

这样就不用每次都敲完整命令了，对于频繁执行的命令工具，能显著提升效率。

---

总结

在这份模板中，我没有包含数据库连接的设置。维护一个模块化的项目结构和代码库，会在后续带来超出预期的收益：可预期、合理的功能与工具组织，会让代码库的“美感”超越功能本身。尽管这一套模式在过去一年里显著提升了我的生产力，我仍在持续学习并拥抱更好的后端架构方式与实践。这份模板的完整代码可在 Github 查看：
https://github.com/brianobot/fastAPI_project_structure

如果你觉得有用请关注微信公众号：LLM大模型观察站
原文地址：https://mp.weixin.qq.com/s/bEWoeaY4KacuprpFEFe2jw