博客 / 詳情

返回

FastAPI 項目架構指南

本文介紹了在 Python 項目中使用 FastAPI 構建產品的代碼架構設計模式,通過良好的代碼架構,可以清晰的組織代碼功能,有助於開發功能良好的產品。原文:FastAPI Architecture Guide: Build Scalable and Secure Systems with This Production-Ready Template

在生產環境中運行這個架構之後,可以自信的説,該架構可以輕鬆擴容、很好的進行維護,並且可以保持生產力。如果想在項目中直接使用這個 FastAPI 架構,請查看 GitHub 的完整項目設置:FastAPI Project Structure

簡介

在過去一年裏,我幾乎每天都用 FastAPI。我在這段時間裏開發了一個標準的項目架構,幫助我輕鬆應對變更和添加新功能,同時又能很好的擴展。本文分解該 FastAPI 架構的每個組件。無論你希望更快迭代想法,組織不斷增長的代碼庫,還是加強應用程序安全性,此架構都將幫助你自信的構建並像專家一樣發展項目。

項目架構概述

中間件體系架構模式,具有雙向連接多個應用程序和數據庫的集成平台

一個組織良好的 FastAPI 項目的魅力在於其可預測性。當每個組件都有其合理的歸屬位置時,就能節省尋找文件的時間,而將更多時間用於構建功能。這種架構清晰劃分了各個職責:路由邏輯留在路由器中,業務邏輯存在於服務中,數據驗證在模式中進行,配置則集中於專門的模塊中。

App 目錄:你的應用引擎

App 目錄是應用程序的引擎和核心,包含支持大多數後端基礎設施用例所需的所有模塊和代碼。下面解釋該目錄中的每個子目錄和模塊、使用它們的原因、要避免的常見陷阱,以及為提高代碼質量和開發人員體驗所做的改進。

__init__.py:包聲明

在每個 Python 包中放置 __init__.py 是標準做法,它將該文件夾聲明為常規 Python 包,從而在整個應用程序中實現適當的導入。

api_router.py:中心化路由管理

這個模塊作為管理應用程序中所有路由的中心化組件,有助於版本控制(如 /v1/users/v2/users),並保持代碼庫清晰易讀。

# apps/api_router.py

from fastapi import APIRouter
api_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)

這種中心化方法意味着,在需要引入 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 限速圖示,當超過請求限制時,客户端請求被限速器阻止並顯示 429 Too Many Requests 錯誤

主模塊是應用程序入口,其中包含了大部分安全策略和規則。在這裏,中間件會依次疊加起來,以保護 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 = [
        # Add allowed origins here
    ]
    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=[
            # Add allowed hosts here
        ],
    )
    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  # required environment variable 
    JWT_ALGORITHM: str = "HS256"  # optional with default

這種方法能在配置錯誤到達生產環境之前將其捕獲。如果缺少必需的環境變量,應用程序在啓動時會立即失敗,而不是在運行時莫名其妙地出現故障。

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 上獲取:https://github.com/brianobot/fastAPI_project_structure

該架構為項目發展提供了空間。先從小規模開始,設置幾條路由和模式,然後隨着應用需求的變化逐步擴展。無論是構建週末項目還是企業系統,這種架構都能提供支持。最重要的是,能讓開發過程充滿樂趣 —— 當我們享受在代碼庫中工作時,就能開發出更優質的軟件。


你好,我是俞凡,在Motorola做過研發,現在在Mavenir做技術工作,對通信、網絡、後端架構、雲原生、DevOps、CICD、區塊鏈、AI等技術始終保持着濃厚的興趣,平時喜歡閲讀、思考,相信持續學習、終身成長,歡迎一起交流學習。為了方便大家以後能第一時間看到文章,請朋友們關注公眾號"DeepNoMind",並設個星標吧,如果能一鍵三連(轉發、點贊、在看),則能給我帶來更多的支持和動力,激勵我持續寫下去,和大家共同成長進步!

本文由mdnice多平台發佈

user avatar pudongping 頭像 yingyongwubideshuitong 頭像 turing_interview 頭像 jianhuan 頭像 TwilightLemon 頭像 haoqingwanqiandehongcha 頭像 juicefs 頭像 codelogs 頭像 insus 頭像 u_12190 頭像 chuck1sn 頭像 tencent_yinshipinxiaojiejie 頭像
31 位用戶收藏了這個故事!

發佈 評論

Some HTML is okay.