大家对FastAPI的印象，一是这个框架真快，二往往是项目写到第三周就开始乱
接口数量一多，路由里堆业务、异常风格不统一、前后端联调经常扯皮，最后变成每次加功能都像拆炸弹

这篇文章的目标非常明确：
不是再讲一遍@app.get，而是带你把一个FastAPI入门项目，做成可维护的工程底盘

你看完后应该能做到：
1. 独立搭建一个可运行、可联调、可扩展的CRUD项目
2. 理解并实践 
outer -> service -> storage的分层职责
3. 建立统一响应结构与统一异常语义，减少联调成本

一、为什么第一步必须先做“单体CRUD+分层+统一异常”
 

很多人上来就做：登录鉴权、消息队列、缓存、微服务、异步任务、A/B 测试……
结果功能看起来很多，但基础协议不稳，后续每一步都在返工

工程上更稳的顺序是：
- 先把接口契约做稳（字段、状态码、错误结构）
- 先把分层边界做清（谁负责收参、谁负责规则、谁负责存取）
- 先把异常语义统一（前端和测试都能稳定处理）

这一阶段不追求大，只追求稳

二、阶段目标与边界
 

本篇阶段目标：
- 用户与会议室CRUD跑通
- 分层结构固定下来
- 统一响应结构固定下来
- 全局异常处理固定下来

后续优化方向：
- 真正数据库持久化（第二篇）
- JWT/OAuth2 鉴权（第三篇）
- 系统化测试（第四篇）
- 性能和部署优化（第五篇）

三、项目结构
 

app/
  main.py
  routers/
    users.py
    rooms.py
  services/
    users_service.py
    rooms_service.py
    exceptions.py
  schemas/
    common.py
    users.py
    rooms.py
  storage/
    memory_db.py
tests/
 

职责划分：

- outers：只处理HTTP输入输出，不写复杂业务规则
- services：业务规则中心，包含唯一性校验、状态流转、组合逻辑等功能
- schemas：请求与响应模型，保证契约稳定
- storage：数据读写细节，当前使用内存，后续可替换数据库
- main.py：装配层，负责路由注册、异常注册、应用启动

第一步：定义统一响应结构
 

先做schemas/common.py：
 

from typing import Any
from pydantic import BaseModel


class ApiError(BaseModel):
    code: str
    details: Any | None = None


class ApiResponse(BaseModel):
    success: bool
    message: str
    data: Any | None = None
    error: ApiError | None = None


def success_response(data: Any = None, message: str = "成功") -> dict:
    return ApiResponse(success=True, message=message, data=data).model_dump()


def error_response(message: str, code: str, details: Any | None = None) -> dict:
    return ApiResponse(
        success=False,
        message=message,
        error=ApiError(code=code, details=details),
    ).model_dump()

为什么这一步要最先做：
- 前端只要认一个结构——success/message/data/error
- 测试断言成本更低
- 后续你换业务实体，也不需要改交互协议

第二步：定义输入模型与字段边界
 

schemas/users.py：
 

from pydantic import BaseModel, EmailStr, Field


class UserCreate(BaseModel):
    name: str = Field(min_length=2, max_length=50)
    email: EmailStr


class UserUpdate(BaseModel):
    name: str | None = Field(default=None, min_length=2, max_length=50)
    email: EmailStr | None = None
`

schemas/rooms.py：

`python
from pydantic import BaseModel, Field


class RoomCreate(BaseModel):
    name: str = Field(min_length=2, max_length=60)
    capacity: int = Field(gt=0, le=200)
    location: str = Field(default="", max_length=120)
    has_projector: bool = False


class RoomUpdate(BaseModel):
    name: str | None = Field(default=None, min_length=2, max_length=60)
    capacity: int | None = Field(default=None, gt=0, le=200)
    location: str | None = Field(default=None, max_length=120)
    has_projector: bool | None = None

这层的价值：
- 把脏数据拦在入口
- 让接口文档天然可读
- 明确字段上下限，避免后续业务层被污染

第三步：抽出业务异常类型
 

services/exceptions.py：

class EntityNotFoundError(Exception):
    def __init__(self, entity_name: str, entity_id: int):
        super().__init__(f"{entity_name}不存在: {entity_id}")


class BusinessRuleError(Exception):
    pass

这一步看似简单，实际很关键：
- 让service层用业务语言报错
- 让router层做HTTP映射
- 异常职责分离后，代码会更干净

第四步：先用内存存储跑通主链路
 

storage/memory_db.py：
 

from typing import Any

users: dict[int, dict[str, Any]] = {}
rooms: dict[int, dict[str, Any]] = {}
_user_id_seq = 0
_room_id_seq = 0


def next_user_id() -> int:
    global _user_id_seq
    _user_id_seq += 1
    return _user_id_seq


def next_room_id() -> int:
    global _room_id_seq
    _room_id_seq += 1
    return _room_id_seq


