在生产环境中运行这个架构之后，可以自信地说，该架构能够轻松扩容、便于维护，并且能够保持高效的开发生产力。如果您希望在项目中直接使用这个经过验证的FastAPI架构，可以查看GitHub上的完整项目设置：FastAPI Project Structure。
核心架构设计思路
在长期的Python项目开发实践中，我总结并提炼出了一套标准的项目架构模式。它帮助我轻松应对需求变更和功能迭代，同时确保了系统的可扩展性。本文将深入剖析该架构的每一个组件，无论你是希望快速验证产品创意、组织日益庞大的代码库，还是增强应用程序的安全性，这套架构都能助你更自信地构建和发展项目。
项目架构概述
一个组织良好的FastAPI项目的魅力在于其可预测性。当每个组件都有其明确且合理的归属位置时，你就能节省大量寻找文件的时间，从而将更多精力投入到核心功能的构建上。这种架构清晰地划分了职责边界：路由逻辑归路由器管理，业务逻辑由服务层实现，数据验证在模式层完成，而配置则集中在专门的模块中。
App 目录：应用核心引擎
App 目录是应用程序的引擎和核心，包含了支持常见后端基础设施用例所需的所有模块和代码。下面将详细解释该目录中的每个子目录和模块的作用、使用原因、需要避免的常见陷阱，以及为提升代码质量和开发体验所做的设计。
init.py：包声明
在每个 Python 包中放置 __init__.py 文件是标准做法，它将文件夹声明为一个常规的Python包，从而实现整个应用程序中正确的导入机制。
api_router.py：集中化路由管理
此模块作为管理应用程序所有路由的中心化枢纽。它有助于实现API版本控制（例如 /v1/users、/v2/users），并保持代码库的清晰与可读性。
# apps/api_router.py
from fastapi import APIRouter
api_v1 = APIRouter(prefix="/v1")
api_v2 = APIRouter(prefix="/v2")
# 将具体路由包含到根路由中
# from app.routers import routers as user_routers
# api_v1.include_router(users_routers)
这种中心化管理方式意味着，在需要引入API版本控制或弃用旧端点时，你有一个单一的事实来源。无需在分散的路由文件中苦苦搜寻，搞不清哪个版本服务于哪个端点。
logger.py：生产级日志记录
日志记录是每个软件应用不可或缺的部分，它能提供运行时可见性，并在出错时帮助快速定位和解决问题。这个模板中的日志配置在绝大多数项目中都表现优异，并可通过集成外部处理器（如发送到可视化仪表板或分析工具）轻松扩展。
该配置使用 TimedRotatingFileHandler 来按天保存日志，使得调试工作更加高效：
# apps/logger.py
import logging
import json
from logging.handlers import TimedRotatingFileHandler
logger = logging.getLogger()
class JsonFormatter(logging.Formatter):
def format(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)
JSON格式使得日志易于被机器解析，非常适合与ELK Stack或CloudWatch等日志聚合系统集成。定时轮换机制可以在保留一周历史数据的同时，防止日志文件占用过多磁盘空间。
main.py：应用程序入口与安全中心
主模块是应用的入口点，集中了大部分安全策略。在这里，各种中间件被层层叠加，以保护API免受常见威胁：
# apps/main.py
from contextlib import asynccontextmanager
from datetime import datetime, UTC
from fastapi import FastAPI
from 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, JSONResponse
from fastapi.requests import Request
from fastapi import HTTPException
from slowapi import Limiter
from slowapi.util import get_remote_address
from app.logger import logger
from app.api_router import api
from app.settings import Settings
from app.middlewares import log_request_middleware
settings = Settings()
@asynccontextmanager
async def lifespan(app: FastAPI):
yield
def initiate_app():
app = FastAPI(
title="FastAPI Sample Project",
summary="API for FastAPI Sample Project",
lifespan=lifespan,
)
origins = [
# 在此处添加允许的源
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.add_middleware(GZipMiddleware, minimum_size=100)
app.add_middleware(
TrustedHostMiddleware,
allowed_hosts=[
# 在此处添加允许的主机
],
)
app.add_middleware(BaseHTTPMiddleware, dispatch=log_request_middleware)
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.include_router(api)
return app
app = initiate_app()
- CORS中间件 处理跨源资源共享，控制哪些域名可以访问你的API，防止未经授权的网站发起请求。
- GZip中间件 压缩超出指定大小的响应，减少带宽使用并提升慢速连接客户端的响应速度。
- TrustedHost中间件 限制可以运行应用程序的主机，防御主机头注入攻击。
- 速率限制 保护API免遭滥用和拒绝服务攻击，通过限制单个客户端在给定时间内的请求数量来实现。
统一的异常处理程序确保在整个应用程序中提供一致的错误响应：
@app.exception_handler(HTTPException)
async def http_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)
async def global_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"])
async def root():
return RedirectResponse("/docs")
将根路径重定向到 /docs 是一个虽小但贴心的设计——任何访问API基础URL的用户都能立即看到交互式文档，使API更易于被发现和使用。
settings.py：环境配置管理
使用Pydantic的 BaseSettings 来处理环境变量异常简洁清晰。你可以为环境变量定义类型，这些类型在应用程序启动时会被验证，并为可选变量提供默认值：
# apps/settings.py
from pydantic_settings import BaseSettings
from pydantic import ConfigDict, AnyUrl
class Settings(BaseSettings):
model_config = ConfigDict(env_file=".env")
example_secret: str = "example secret value"
JWT_SECRET: str # 必需的环境变量
JWT_ALGORITHM: str = "HS256" # 可选，带默认值
这种方法能在配置错误进入生产环境之前将其捕获。如果缺少必需的环境变量，应用程序会在启动时立即失败，而不是在运行时莫名其妙地崩溃。
dependencies.py：集中式依赖管理
此模块集中管理自定义的路由依赖项。像 get_db（用于获取数据库会话）这样的常用依赖在此声明。一个独立的依赖模块有助于保持代码组织更清晰、更模块化（不同的路由通常依赖这些共享依赖），并且使测试更容易，因为逻辑可以被隔离和验证。
middlewares.py：自定义中间件中心
此模块集中管理应用程序的自定义中间件。正如依赖模块所描述的那样，模块化的优势在此同样适用。用于请求日志记录、身份验证检查或性能监控的中间件都集中在一个可预测的位置。
目录组织结构
路由器目录：路由逻辑分离
我倾向于为每个逻辑路由组创建独立的模块，这便于开发和维护。例如，auth.py 模块包含与用户认证和资料管理相关的所有路由，product.py 包含产品管理路由，admin.py 包含所有管理员API路由。
在可能的情况下，我会将路由函数的代码控制在两行以内：路由函数仅用于声明路由、定义依赖项以及指定请求参数。每个路由的核心业务逻辑则存在于对应的服务函数中：
# apps/routers/auth.py
from typing import Annotated
from pydantic import EmailStr
from fastapi.routing import APIRouter
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi import Depends, Body, BackgroundTasks
from app.dependencies import get_db
from app.schemas import auth as auth_schemas
from app.services import auth as auth_services
router = 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)
async def signup(
db: DBDep,
bg_task: BackgroundTasks,
request_data: auth_schemas.UserSignUpData,
):
return await auth_services.signup_user(request_data, db, bg_task)
这种简洁的路由设计将业务逻辑委托给服务层。路由本身只关注依赖关系、请求体或查询参数。在调试或添加功能时，你能立刻知道从何处着手——路由定义了API接口，而服务实现了具体功能。
模式目录：数据验证层
我将所有Pydantic模型都放在模式模块中。与路由目录类似，模式目录包含了所有模式模块。通常，每个路由模块会对应一个模式模块和一个服务模块。这样，每个应用组件都能以一种逻辑且可预测的方式分类。
例如，auth.py 对应的模式模型：
# apps/schemas/auth.py
from uuid import UUID
from datetime import datetime
from typing import Annotated
from pydantic import BaseModel, EmailStr, Field
class UserSignUpData(BaseModel):
password: Annotated[str, Field(min_length=8)]
email: Annotated[EmailStr, Field(max_length=254)]
class UserModel(BaseModel):
id: UUID
email: EmailStr
date_created: datetime
date_updated: datetime
Pydantic模式提供了自动验证、序列化和文档生成功能，它在API和客户端之间充当契约，确保在应用程序边界的数据完整性。
服务目录：业务逻辑实现
服务目录存放实际的业务逻辑模块——包括对第三方API的调用、数据库查询以及复杂的计算操作。通过这种方式组织，实现了路由层（声明式API需求）与具体实现之间的关注点分离。
与“路由”和“模式”目录保持一致，在组件层面也维持着一对一的对应关系。以“认证”组件为例：
# apps/services/auth.py
from datetime import timedelta
from fastapi import HTTPException
from passlib.context import CryptContext
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.settings import Settings
from app.models import User as UserDB
from app.schemas import auth as auth_schema
settings = Settings()
JWT_SECRET = settings.JWT_SECRET
JWT_ALGORITHM = settings.JWT_ALGORITHM
ACCESS_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")
def verify_password(plain_password: str, hashed_password: str):
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str):
return pwd_context.hash(password)
async def get_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()
async def create_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")
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_user
async def signup_user(
data: auth_schema.UserSignUpData,
session: AsyncSession,
):
return await create_user(data, session)
这种分层架构使得测试变得极为简便。你可以独立于业务逻辑测试路由验证，在测试路由时模拟服务函数；也可以使用测试数据库单独测试服务函数，而无需触及HTTP层。
提升开发效率：Makefile 的优势
除了核心的Python后端组件外，强烈建议使用像 Makefile 这样的工具来简化常见的命令行操作——例如启动FastAPI服务器、运行pytest以及生成代码覆盖率报告：
# Makefile
run-local:
fastapi dev app/main.py
test-local:
pytest -s --cov
coverage-report:
coverage report
coverage-html:
coverage report && coverage html
要使用make，首先需要确保系统已安装。一旦项目根目录下存在上述Makefile，启动FastAPI服务器就变得非常简单：
make run-local
无需每次都输入冗长的完整命令。对于频繁使用的操作，这能极大地提升效率。同时，它还能确保团队成员使用一致的命令——无论个人环境配置如何，所有人都使用相同的指令。
可扩展性与生产就绪性
该架构已在生产环境中得到验证，能够同时处理500多个并发用户。清晰的关注点分离意味着你可以独立地扩展不同组件。需要优化数据库查询？专注于服务模块。想要添加缓存？在依赖层进行注入。需要更换身份验证提供商？无需修改路由，直接调整身份验证服务即可。
中间件提供了多层次的安全防护：速率限制防止滥用，CORS防止未授权访问，可信主机中间件防御注入攻击，而全面的日志记录则在出现问题时提供清晰的监控视图。
总结
在这个模板中，我们有意没有包含数据库连接的具体设置部分——因为这很大程度上取决于你所选择的数据库系统（如PostgreSQL、MongoDB等）。保持项目结构和代码库的模块化所带来的好处远超预期：对功能和工具进行有条理、符合逻辑的组织，使得代码库的“可维护性”成为其最突出的优点。
虽然这一模式在过去一年极大地提升了我的开发效率，但我仍在持续学习并采纳更优的后端架构方法和实践。此模板的完整代码可在 GitHub 上获取：FastAPI Project Structure。
该架构为项目的成长预留了空间。你可以从几条路由和模式开始，然后随着应用需求的变化逐步扩展。无论是构建一个周末项目还是一个企业级系统，这种架构都能提供坚实的支持。最重要的是，它能让开发过程充满乐趣——当我们享受在代码库中工作时，我们就能创造出更优质的软件。
发表于 前天 03:31
|
查看: 6|
回复: 0
