Skip to content

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 异步 / 性能更优)
uvlatest包管理 + 虚拟环境(替代 poetry / pip-tools)
fastapi>=0.115Web 框架
uvicorn[standard]>=0.30ASGI 服务器(开发)
gunicorn>=22进程管理(生产)
pydantic>=2.7数据验证
pydantic-settings>=2.4配置 / 环境变量
sqlalchemy[asyncio]>=2.0.30ORM(异步)
asyncpg>=0.29PostgreSQL 异步驱动
alembic>=1.13数据库迁移
httpx>=0.27调用外部 HTTP(替代 requests)
redis>=5.0(async)缓存 / 队列
pytest + pytest-asynciolatest测试
ruff + mypylatestlint / 类型检查

不要requestspymysqlflask-* —— 都不要装。


三、目录结构(公司标准)

我们用 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 mypy

4.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

访问:


五、配置: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: datetime

6.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 lead

src/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.metadata

Alembic 默认是同步驱动,所以 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 c

tests/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 migrate

10.2 给 AI 装一个公司专属 Skill

~/.claude/skills/qdy-fastapi/
└── SKILL.md
yaml
---
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 时间。


十一、参考资料


继续 → 02 · 数据库 · PostgreSQL

以股东之心学习 · 以工程师之手交付 · 以 AI 集群之力放大。持之以恒,勇敢探索。