FastAPI 生产级项目脚手架模板指南async 模式、依赖注入与分层架构实践【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents本文以 agents 仓库 api-scaffolding 插件下的fastapi-templates技能文档为主体完整讲解如何用 FastAPI 搭建生产级后端从目录分层、依赖注入到 Repository/Service 分层实现、JWT 认证与异步测试。读者阅读后可掌握一套可直接复制落地的工程骨架并通过仓库源码佐证每个模式的实现细节。技能定位何时使用 fastapi-templatesfastapi-templates 是 api-scaffolding 插件中负责 FastAPI 项目初始化的技能其 YAML 头部声明了明确的激活语义Create production-ready FastAPI projects with async patterns, dependency injection, and comprehensive error handling. Use when building new FastAPI applications or setting up backend API projects.按 文档 与技能规范它适合在以下场景被自动触发从零起步搭建新的 FastAPI 项目用 Python 实现 async REST API构建高性能 Web 服务与微服务创建对接 PostgreSQL、MongoDB 的异步应用需要带规范结构与完整测试的 API 工程初始化。在同插件 agents/fastapi-pro.md 的视角中该技能与fastapi-pro智能体互补Agent 负责高层编排技能则提供“开箱即用的模板与模式”。从仓库的插件注册文件marketplace.json可看到 api-scaffolding 插件及目录整体被纳入 marketplace 供各 harness 使用即该技能实际是以“渐进式披露”方式按需加载的领域知识包。一、生产级项目目录结构技能文档给出了推荐布局核心思想是把路由、核心配置、模型、Schema、业务逻辑、数据访问严格分层并内置版本化 APIv1app/ ├── api/ # API routes │ ├── v1/ │ │ ├── endpoints/ │ │ │ ├── users.py │ │ │ ├── auth.py │ │ │ └── items.py │ │ └── router.py │ └── dependencies.py # Shared dependencies ├── core/ # Core configuration │ ├── config.py │ ├── security.py │ └── database.py ├── models/ # Database models │ ├── user.py │ └── item.py ├── schemas/ # Pydantic schemas │ ├── user.py │ └── item.py ├── services/ # Business logic │ ├── user_service.py │ └── auth_service.py ├── repositories/ # Data access │ ├── user_repository.py │ └── item_repository.py └── main.py # Application entry分层职责划分要点目录职责约束api/v1/endpointsHTTP 路由、状态码、参数校验只做“翻译”不写业务api/dependencies.py共享依赖当前用户、DB 会话可被多个路由复用core配置、数据库引擎、安全工具全局单例modelsSQLAlchemy ORM 模型与数据库表一一对应schemasPydantic 校验/序列化模型API 对外契约services业务规则、密码哈希、事务编排依赖 repositoryrepositories数据访问与查询依赖 sessionmain.py应用入口、lifespan、中间件、路由挂载只做组装二、应用入口lifespan、CORS 与路由挂载1. 完整的 main.py 应用骨架references/details.md的Pattern 1给出一个完整 FastAPI 应用# main.py from fastapi import FastAPI, Depends from fastapi.middleware.cors import CORSMiddleware from contextlib import asynccontextmanager asynccontextmanager async def lifespan(app: FastAPI): Application lifespan events. # Startup await database.connect() yield # Shutdown await database.disconnect() app FastAPI( titleAPI Template, version1.0.0, lifespanlifespan ) # CORS middleware app.add_middleware( CORSMiddleware, allow_origins[*], allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # Include routers from app.api.v1.router import api_router app.include_router(api_router, prefix/api/v1)设计说明lifespan 上下文统一承载启动/关闭钩子在yield之前执行数据库连接等初始化在yield之后执行清理。相比已废弃的app.on_eventlifespan 是更受推荐的生命周期方案CORS 中间件中的allow_origins[*]仅为模板演示值生产环境应替换为真实的受信域名白名单allow_credentialsTrue与*组合在带 Cookie 的场景下会受限需按实际情况取舍子路由统一通过api_router聚合并以prefix/api/v1挂载天然形成版本化 API。2. 配置管理Pydantic Settings lru_cache# core/config.py from pydantic_settings import BaseSettings from functools import lru_cache class Settings(BaseSettings): Application settings. DATABASE_URL: str SECRET_KEY: str ACCESS_TOKEN_EXPIRE_MINUTES: int 30 API_V1_STR: str /api/v1 class Config: env_file .env lru_cache() def get_settings() - Settings: return Settings()要点必填项DATABASE_URL、SECRET_KEY不带默认值缺失时启动即报错避免“带病上线”带默认值的可选配置ACCESS_TOKEN_EXPIRE_MINUTES30默认 30 分钟、API_V1_STR/api/v1.env通过Config.env_file指定字段名与系统环境变量对齐时自动注入lru_cache()保证全进程只解析一次配置重复调用get_settings()命中缓存是高频调用路径上的标准优化。3. 异步数据库会话core/database.py# core/database.py from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from app.core.config import get_settings settings get_settings() engine create_async_engine( settings.DATABASE_URL, echoTrue, futureTrue ) AsyncSessionLocal sessionmaker( engine, class_AsyncSession, expire_on_commitFalse ) Base declarative_base() async def get_db() - AsyncSession: Dependency for database session. async with AsyncSessionLocal() as session: try: yield session await session.commit() except Exception: await session.rollback() raise finally: await session.close()要点异步引擎 AsyncSession配合asyncpg/aiosqlite等异步驱动使用避免阻塞事件循环expire_on_commitFalse防止提交后属性过期导致懒加载触发额外 IOget_db()是贯穿全项目的核心依赖以async with管理会话生命周期正常路径提交、异常路径回滚finally中确保关闭。这个依赖将作为Depends(get_db)注入到每个需要访问数据库的路由与鉴权逻辑中。三、依赖注入Dependency InjectionFastAPI 内建的 DI 体系以Depends为核心技能文档将其归纳为四类典型应用数据库会话管理Depends(get_db)注入AsyncSession认证 / 授权Depends(get_current_user)注入当前登录用户共享业务逻辑将跨路由复用的服务能力抽成依赖配置注入通过依赖函数暴露Settings。依赖注入的价值在于路由声明式表达“我需要什么”框架负责组装测试时可借助app.dependency_overrides无缝替换依赖下文测试章节会展示实现真正的可测性。四、Repository 模式泛型 CRUD 基类references/details.md的Pattern 2用 PythonGeneric类型变量抽象出数据库无关的 CRUD 基类# repositories/base_repository.py from typing import Generic, TypeVar, Type, Optional, List from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from pydantic import BaseModel ModelType TypeVar(ModelType) CreateSchemaType TypeVar(CreateSchemaType, boundBaseModel) UpdateSchemaType TypeVar(UpdateSchemaType, boundBaseModel) class BaseRepository(Generic[ModelType, CreateSchemaType, UpdateSchemaType]): Base repository for CRUD operations. def __init__(self, model: Type[ModelType]): self.model model async def get(self, db: AsyncSession, id: int) - Optional[ModelType]: Get by ID. result await db.execute( select(self.model).where(self.model.id id) ) return result.scalars().first() async def get_multi( self, db: AsyncSession, skip: int 0, limit: int 100 ) - List[ModelType]: Get multiple records. result await db.execute( select(self.model).offset(skip).limit(limit) ) return result.scalars().all() async def create( self, db: AsyncSession, obj_in: CreateSchemaType ) - ModelType: Create new record. db_obj self.model(**obj_in.dict()) db.add(db_obj) await db.flush() await db.refresh(db_obj) return db_obj async def update( self, db: AsyncSession, db_obj: ModelType, obj_in: UpdateSchemaType ) - ModelType: Update record. update_data obj_in.dict(exclude_unsetTrue) for field, value in update_data.items(): setattr(db_obj, field, value) await db.flush() await db.refresh(db_obj) return db_obj async def delete(self, db: AsyncSession, id: int) - bool: Delete record. obj await self.get(db, id) if obj: await db.delete(obj) return True return False实现细节解读三个类型变量ModelType对应 ORM 模型、CreateSchemaType/UpdateSchemaType约束为 PydanticBaseModel子类为全文件提供静态类型检查更新语义obj_in.dict(exclude_unsetTrue)只取客户端显式传入的字段天然支持 PATCH 部分更新刷新时机flush()将变更发给数据库以拿到自增主键refresh()再回读服务端生成值如created_at、默认值之所以使用flush()而非commit()是把提交权上交给get_db()依赖统一管理——事务边界在会话层而非零散散落在各 repository。在此基础上扩展具体实体# repositories/user_repository.py from app.repositories.base_repository import BaseRepository from app.models.user import User from app.schemas.user import UserCreate, UserUpdate class UserRepository(BaseRepository[User, UserCreate, UserUpdate]): User-specific repository. async def get_by_email(self, db: AsyncSession, email: str) - Optional[User]: Get user by email. result await db.execute( select(User).where(User.email email) ) return result.scalars().first() async def is_active(self, db: AsyncSession, user_id: int) - bool: Check if user is active. user await self.get(db, user_id) return user.is_active if user else False user_repository UserRepository(User)五、Service 层业务逻辑与安全边界references/details.md的Pattern 3演示业务规则放在 service 而非 endpoint保证“瘦路由、胖服务”# services/user_service.py from typing import Optional from sqlalchemy.ext.asyncio import AsyncSession from app.repositories.user_repository import user_repository from app.schemas.user import UserCreate, UserUpdate, User from app.core.security import get_password_hash, verify_password class UserService: Business logic for users. def __init__(self): self.repository user_repository async def create_user( self, db: AsyncSession, user_in: UserCreate ) - User: Create new user with hashed password. # Check if email exists existing await self.repository.get_by_email(db, user_in.email) if existing: raise ValueError(Email already registered) # Hash password user_in_dict user_in.dict() user_in_dict[hashed_password] get_password_hash(user_in_dict.pop(password)) # Create user user await self.repository.create(db, UserCreate(**user_in_dict)) return user async def authenticate( self, db: AsyncSession, email: str, password: str ) - Optional[User]: Authenticate user. user await self.repository.get_by_email(db, email) if not user: return None if not verify_password(password, user.hashed_password): return None return user async def update_user( self, db: AsyncSession, user_id: int, user_in: UserUpdate ) - Optional[User]: Update user. user await self.repository.get(db, user_id) if not user: return None if user_in.password: user_in_dict user_in.dict(exclude_unsetTrue) user_in_dict[hashed_password] get_password_hash( user_in_dict.pop(password) ) user_in UserUpdate(**user_in_dict) return await self.repository.update(db, user, user_in)关键安全实践密码永不明文入库create_user先把password弹出并哈希为hashed_password再落库update_user亦只在提交了新密码时才重新哈希邮箱唯一性检查前置重复注册抛ValueError由路由层捕获并翻译成 HTTP 400认证逻辑集中authenticate统一实现“查用户 → 验密码”流程供登录接口与未来所有需要校验凭证的入口复用。六、安全组件JWT 签发与密码哈希references/details.md的Pattern 5前半部分给出core/security.py# core/security.py from datetime import datetime, timedelta from typing import Optional from jose import JWTError, jwt from passlib.context import CryptContext from app.core.config import get_settings settings get_settings() pwd_context CryptContext(schemes[bcrypt], deprecatedauto) ALGORITHM HS256 def create_access_token(data: dict, expires_delta: Optional[timedelta] None): Create JWT access token. to_encode data.copy() if expires_delta: expire datetime.utcnow() expires_delta else: expire datetime.utcnow() timedelta(minutes15) to_encode.update({exp: expire}) encoded_jwt jwt.encode(to_encode, settings.SECRET_KEY, algorithmALGORITHM) return encoded_jwt def verify_password(plain_password: str, hashed_password: str) - bool: Verify password against hash. return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password: str) - str: Hash password. return pwd_context.hash(password)要点bcrypt 密码哈希CryptContext(schemes[bcrypt], deprecatedauto)deprecatedauto允许后续无缝迁移到更强的哈希算法JWTHS256create_access_token将sub等载荷复制后注入exp过期时间未显式传expires_delta时默认 15 分钟算法与密钥统一读取自SettingsSECRET_KEY、ACCESS_TOKEN_EXPIRE_MINUTES保证敏感信息不进代码库。七、API 端点与鉴权依赖1. 依赖层从 Bearer Token 还原当前用户references/details.md的Pattern 5后半部分定义全局鉴权依赖api/dependencies.py# api/dependencies.py from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from jose import JWTError, jwt from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import get_db from app.core.security import ALGORITHM from app.core.config import get_settings from app.repositories.user_repository import user_repository oauth2_scheme OAuth2PasswordBearer(tokenUrlf{settings.API_V1_STR}/auth/login) async def get_current_user( db: AsyncSession Depends(get_db), token: str Depends(oauth2_scheme) ): Get current authenticated user. credentials_exception HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detailCould not validate credentials, headers{WWW-Authenticate: Bearer}, ) try: payload jwt.decode(token, settings.SECRET_KEY, algorithms[ALGORITHM]) user_id: int payload.get(sub) if user_id is None: raise credentials_exception except JWTError: raise credentials_exception user await user_repository.get(db, user_id) if user is None: raise credentials_exception return user要点OAuth2PasswordBearer(tokenUrl.../auth/login)告诉 Swagger UI 通过该地址获取 token自动生成“Authorize”交互入口token 解码失败、缺少sub、用户不存在三种情况统一收敛为 401 WWW-Authenticate: Bearer响应头符合 OAuth2 规范该依赖可被任意受保护路由以current_user: User Depends(get_current_user)方式注入。2. 端点层CRUD 与权限边界references/details.md的Pattern 4展示端点如何组合 service、repository 与鉴权依赖# api/v1/endpoints/users.py from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.ext.asyncio import AsyncSession from typing import List from app.core.database import get_db from app.schemas.user import User, UserCreate, UserUpdate from app.services.user_service import user_service from app.api.dependencies import get_current_user router APIRouter() router.post(/, response_modelUser, status_codestatus.HTTP_201_CREATED) async def create_user( user_in: UserCreate, db: AsyncSession Depends(get_db) ): Create new user. try: user await user_service.create_user(db, user_in) return user except ValueError as e: raise HTTPException(status_code400, detailstr(e)) router.get(/me, response_modelUser) async def read_current_user( current_user: User Depends(get_current_user) ): Get current user. return current_user router.get(/{user_id}, response_modelUser) async def read_user( user_id: int, db: AsyncSession Depends(get_db), current_user: User Depends(get_current_user) ): Get user by ID. user await user_service.repository.get(db, user_id) if not user: raise HTTPException(status_code404, detailUser not found) return user router.patch(/{user_id}, response_modelUser) async def update_user( user_id: int, user_in: UserUpdate, db: AsyncSession Depends(get_db), current_user: User Depends(get_current_user) ): Update user. if current_user.id ! user_id: raise HTTPException(status_code403, detailNot authorized) user await user_service.update_user(db, user_id, user_in) if not user: raise HTTPException(status_code404, detailUser not found) return user router.delete(/{user_id}, status_codestatus.HTTP_204_NO_CONTENT) async def delete_user( user_id: int, db: AsyncSession Depends(get_db), current_user: User Depends(get_current_user) ): Delete user. if current_user.id ! user_id: raise HTTPException(status_code403, detailNot authorized) deleted await user_service.repository.delete(db, user_id) if not deleted: raise HTTPException(status_code404, detailUser not found)HTTP 语义设计操作方法/路径成功状态码错误场景 → 状态码创建用户POST /api/v1/users/201 Created邮箱重复 → 400查询自己GET /api/v1/users/me200 OK未认证 → 401查询单用户GET /api/v1/users/{user_id}200 OK不存在 → 404更新用户PATCH /api/v1/users/{user_id}200 OK越权 → 403不存在 → 404删除用户DELETE /api/v1/users/{user_id}204 No Content越权 → 403不存在 → 404边界控制经验写操作先做“属主校验”current_user.id ! user_id直接 403再做存在性校验符合“越权优先、资源次之”的安全直觉/me作为便捷端点省去前端拼装当前用户 ID 的成本。八、异步测试httpx 内存 SQLitereferences/details.md与 SKILL.md 的 Testing 章节一致给出完整的 pytest 异步测试方案。通过app.dependency_overrides[get_db] override_get_db把真实数据库替换为内存级异步 SQLite实现无需外部服务的隔离测试# tests/conftest.py import pytest import asyncio from httpx import AsyncClient from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from app.main import app from app.core.database import get_db, Base TEST_DATABASE_URL sqliteaiosqlite:///:memory: pytest.fixture(scopesession) def event_loop(): loop asyncio.get_event_loop_policy().new_event_loop() yield loop loop.close() pytest.fixture async def db_session(): engine create_async_engine(TEST_DATABASE_URL, echoTrue) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) AsyncSessionLocal sessionmaker( engine, class_AsyncSession, expire_on_commitFalse ) async with AsyncSessionLocal() as session: yield session pytest.fixture async def client(db_session): async def override_get_db(): yield db_session app.dependency_overrides[get_db] override_get_db async with AsyncClient(appapp, base_urlhttp://test) as client: yield client对应端点级测试# tests/test_users.py import pytest pytest.mark.asyncio async def test_create_user(client): response await client.post( /api/v1/users/, json{ email: testexample.com, password: testpass123, name: Test User } ) assert response.status_code 201 data response.json() assert data[email] testexample.com assert id in data测试策略解读dependency_overrides是核心替换点用会话内 fixture 提供的db_session覆盖get_db路由代码零改动即可在测试库上运行内存 SQLite启动即建表Base.metadata.create_all无需迁移文件即可验证 schema 与接口契约scopesession的event_loopfixture 处理 asyncio 事件循环在 pytest 中的生命周期兼容问题断言采用“状态码 关键业务字段”双保险如 201 email 回读一致 自增id存在。九、技能如何在仓库生态中被使用从结构看fastapi-templates 遵循 Agent Skills 规范的三层渐进式披露frontmatter元数据→ SKILL.md导航与概览→references/details.md按需加载的详细模式。SKILL.md 明确写道Detailed sections (starting with## Implementation Patterns) live in references/details.md。Read that file when the navigation summary above is insufficient——当导航摘要不足以支撑任务时再读取细节文件从而在 token 开销与信息密度之间取得平衡。在与 agents/fastapi-pro.md 的协作中Agent 负责架构决策API 契约设计、微服务拆分、限流/熔断选型等技能负责提供可执行的工程模板。而在更大的 Agent 工作流中这一组合常出现在如下链条见 docs/agent-skills.md 对 Agent 与 Skill 协作的说明backend-architect agent → 规划 API 架构 ↓ api-design-principles skill → 提供 REST/GraphQL 最佳实践 ↓ fastapi-templates skill → 提供生产级项目模板落地结语fastapi-templates 的核心价值在于把 FastAPI 社区验证过的一套“工程默认值”固化下来分层目录 Depends依赖注入 lifespan/CORS 组装 Pydantic Settings 配置 Repository/Service 双层解耦 JWT/bcrypt 安全基座 依赖覆盖式异步测试。按 references/details.md 从 Pattern 1 到 Pattern 5 依序落地即可得到一个结构清晰、可测试、可扩展、开箱即带认证的生产级 FastAPI 项目骨架。【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
