主题
03-01 · 后端 · FastAPI
FastAPI 是公司后端默认框架。 它的几个特点天然适配 AI 协同:异步、强类型(Pydantic v2 + SQLAlchemy 2.x)、自动 OpenAPI、生态成熟。
一、为什么是 FastAPI
| 候选方案 | 评价 |
|---|---|
| FastAPI(推荐) | 异步原生、Pydantic v2、社区最大、AI 模型对它最熟 |
| Django + DRF | 生态全但偏重,async 支持不均,AI 协同体验不如 FastAPI |
| Flask | 太简单,没有现成的依赖注入与 schema 验证 |
| Node.js(Hono / Express) | 适合纯 JS 团队,但和我们 Python AI 栈协作不便 |
| Go(Gin / Fiber) | 性能极佳但 AI 协同效率低于 FastAPI |
公司硬约定:除非 RFC 通过,否则新后端项目一律 FastAPI。
二、版本与依赖锁定(公司默认)
| 库 | 版本约束 | 用途 |
|---|---|---|
| Python | >=3.11,<3.13 | 运行时(3.11 异步 / 性能更优) |
| uv | latest | 包管理 + 虚拟环境(替代 poetry / pip-tools) |
| fastapi | >=0.115 | Web 框架 |
| uvicorn[standard] | >=0.30 | ASGI 服务器(开发) |
| gunicorn | >=22 | 进程管理(生产) |
| pydantic | >=2.7 | 数据验证 |
| pydantic-settings | >=2.4 | 配置 / 环境变量 |
| sqlalchemy[asyncio] | >=2.0.30 | ORM(异步) |
| asyncpg | >=0.29 | PostgreSQL 异步驱动 |
| alembic | >=1.13 | 数据库迁移 |
| httpx | >=0.27 | 调用外部 HTTP(替代 requests) |
| redis | >=5.0(async) | 缓存 / 队列 |
| pytest + pytest-asyncio | latest | 测试 |
| ruff + mypy | latest | lint / 类型检查 |
不要:
requests、pymysql、flask-*—— 都不要装。
三、目录结构(公司标准)
我们用 src/ 布局,避免 import 路径问题:
my-fastapi-app/
├── pyproject.toml ← 依赖与构建配置
├── uv.lock ← 锁文件(必须 commit)
├── README.md
├── AGENTS.md / CLAUDE.md ← 给 AI 的项目说明
├── Makefile
├── docker-compose.yml ← dev 用 PG / Redis
├── .env.example ← 环境变量示例
├── alembic.ini
├── alembic/
│ ├── env.py
│ └── versions/ ← 迁移文件
├── src/
│ └── app/
│ ├── __init__.py
│ ├── main.py ← FastAPI 入口
│ ├── config.py ← Pydantic Settings
│ ├── api/
│ │ ├── __init__.py
│ │ ├── deps.py ← 通用依赖(DB / 认证)
│ │ ├── v1/
│ │ │ ├── __init__.py
│ │ │ ├── leads.py
│ │ │ └── users.py
│ │ └── router.py ← 总路由聚合
│ ├── core/
│ │ ├── security.py
│ │ └── logging.py
│ ├── db/
│ │ ├── base.py ← Declarative Base
│ │ └── session.py ← async engine + session factory
│ ├── models/ ← SQLAlchemy ORM 模型
│ ├── schemas/ ← Pydantic 模型(DTO)
│ ├── services/ ← 业务逻辑层
│ └── repositories/ ← 数据访问层
└── tests/
├── conftest.py
├── test_api/
└── test_services/三层结构:
api/接 HTTP、services/写业务、repositories/操作 DB。不要把数据库查询写到 router 里。
四、最小可运行项目(5 分钟跑通)
4.1 装 uv
bash
# macOS / Linux
curl -LsSf https://astral.sh/uv/install.sh | sh
# Windows
powershell -c "irm https://astral.sh/uv/install.ps1 | iex"
uv --version为什么 uv?它是 Astral 团队(ruff 作者)写的,比 pip 快 10-100 倍,原生支持 lock 文件,正在替代 poetry / pip-tools / virtualenv 成为 Python 包管理的事实标准。
4.2 起项目
bash
mkdir my-api && cd my-api
uv init --package
uv add fastapi "uvicorn[standard]" pydantic pydantic-settings \
"sqlalchemy[asyncio]" asyncpg alembic httpx
uv add --dev pytest pytest-asyncio ruff mypy4.3 写最小代码
src/my_api/main.py:
python
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI(title="My API", version="0.1.0")
class HealthResp(BaseModel):
status: str
version: str
@app.get("/health", response_model=HealthResp)
async def health() -> HealthResp:
return HealthResp(status="ok", version=app.version)4.4 跑起来
bash
uv run uvicorn my_api.main:app --reload --port 8000访问:
- http://localhost:8000/health
- http://localhost:8000/docs ← FastAPI 自动生成 OpenAPI
五、配置:Pydantic Settings + .env
src/app/config.py:
python
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
extra="ignore",
)
APP_ENV: str = "development"
APP_BASE_URL: str = "http://localhost:8000"
DATABASE_URL: str # postgresql+asyncpg://user:pass@host:5432/dbname
REDIS_URL: str = "redis://localhost:6379/0"
OPENAI_API_KEY: str | None = None
ANTHROPIC_API_KEY: str | None = None
settings = Settings() # type: ignore[call-arg].env(永远不要 commit):
ini
APP_ENV=development
DATABASE_URL=postgresql+asyncpg://qdy:qdy@localhost:5432/qdy_dev
REDIS_URL=redis://localhost:6379/0
ANTHROPIC_API_KEY=sk-ant-xxx六、异步 SQLAlchemy 2.x + asyncpg
公司硬约定:新项目一律 SQLAlchemy 2.x typed + async + asyncpg。不要用同步
psycopg2。
6.1 引擎与 session
src/app/db/session.py:
python
from collections.abc import AsyncIterator
from sqlalchemy.ext.asyncio import (
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from app.config import settings
engine = create_async_engine(
settings.DATABASE_URL,
pool_size=10,
max_overflow=20,
pool_pre_ping=True, # 自动剔除断连
pool_recycle=1800, # 30 分钟回收一次
echo=False,
)
async_session_maker = async_sessionmaker(
engine,
expire_on_commit=False,
class_=AsyncSession,
)
async def get_db() -> AsyncIterator[AsyncSession]:
async with async_session_maker() as session:
try:
yield session
finally:
await session.close()6.2 ORM 模型(typed)
src/app/db/base.py:
python
from datetime import datetime
from sqlalchemy import DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
class Base(DeclarativeBase):
pass
class TimestampMixin:
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
)src/app/models/lead.py:
python
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base, TimestampMixin
class Lead(Base, TimestampMixin):
__tablename__ = "leads"
id: Mapped[int] = mapped_column(primary_key=True)
company_name: Mapped[str] = mapped_column(String(200), index=True)
industry: Mapped[str | None] = mapped_column(String(100))
score: Mapped[float] = mapped_column(default=0.0)6.3 Pydantic schema
src/app/schemas/lead.py:
python
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
class LeadIn(BaseModel):
company_name: str = Field(min_length=2, max_length=200)
industry: str | None = None
score: float = 0.0
class LeadOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
company_name: str
industry: str | None
score: float
created_at: datetime6.4 Repository / Service / Router 三层
src/app/repositories/lead_repo.py:
python
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.lead import Lead
class LeadRepository:
def __init__(self, db: AsyncSession) -> None:
self.db = db
async def list(self, limit: int = 50) -> list[Lead]:
result = await self.db.execute(
select(Lead).order_by(Lead.score.desc()).limit(limit)
)
return list(result.scalars().all())
async def create(self, **kwargs) -> Lead:
lead = Lead(**kwargs)
self.db.add(lead)
await self.db.commit()
await self.db.refresh(lead)
return leadsrc/app/api/v1/leads.py:
python
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.session import get_db
from app.repositories.lead_repo import LeadRepository
from app.schemas.lead import LeadIn, LeadOut
router = APIRouter(prefix="/leads", tags=["leads"])
@router.get("", response_model=list[LeadOut])
async def list_leads(db: AsyncSession = Depends(get_db)) -> list[LeadOut]:
repo = LeadRepository(db)
leads = await repo.list()
return [LeadOut.model_validate(l) for l in leads]
@router.post("", response_model=LeadOut, status_code=201)
async def create_lead(
payload: LeadIn,
db: AsyncSession = Depends(get_db),
) -> LeadOut:
repo = LeadRepository(db)
lead = await repo.create(**payload.model_dump())
return LeadOut.model_validate(lead)总路由 src/app/api/router.py:
python
from fastapi import APIRouter
from app.api.v1 import leads
api_router = APIRouter(prefix="/api/v1")
api_router.include_router(leads.router)挂到主 app:
python
# src/app/main.py
from fastapi import FastAPI
from app.api.router import api_router
app = FastAPI(title="QDY API", version="0.1.0")
app.include_router(api_router)七、Alembic 迁移(公司硬约定)
所有数据库变更必须有 migration。禁止手改生产库结构。
7.1 初始化
bash
uv run alembic init alembic修改 alembic/env.py:
python
from app.config import settings
from app.db.base import Base
import app.models # noqa: F401 ← 重要,加载所有模型
# 关键:把 sqlalchemy.url 替换成 settings.DATABASE_URL
config.set_main_option("sqlalchemy.url", settings.DATABASE_URL.replace("+asyncpg", "+psycopg"))
target_metadata = Base.metadataAlembic 默认是同步驱动,所以 URL 用
psycopg。运行时仍用asyncpg。
7.2 生成 + 应用
bash
# 生成迁移
uv run alembic revision --autogenerate -m "create leads"
# 应用
uv run alembic upgrade head
# 回滚一格
uv run alembic downgrade -1八、测试
tests/conftest.py:
python
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from app.main import app
@pytest_asyncio.fixture
async def client() -> AsyncClient:
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test",
) as c:
yield ctests/test_api/test_leads.py:
python
import pytest
@pytest.mark.asyncio
async def test_list_leads(client):
resp = await client.get("/api/v1/leads")
assert resp.status_code == 200
assert isinstance(resp.json(), list)跑测试:
bash
uv run pytest -q九、Makefile(公司统一)
makefile
.PHONY: install dev test lint format migrate ship
install:
uv sync
dev:
docker compose up -d db redis
uv run uvicorn app.main:app --reload --port 8000
test:
uv run pytest -q
lint:
uv run ruff check .
uv run mypy src
format:
uv run ruff format .
uv run ruff check --fix .
migrate:
uv run alembic upgrade head
ship: lint test
@echo "all green"十、AI 协同最佳实践
10.1 在 CLAUDE.md / AGENTS.md 里写清楚
markdown
# Project: QDY API
## 技术栈
- Python 3.11 + FastAPI + Pydantic v2
- SQLAlchemy 2.x async + asyncpg + Alembic
- uv 包管理(不要用 pip / poetry)
- pytest-asyncio
## 三层结构
api/ → services/ → repositories/ 严格分层。
不要在 router 里直接写 SQL。
## 命名约定
- 文件:snake_case
- 类:PascalCase
- API 路径:/api/v1/{resource}
- 时间字段:UTC,timezone=True
## 必走命令
- make dev / make test / make migrate10.2 给 AI 装一个公司专属 Skill
~/.claude/skills/qdy-fastapi/
└── SKILL.mdyaml
---
name: qdy-fastapi
description: 当用户在 FastAPI 项目里要求"加一个 API"、"加一个表"时启用
---
# QDY FastAPI 标准做法
## 加一个新资源(如 leads)的步骤
1. 在 src/app/models/ 写 ORM
2. 在 src/app/schemas/ 写 Pydantic
3. 在 src/app/repositories/ 写仓储
4. 在 src/app/api/v1/ 写 router
5. 在 src/app/api/router.py 里挂上
6. uv run alembic revision --autogenerate -m "..."
7. 写 tests/test_api/test_xxx.py
8. uv run pytest 通过后才提交
## 禁止
- 在 router 里直接 db.execute
- 用同步 psycopg2
- 全局变量保存 Session挂上之后,AI 再写新资源会 完全按公司规范,省掉大量 review 时间。
十一、参考资料
- 📚 FastAPI 官方文档
- 📚 SQLAlchemy 2.x ORM Quick Start
- 📚 Pydantic v2 文档
- 📚 uv 官方文档
- 📚 Alembic Tutorial
- 📚 FastAPI 数据库实战(中文)