def reset_storage() -> None:
    global _user_id_seq, _room_id_seq
    users.clear()
    rooms.clear()
    _user_id_seq = 0
    _room_id_seq = 0

第五步：业务规则放service，不放router
 

services/users_service.py：

from app.schemas.users import UserCreate, UserUpdate
from app.services.exceptions import BusinessRuleError, EntityNotFoundError
from app.storage.memory_db import users, next_user_id


class UserService:
    def list_users(self) -> list[dict]:
        return list(users.values())

    def get_user(self, user_id: int) -> dict:
        user = users.get(user_id)
        if not user:
            raise EntityNotFoundError("用户", user_id)
        return user

    def create_user(self, payload: UserCreate) -> dict:
        if any(item["email"] == payload.email for item in users.values()):
            raise BusinessRuleError("邮箱已存在")

        user_id = next_user_id()
        user = {"id": user_id, **payload.model_dump()}
        users[user_id] = user
        return user

    def update_user(self, user_id: int, payload: UserUpdate) -> dict:
        existing = users.get(user_id)
        if not existing:
            raise EntityNotFoundError("用户", user_id)

        update_data = payload.model_dump(exclude_none=True)
        new_email = update_data.get("email")
        if new_email and any(
            item["email"] == new_email and item["id"] != user_id
            for item in users.values()
        ):
            raise BusinessRuleError("邮箱已存在")

        existing.update(update_data)
        return existing

    def delete_user(self, user_id: int) -> dict:
        existing = users.get(user_id)
        if not existing:
            raise EntityNotFoundError("用户", user_id)
        del users[user_id]
        return existing

同理可以实现RoomService，名称唯一、忽略大小写

第六步：router只做协议层翻译
 

outers/users.py：

from fastapi import APIRouter, HTTPException, status
from app.schemas.common import success_response
from app.schemas.users import UserCreate, UserUpdate
from app.services.exceptions import EntityNotFoundError, BusinessRuleError
from app.services.users_service import UserService

router = APIRouter(prefix="/users", tags=["用户管理"])
service = UserService()


@router.get("")
def list_users():
    return success_response(data=service.list_users(), message="用户列表获取成功")


@router.post("", status_code=status.HTTP_201_CREATED)
def create_user(payload: UserCreate):
    try:
        user = service.create_user(payload)
    except BusinessRuleError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    return success_response(data=user, message="用户创建成功")

关键原则：
- router 不写复杂规则
- router 负责异常 -> HTTP状态码的映射

第七步：在main.py统一注册异常处理
 

main.py：

from fastapi import FastAPI, Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
from app.schemas.common import error_response, success_response

app = FastAPI(title="FastAPI 工程化入门", version="0.1.0")


@app.get("/health")
def health_check():
    return success_response(data={"status": "ok"}, message="服务运行正常")


@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(_: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content=error_response(message=str(exc.detail), code=f"HTTP_{exc.status_code}"),
    )


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(_: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content=error_response(
            message="参数校验失败",
            code="VALIDATION_ERROR",
            details=exc.errors(),
        ),
    )


@app.exception_handler(Exception)
async def unhandled_exception_handler(_: Request, exc: Exception):
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content=error_response(
            message="服务器未知错误",
            code="INTERNAL_SERVER_ERROR",
            details=str(exc),
        ),
    )

优点：
- 所有错误出口统一
- 前端处理逻辑显著简化
- 排障与日志定位更稳定

状态码建议
 

建议语义：
- 200：查询/更新/删除成功
- 201：创建成功
- 400：业务规则冲突，如唯一性失败
- 404：资源不存在
- 422：请求体或参数校验失败
- 500：未预期错误

最小运行与验证
 

启动：

python -m uvicorn app.main:app --reload

访问：
- http://127.0.0.1:8000/docs
- http://127.0.0.1:8000/redoc
- http://127.0.0.1:8000/health

最小验收清单：
- CRUD 路由都能通
- 业务冲突返回 400 且结构统一
- 不存在资源返回 404 且结构统一
- 参数错误返回 422 且含错误详情

四、容易踩的坑
 

1. 在router里写规则：
- 结果：后续复用困难，代码越来越长
- 建议：router只收参和返回，规则下沉service

2. 响应结构不统一：
- 结果：前端每个接口都写不同分支
- 建议：全局统一 success/message/data/error

3. 直接上数据库再谈架构：
- 结果：基础边界没稳，排错复杂度翻倍
- 建议：先内存跑通，再数据库迁移

4. 过早追求功能数量：
- 结果：接口越来越多，协议越来越乱
- 建议：先把协议和异常做稳，再加功能

五、本篇小结
 

这一篇你真正拿到的，不是一个会跑的FastAPI Demo
而是一套能承接后续演进的工程起点：
- 分层职责清晰
- 接口契约稳定
- 异常语义统一

下一篇我们进入第二阶段：
把当前内存存储替换为数据库，并给出SQLite -> MySQL/PostgreSQL的平滑迁移路径