FastAPI SDK：企业级Web服务开发的模块化脚手架与最佳实践

1. 项目概述一个为FastAPI应用量身定制的SDK如果你正在用FastAPI构建Web服务并且已经厌倦了在每个新项目中重复编写那些“样板代码”——比如统一的响应封装、全局异常处理、数据库会话管理、或是繁琐的配置加载——那么iimeta/fastapi-sdk这个项目很可能就是你一直在寻找的“瑞士军刀”。这不是一个简单的工具包而是一个旨在为FastAPI应用开发提供“开箱即用”企业级能力的软件开发工具包SDK。它的核心目标是让开发者能够从繁琐的通用基础设施中解放出来更专注于业务逻辑的创新与实现。简单来说fastapi-sdk扮演了一个“脚手架”和“最佳实践集成器”的角色。它预设了一套经过验证的、适用于中大型FastAPI项目的架构模式和工具链。当你基于这个SDK启动一个新项目时你得到的不是一个空白的画布而是一个已经搭好了主梁、布好了水电的毛坯房。你不需要再从零开始思考如何组织项目结构、如何处理跨域请求、如何优雅地记录日志或是如何安全地管理密钥。这些“非功能性需求”已经被标准化并封装好了。这个SDK特别适合那些需要快速构建标准化、可维护、且易于扩展的后端API服务的团队。无论是初创公司需要快速验证产品原型还是成熟团队需要统一内部的技术栈和开发规范fastapi-sdk都能显著降低项目的启动成本和长期维护成本。它不仅仅是代码的集合更是一种开发范式的倡导引导开发者走向更清晰、更健壮的应用架构。2. 核心架构与设计哲学拆解2.1 模块化与“即插即用”设计fastapi-sdk的核心设计思想是高度的模块化。它没有试图将你锁定在一个庞大而僵化的框架里而是将功能分解为一个个独立的、松耦合的组件。你可以根据项目的实际需要像搭积木一样选择性地引入这些组件。例如你的项目可能暂时不需要消息队列集成那么你就可以完全不引入相关模块保持项目的轻量。这种设计带来了极大的灵活性。每个模块都遵循清晰的接口定义内部实现细节被良好地封装。这意味着即使SDK内部的某个工具比如从python-dotenv切换到pydantic-settings来管理配置在未来发生了升级或替换只要对外接口保持不变你的业务代码就完全不需要修改。这种“面向接口而非实现”的设计是保障项目长期健康演进的基石。2.2 约定优于配置Convention Over Configuration为了减少开发者在项目初始化时的决策负担和配置工作量fastapi-sdk大量采用了“约定优于配置”的原则。它为项目结构、配置文件命名、环境变量前缀等制定了一套合理的默认约定。例如它可能默认约定项目的配置文件位于config/目录下根据APP_ENV环境变量加载对应的config/production.py或config/development.py。所有路由控制器Controller应放在app/apis/v1/目录下并遵循特定的命名规范。数据库模型Model集中在app/models/目录。中间件Middleware和依赖注入Dependency有专门的注册方式。当你遵循这些约定时大部分功能会自动生效无需编写冗长的配置代码。当然如果你有特殊需求几乎所有的默认约定都支持通过配置项进行覆盖。这就在“快速启动”和“深度定制”之间取得了很好的平衡。2.3 与FastAPI原生生态的深度集成fastapi-sdk并非要取代FastAPI而是作为其能力的增强和补充。它深度拥抱FastAPI的原生特性如依赖注入系统Dependency Injection、后台任务BackgroundTasks、APIRouter等。SDK提供的很多功能都是通过编写高质量的FastAPI依赖项、中间件或生命周期事件处理器来实现的。例如SDK可能提供一个get_db依赖项它封装了数据库会话的创建、提交和异常回滚逻辑。在你的路由处理函数中你只需要声明这个依赖就可以安全地使用数据库会话完全不用操心资源泄露或事务处理的问题。这种集成方式使得SDK的使用体验非常自然感觉就像是FastAPI本身功能的一部分。3. 核心功能模块深度解析3.1 统一响应封装与异常处理这是提升API开发者体验和客户端体验最直接的功能。一个未经处理的FastAPI视图在成功时可能直接返回一个Python字典或Pydantic模型在出错时则抛出HTTPException。这会导致客户端需要处理多种不同结构的响应。fastapi-sdk通过一个自定义的APIRouter和全局异常处理器强制所有接口返回统一的结构。一个典型的成功响应可能如下所示{ code: 200, message: success, data: { ... // 你的业务数据 }, timestamp: 1678886400 }而一个业务逻辑错误的响应则可能是{ code: 10001, message: 用户不存在, data: null, timestamp: 1678886400 }实现原理SDK通常会创建一个继承自fastapi.APIRouter的自定义路由器如BaseRouter。在这个自定义路由器的add_api_route方法中对路由处理函数的返回值进行包装。同时通过app.add_exception_handler注册一个全局异常处理器捕获HTTPException、ValidationError以及自定义的业务异常并将它们转换为上述统一的错误格式。实操心得统一响应格式后前端团队会非常感激你。但要注意对于文件下载、Server-Sent Events (SSE) 或 WebSocket 这类特殊接口需要将它们排除在统一包装器之外通常可以通过路由装饰器的额外参数如response_modelNone或白名单机制来实现。3.2 增强型配置管理管理不同环境开发、测试、生产的配置是项目的基石。fastapi-sdk的配置管理模块通常基于pydantic-settings构建这带来了两大好处类型安全和环境变量自动加载。核心工作流程定义配置模型使用Pydantic模型定义所有配置项包括类型、默认值和验证规则。from pydantic_settings import BaseSettings class Settings(BaseSettings): app_name: str My FastAPI App database_url: str redis_url: Optional[str] None debug: bool False class Config: env_file .env env_prefix APP_ # 环境变量需要以 APP_ 开头环境隔离通过APP_ENVproduction这样的环境变量动态加载对应的配置文件实现配置覆盖。全局访问通过一个单例或依赖注入的方式让配置对象在应用内随处可安全访问。优势你不再需要手动解析.env文件或os.getenv。所有配置都有明确的类型IDE可以提供自动补全和类型检查。配置项的来源环境变量、.env文件、默认值和优先级清晰可控。3.3 数据库集成与会话管理fastapi-sdk通常会集成SQLAlchemy用于关系型数据库和Alembic用于数据库迁移并提供一套“最佳实践”的使用模式。核心封装会话工厂与依赖注入SDK会创建一个全局的SessionLocal工厂并提供一个get_db依赖项。这个依赖项确保每个请求获得一个独立的数据库会话并在请求结束后自动关闭防止会话泄露。async def get_db(): db SessionLocal() try: yield db finally: db.close()Repository模式可选但推荐为了进一步解耦业务逻辑和数据库操作SDK可能鼓励或提供Repository层的抽象。这层抽象将具体的SQLAlchemy查询封装起来业务层通过Repository接口与数据库交互使得数据访问逻辑更易于测试和维护。自动化迁移集成Alembic并提供简单的命令行指令来生成和管理数据库迁移脚本。注意事项在异步FastAPI应用中使用同步的SQLAlchemy时需要特别注意。虽然get_db依赖是async def但SQLAlchemy的核心操作是同步的。在路由函数内执行耗时较长的数据库查询可能会阻塞事件循环。对于高性能场景应考虑使用asyncpg驱动配合SQLAlchemy的异步API或者将耗时操作移交到后台线程池。3.4 认证与授权中间件安全是API的重中之重。fastapi-sdk通常会提供基于JWTJSON Web Token的认证方案。典型流程用户登录服务端验证凭证后使用密钥从配置中读取签发一个JWT令牌返回给客户端。客户端在后续请求的Authorization请求头中携带此令牌格式Bearer token。SDK提供的认证中间件或依赖项会自动验证令牌的签名和有效期。如果令牌有效中间件会将解码出的用户信息如user_id存入请求状态request.state或直接作为依赖项的返回值供路由处理函数使用。进阶功能基于角色的访问控制RBAC。SDK可能允许你在路由上通过装饰器或元数据定义所需的权限角色例如router.get(/admin, dependencies[Depends(requires_role(admin))])。认证依赖项在验证JWT后会进一步检查当前用户是否拥有所需角色。安全要点密钥管理签名密钥必须足够复杂且绝不能硬编码在代码中应从环境变量或安全的密钥管理服务获取。令牌过期必须为JWT设置合理的短有效期如15-30分钟并通过刷新令牌Refresh Token机制来平衡安全性与用户体验。令牌注销标准的JWT本身是无状态的服务端无法主动使其失效。如需实现“立即踢下线”功能需要引入令牌黑名单或使用状态化的会话方案这通常会牺牲一部分JWT的无状态优势。3.5 结构化日志与监控“出了问题才知道日志的重要性”是很多开发者的血泪教训。fastapi-sdk的日志模块旨在提供生产就绪的日志能力。核心特性JSON格式化不同于默认的文本日志SDK可能将日志输出为JSON格式。这使得日志更容易被日志收集系统如ELK Stack、Loki解析和索引。请求ID追踪为每个进入系统的请求生成一个唯一的ID如UUID并将这个ID记录在该请求生命周期内产生的所有日志中。这是排查复杂分布式问题的关键。集成上下文自动在日志记录中附加请求路径、客户端IP、用户ID如果已认证等信息。性能监控可能通过中间件记录每个请求的处理耗时、状态码并暴露给Prometheus等监控系统生成API的延迟和成功率指标。配置示例SDK可能会使用structlog库来增强日志能力。配置完成后你只需要logger.info(user logged in, user_iduser.id)就能得到一条包含时间戳、日志级别、消息体以及丰富上下文的JSON日志。3.6 实用工具集除了上述核心模块SDK通常还会包含一系列“锦上添花”的实用工具进一步提效异步任务队列集成提供与Celery或ARQ的简易集成模式方便你将耗时任务如发送邮件、处理图片异步化。缓存抽象层提供对Redis或内存缓存的统一操作接口简化缓存的使用和切换。健康检查端点自动添加/health或/ready端点用于负载均衡器或Kubernetes的存活性和就绪性探针。API文档增强自动为Swagger UI或ReDoc文档分组、添加标签或设置统一的认证方式。4. 从零开始使用fastapi-sdk搭建一个用户管理系统4.1 项目初始化与环境搭建假设我们的项目名为user-service。创建虚拟环境与安装python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows pip install fastapi-sdk # 假设sdk已发布到PyPI pip install pydantic-settings sqlalchemy alembic python-jose[cryptography] passlib[bcrypt]使用SDK脚手架如果提供 许多SDK会提供一个命令行工具来快速生成项目骨架。fastapi-sdk init user-service --template standard cd user-service这会生成一个包含标准目录结构、基础配置和示例代码的项目。核心目录结构脚手架生成或手动创建user-service/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI应用创建和组装入口 │ ├── core/ # 核心配置、依赖、安全等 │ │ ├── config.py │ │ ├── dependencies.py │ │ └── security.py │ ├── models/ # SQLAlchemy 数据模型 │ │ └── user.py │ ├── schemas/ # Pydantic 请求/响应模型 │ │ ├── user.py │ │ └── token.py │ ├── crud/ # 数据库操作层 (Repository模式) │ │ └── user.py │ ├── apis/ # 路由层 │ │ ├── __init__.py │ │ └── v1/ # API版本v1 │ │ ├── __init__.py │ │ ├── endpoints/ │ │ │ ├── login.py │ │ │ └── users.py │ │ └── router.py # 聚合v1所有路由 │ └── utils/ # 工具函数 │ └── logger.py ├── alembic/ # 数据库迁移脚本 ├── tests/ # 测试用例 ├── .env.example # 环境变量示例文件 ├── .env # 本地环境变量被.gitignore忽略 ├── requirements.txt └── pyproject.toml4.2 配置与数据库模型定义配置 (app/core/config.py)from pydantic_settings import BaseSettings from typing import Optional class Settings(BaseSettings): PROJECT_NAME: str User Service API_V1_STR: str /api/v1 SECRET_KEY: str # 用于JWT签名必须从环境变量设置 ALGORITHM: str HS256 ACCESS_TOKEN_EXPIRE_MINUTES: int 30 DATABASE_URL: str # 如postgresql://user:passlocalhost/dbname FIRST_SUPERUSER: str FIRST_SUPERUSER_PASSWORD: str class Config: env_file .env env_prefix USER_SERVICE_ settings Settings()数据库模型 (app/models/user.py)from sqlalchemy import Boolean, Column, Integer, String from app.core.database import Base # Base来自SDK或自定义 class User(Base): __tablename__ users id Column(Integer, primary_keyTrue, indexTrue) email Column(String, uniqueTrue, indexTrue, nullableFalse) hashed_password Column(String, nullableFalse) full_name Column(String) is_active Column(Boolean(), defaultTrue) is_superuser Column(Boolean(), defaultFalse)Pydantic模式 (app/schemas/user.py)from pydantic import BaseModel, EmailStr from typing import Optional class UserBase(BaseModel): email: Optional[EmailStr] None full_name: Optional[str] None is_active: Optional[bool] True class UserCreate(UserBase): email: EmailStr password: str class UserInDB(UserBase): id: int class Config: from_attributes True # 原orm_mode支持从ORM对象读取4.3 实现核心业务逻辑用户注册与登录密码工具 (app/core/security.py)from passlib.context import CryptContext from jose import JWTError, jwt from datetime import datetime, timedelta from app.core.config import settings pwd_context CryptContext(schemes[bcrypt], deprecatedauto) def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): return pwd_context.hash(password) def create_access_token(data: dict, expires_delta: Optional[timedelta] None): to_encode data.copy() expire datetime.utcnow() (expires_delta or timedelta(minutessettings.ACCESS_TOKEN_EXPIRE_MINUTES)) to_encode.update({exp: expire}) encoded_jwt jwt.encode(to_encode, settings.SECRET_KEY, algorithmsettings.ALGORITHM) return encoded_jwt数据访问层 (app/crud/user.py)from sqlalchemy.orm import Session from app.models.user import User from app.schemas.user import UserCreate from app.core.security import get_password_hash def get_user_by_email(db: Session, email: str): return db.query(User).filter(User.email email).first() def create_user(db: Session, user_in: UserCreate): hashed_password get_password_hash(user_in.password) db_user User( emailuser_in.email, hashed_passwordhashed_password, full_nameuser_in.full_name, ) db.add(db_user) db.commit() db.refresh(db_user) return db_user认证依赖项 (app/core/dependencies.py)from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from jose import JWTError, jwt from sqlalchemy.orm import Session from app.core.config import settings from app.crud.user import get_user_by_email from app.core.database import get_db # 来自SDK的get_db oauth2_scheme OAuth2PasswordBearer(tokenUrlf{settings.API_V1_STR}/login) async def get_current_user( db: Session Depends(get_db), token: str Depends(oauth2_scheme) ): credentials_exception HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail无法验证凭证, headers{WWW-Authenticate: Bearer}, ) try: payload jwt.decode(token, settings.SECRET_KEY, algorithms[settings.ALGORITHM]) email: str payload.get(sub) if email is None: raise credentials_exception except JWTError: raise credentials_exception user get_user_by_email(db, emailemail) if user is None: raise credentials_exception return user路由端点 (app/apis/v1/endpoints/login.py)from fastapi import APIRouter, Depends, HTTPException from fastapi.security import OAuth2PasswordRequestForm from sqlalchemy.orm import Session from datetime import timedelta from app.core import security from app.core.dependencies import get_db from app.crud.user import get_user_by_email from app.schemas.token import Token router APIRouter() router.post(/login, response_modelToken) async def login_access_token( db: Session Depends(get_db), form_data: OAuth2PasswordRequestForm Depends() ): user get_user_by_email(db, emailform_data.username) if not user or not security.verify_password(form_data.password, user.hashed_password): raise HTTPException(status_code400, detail邮箱或密码错误) access_token_expires timedelta(minutes30) access_token security.create_access_token( data{sub: user.email}, expires_deltaaccess_token_expires ) return {access_token: access_token, token_type: bearer}4.4 应用组装与启动主应用文件 (app/main.py)from fastapi import FastAPI from app.core.config import settings from app.core.logger import setup_logging from app.apis.v1.router import api_router # 聚合了所有v1路由 from app.core.database import engine, Base from app.core.middleware import add_middleware # SDK提供的中间件设置函数 # 创建表生产环境应使用Alembic迁移 Base.metadata.create_all(bindengine) # 初始化应用 app FastAPI(titlesettings.PROJECT_NAME) # 设置日志来自SDK或自定义 setup_logging() # 添加全局中间件如CORS、请求ID、日志 add_middleware(app) # 挂载API路由 app.include_router(api_router, prefixsettings.API_V1_STR) # 健康检查端点可由SDK提供 app.get(/health) def health_check(): return {status: healthy}至此一个具备统一响应、JWT认证、结构化日志、数据库ORM等企业级特性的用户管理API服务就搭建完成了。你可以通过运行uvicorn app.main:app --reload启动开发服务器。5. 生产环境部署与优化指南5.1 配置管理与安全绝对禁止将敏感信息如SECRET_KEY、数据库密码写入代码或提交到版本库。.env文件必须列入.gitignore。生产环境配置建议使用环境变量在Docker容器或服务器上直接设置环境变量这是最安全、最灵活的方式。密钥管理服务对于云原生部署使用AWS Secrets Manager、Azure Key Vault或HashiCorp Vault等专业服务管理密钥。配置文件分离将配置按环境production.py,staging.py分离并通过APP_ENV环境变量动态加载。5.2 性能与可扩展性数据库连接池确保SQLAlchemy配置了合适的连接池大小pool_size,max_overflow避免连接耗尽或浪费。异步化评估将核心IO操作如数据库查询、外部API调用异步化的必要性。对于高并发场景使用asyncpgsqlalchemy[asyncio]或databases库可以大幅提升吞吐量。缓存策略对频繁读取、很少变化的数据如用户信息、配置项使用Redis进行缓存减轻数据库压力。静态文件服务使用Nginx或CDN来服务静态文件不要让FastAPI处理静态资源。5.3 监控与告警指标暴露使用prometheus-fastapi-instrumentator等中间件将请求耗时、状态码等指标暴露给Prometheus。集中式日志确保日志以JSON格式输出并使用Fluentd、Filebeat等日志收集器将日志发送到ELK或Loki等集中式日志系统。健康检查除了基础的/health实现一个有业务意义的就绪探针如/ready检查数据库、缓存等下游依赖是否正常。应用性能管理(APM)集成如OpenTelemetry的分布式追踪或使用商业APM工具如Datadog, New Relic来监控应用性能。5.4 容器化部署示例 (Docker)一个简单的Dockerfile示例如下FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . # 设置环境变量确保使用生产配置 ENV APP_ENVproduction CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 8000, --workers, 4]使用docker-compose.yml可以方便地编排应用、数据库和缓存服务。6. 常见问题排查与调试技巧6.1 依赖注入与上下文管理问题问题现象在后台任务、事件处理器或测试中调用依赖项如get_db时报错提示“上下文”或“事件循环”相关错误。根因分析FastAPI的依赖注入系统与请求上下文紧密绑定。get_db这类依赖通常设计为在每个请求生命周期内工作它们依赖于特定的请求上下文。在请求上下文之外如Celery任务、asyncio.create_task创建的后台任务、或普通的单元测试中这个上下文是不存在的。解决方案对于后台任务不要在任务函数内部直接使用Depends(get_db)。应该将数据库会话或会话工厂作为参数显式传递给任务函数。或者在任务函数内部手动创建和管理一个新的会话上下文。# 错误示例在后台任务中 async def background_task(user_id: int, db: Session Depends(get_db)): ... # 正确示例 from app.core.database import SessionLocal async def background_task(user_id: int): db SessionLocal() try: # ... 使用db进行操作 pass finally: db.close()对于单元测试使用pytest配合pytest-asyncio。在测试用例中你可以使用TestClient来模拟请求或者手动创建应用并覆盖依赖项。from fastapi.testclient import TestClient from app.main import app client TestClient(app) response client.post(/login, data{username: ..., password: ...})6.2 数据库会话与事务管理混乱问题现象出现“此会话正在被其他事务使用”错误或数据修改未按预期提交/回滚。核心原则一个请求一个会话。确保get_db依赖为每个请求创建独立的新会话并在请求结束时关闭它。常见陷阱与技巧不要在全局或模块级别缓存会话绝对不要将SessionLocal()创建的会话赋值给一个全局变量并在多个请求间共享。正确处理嵌套依赖如果依赖项A又依赖于get_dbFastAPI会智能地处理通常不会创建多个会话。但要确保依赖树清晰。手动事务控制对于复杂的业务逻辑可能需要手动控制事务边界。可以使用db.begin()开启嵌套事务但要小心处理回滚逻辑。try: db.begin_nested() # 如果需要保存点 # 一系列操作... db.commit() except Exception: db.rollback() raise6.3 JWT认证相关故障问题排查清单问题可能原因解决方案返回401 Unauthorized1. 请求头未携带Authorization: Bearer token。2. Token已过期。3. Token签名无效密钥不匹配。4. Token解码失败算法不一致。1. 检查前端请求头。2. 检查Token中的exp声明。3. 确认服务端SECRET_KEY与签发时一致。4. 确认ALGORITHM配置一致。登录成功但后续请求认证失败用户状态在签发Token后发生改变如被禁用但Token未更新。在get_current_user依赖中增加对用户状态的检查如if not user.is_active。或在Token中嵌入版本号用户状态改变时使其版本号失效。无法实现“立即踢下线”JWT本身是无状态的服务端无法主动作废未过期的Token。实现令牌黑名单将作废的Token ID存入Redis并设置过期时间在认证时检查黑名单。这会引入状态牺牲部分JWT优势。6.4 日志不输出或格式错误检查步骤日志级别设置确认环境变量或配置中设置的日志级别如LOG_LEVELINFO低于你期望输出的级别如DEBUG。JSON格式问题如果使用JSON日志确保没有非序列化的对象被传入日志记录。structlog的处理器配置是否正确。多进程/Worker模式在使用uvicorn --workers 4时确保日志配置能正确处理多进程写入或者将日志发送到系统日志如syslog或网络服务。6.5 跨域请求CORS问题问题现象前端应用调用API时浏览器控制台报CORS错误。解决方案确保在FastAPI应用中正确配置了CORS中间件。fastapi-sdk通常会在其中间件模块中提供便捷的配置方法。from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_origins[https://your-frontend.com], # 生产环境应具体指定 allow_credentialsTrue, allow_methods[*], allow_headers[*], )重要提示在生产环境中切勿将allow_origins设置为[*]这会带来严重的安全风险。应明确列出允许的前端域名。7. 进阶扩展SDK与定制化开发fastapi-sdk的强大之处在于其可扩展性。当你需要集成SDK尚未包含的组件时可以遵循其设计模式进行平滑集成。7.1 集成第三方服务以发送邮件为例在配置中添加在Settings模型中添加邮件服务器的配置项。class Settings(BaseSettings): # ... 其他配置 SMTP_SERVER: str SMTP_PORT: int SMTP_USERNAME: str SMTP_PASSWORD: str EMAILS_FROM: str创建服务模块在app/utils/下创建email.py封装发送邮件的逻辑。import smtplib from email.mime.text import MIMEText from app.core.config import settings def send_email(to: str, subject: str, body: str): msg MIMEText(body) msg[Subject] subject msg[From] settings.EMAILS_FROM msg[To] to with smtplib.SMTP(settings.SMTP_SERVER, settings.SMTP_PORT) as server: server.login(settings.SMTP_USERNAME, settings.SMTP_PASSWORD) server.send_message(msg)作为依赖项使用你可以创建一个get_email_sender依赖项返回上面的发送函数方便测试时模拟Mock。7.2 编写自定义中间件假设你需要一个中间件来记录每个请求的详细耗时和状态码。import time from fastapi import Request from app.utils.logger import logger async def request_logging_middleware(request: Request, call_next): start_time time.time() response await call_next(request) process_time time.time() - start_time # 使用结构化日志记录 logger.info( request_finished, pathrequest.url.path, methodrequest.method, status_coderesponse.status_code, process_timeprocess_time, client_iprequest.client.host ) # 也可以将耗时添加到响应头 response.headers[X-Process-Time] str(process_time) return response然后在主应用创建后通过app.middleware(http)装饰器或add_middleware函数将其添加到应用中。7.3 创建自定义APIRouter如果你有一组相关的接口需要共享相同的路径前缀、标签、依赖项或响应模型可以创建自定义的Router。from fastapi import APIRouter, Depends from app.core.dependencies import get_current_active_user class AdminRouter(APIRouter): def __init__(self): super().__init__( prefix/admin, tags[admin], dependencies[Depends(get_current_active_user)], # 所有/admin路由都需要登录 responses{404: {description: Not found}}, ) # 使用 admin_router AdminRouter() admin_router.get(/users/) def get_all_users(): ...这种方式能让你的路由组织更加清晰和符合DRYDon‘t Repeat Yourself原则。通过以上步骤你不仅能够熟练使用fastapi-sdk来加速项目开发更能理解其背后的设计思想并具备根据自身业务需求对其进行定制和扩展的能力。记住任何SDK或框架都是工具最终的目标是高效、稳定地实现业务价值。fastapi-sdk为你铺好了道路而如何走好这条路则取决于你对业务的理解和代码的设计。