企业官网建设流程全解析

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-templates

fastapi-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、业务逻辑、数据访问严格分层，并内置版本化 API（v1）：

app/ ├── 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

分层职责划分要点：

二、应用入口：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( title="API Template", version="1.0.0", lifespan=lifespan ) # CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, 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_event，lifespan 是更受推荐的生命周期方案；
• CORS 中间件中的allow_origins=["*"]仅为模板演示值，生产环境应替换为真实的受信域名白名单；allow_credentials=True与"*"组合在带 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_MINUTES=30（默认 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, echo=True, future=True ) AsyncSessionLocal = sessionmaker( engine, class_=AsyncSession, expire_on_commit=False ) 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_commit=False防止提交后属性过期导致懒加载触发额外 IO；
• get_db()是贯穿全项目的核心依赖：以async with管理会话生命周期，正常路径提交、异常路径回滚，finally中确保关闭。这个依赖将作为Depends(get_db)注入到每个需要访问数据库的路由与鉴权逻辑中。

三、依赖注入（Dependency Injection）

FastAPI 内建的 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", bound=BaseModel) UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel) 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_unset=True) 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_unset=True)只取客户端显式传入的字段，天然支持 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_unset=True) 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"], deprecated="auto") 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(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM) 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"], deprecated="auto")，deprecated="auto"允许后续无缝迁移到更强的哈希算法；
• JWT（HS256）：create_access_token将sub等载荷复制后注入exp过期时间；未显式传expires_delta时默认 15 分钟；
• 算法与密钥统一读取自Settings（SECRET_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(tokenUrl=f"{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_code=status.HTTP_401_UNAUTHORIZED, detail="Could 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_model=User, status_code=status.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_code=400, detail=str(e)) @router.get("/me", response_model=User) async def read_current_user( current_user: User = Depends(get_current_user) ): """Get current user.""" return current_user @router.get("/{user_id}", response_model=User) 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_code=404, detail="User not found") return user @router.patch("/{user_id}", response_model=User) 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_code=403, detail="Not authorized") user = await user_service.update_user(db, user_id, user_in) if not user: raise HTTPException(status_code=404, detail="User not found") return user @router.delete("/{user_id}", status_code=status.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_code=403, detail="Not authorized") deleted = await user_service.repository.delete(db, user_id) if not deleted: raise HTTPException(status_code=404, detail="User not found")

HTTP 语义设计：

边界控制经验：写操作先做“属主校验”（current_user.id != user_id直接 403），再做存在性校验，符合“越权优先、资源次之”的安全直觉；/me作为便捷端点省去前端拼装当前用户 ID 的成本。

八、异步测试：httpx + 内存 SQLite

references/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 = "sqlite+aiosqlite:///:memory:" @pytest.fixture(scope="session") 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, echo=True) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) AsyncSessionLocal = sessionmaker( engine, class_=AsyncSession, expire_on_commit=False ) 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(app=app, base_url="http://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": "test@example.com", "password": "testpass123", "name": "Test User" } ) assert response.status_code == 201 data = response.json() assert data["email"] == "test@example.com" assert "id" in data

测试策略解读：

• dependency_overrides是核心替换点：用会话内 fixture 提供的db_session覆盖get_db，路由代码零改动即可在测试库上运行；
• 内存 SQLite启动即建表（Base.metadata.create_all），无需迁移文件即可验证 schema 与接口契约；
• scope="session"的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