2026-03-28-FastAPI 工程实践随记

3037 个字
15 分钟
2026-03-28-FastAPI 工程实践随记
2026-03-29
笔记
FastAPI
/
Python
/
后端
/
Docker
/
工程实践

简单记录一下用 FastAPI 搭 Python 后端服务时摸索出来的一套做法，从目录怎么分、CRUD 怎么写，到怎么把通用逻辑抽成基类，按实际项目里的演进顺序串下来

目录结构#

Terminal window
my_fastapi_project/
├── .env                            # 环境变量配置 (数据库URL、密钥、Debug模式等)
├── .gitignore                      # Git 忽略文件配置
├── alembic.ini                     # Alembic 数据库迁移配置文件
├── docker-compose.yml              # Docker 编排文件
├── Dockerfile                      # Docker 镜像构建文件
├── README.md                       # 项目说明文档
├── requirements.txt                # 项目依赖包列表
│
├── alembic/                        # 数据库迁移目录
│   ├── env.py                      # Alembic 环境配置 (需在此导入所有 Models)
│   ├── script.py.mako              # 迁移脚本模板
│   └── versions/                   # 存放具体的迁移版本文件
│       └── *.py
│
└── app/                            # 应用核心目录
    ├── __init__.py
    ├── main.py                     # FastAPI 应用入口 (创建 app 实例、注册路由、中间件)
    │
    ├── api/                        # API 接口层
    │   ├── __init__.py
    │   ├── dependencies.py         # [关键] 公共依赖注入 (如：get_db, get_current_user, 分页参数)
    │   └── v1/                     # API 版本控制
    │       ├── __init__.py         # 汇总该版本所有路由
    │       └── endpoints/          # 具体的接口端点
    │           ├── __init__.py
    │           ├── auth.py         # 认证接口 (登录、注册)
    │           ├── users.py        # 用户相关接口
    │           └── items.py        # 业务相关接口
    │
    ├── core/                       # 核心配置层
    │   ├── __init__.py
    │   ├── config.py               # 配置管理 (使用 Pydantic BaseSettings 读取 .env)
    │   ├── security.py             # 安全模块 (JWT 生成/校验、密码哈希)
    │   └── exceptions.py           # [补充] 自定义异常类 (如 UserNotFound, InsufficientPermission)
    │
    ├── crud/                       # 数据访问层
    │   ├── __init__.py
    │   ├── base.py                 # [补充] CRUD 基类 (封装通用的 get, create, update, delete 方法)
    │   ├── user.py                 # 用户数据操作 (继承 base.py)
    │   └── item.py                 # 业务数据操作
    │
    ├── db/                         # 数据库连接层
    │   ├── __init__.py
    │   ├── base.py                 # [关键] 导入所有 Models (供 Alembic 扫描)
    │   └── session.py              # 数据库引擎创建、SessionLocal 工厂
    │
    ├── models/                     # 数据库模型层
    │   ├── __init__.py
    │   ├── user.py                 # 用户表 ORM 模型
    │   └── item.py                 # 业务表 ORM 模型
    │
    ├── schemas/                    # 数据校验/序列化层
    │   ├── __init__.py
    │   ├── base.py                 # [补充] 通用 Schema (如分页响应基类)
    │   ├── user.py                 # 用户相关的 Pydantic 模型
    │   │   # 通常包含: UserCreate, UserUpdate, UserInDB, UserResponse
    │   └── item.py                 # 业务相关的 Pydantic 模型
    │
    ├── services/                   # [补充] 业务逻辑层
    │   ├── __init__.py
    │   └── user_service.py         # 复杂业务逻辑封装 (例如: 注册流程含发邮件、初始化数据等)
    │                               # 作用: 防止 API 层代码过于臃肿，解耦 CRUD 操作
    │
    ├── tests/                      # 测试目录
    │   ├── __init__.py
    │   ├── conftest.py             # [关键] Pytest 配置 (定义 client, db_session fixture)
    │   ├── test_main.py            # 测试入口逻辑
    │   └── test_api/               # API 接口测试
    │       ├── __init__.py
    │       └── test_users.py
    │
    └── utils/                      # 工具函数库
        ├── __init__.py
        ├── email.py                # 邮件发送工具
        └── logging.py              # 日志配置工具

目录结构说明#

整套结构的中心是 app/，所有应用代码都收在这里。最外层的 main.py 是入口文件，负责把 FastAPI 应用拉起来。往里走，职责是按层切的：core/ 放配置、安全这类核心功能；api/ 是路由和视图，并且按版本分目录管理；models/ 是数据库模型，schemas/ 是用于请求和响应校验的 Pydantic 模型，二者刚好对应”数据库里长什么样”和”接口上长什么样”两件事；crud/ 专门承载增删改查的数据库操作；db/ 管数据库设置和会话；剩下的 tests/ 放测试代码，utils/ 放工具函数和公用模块。

app/ 之外还有几个项目级的文件值得单独点一下。.env 用来存数据库连接串这类敏感信息；alembic/ 是数据库迁移工具 Alembic 的配置目录；requirements.txt 列依赖；Dockerfile 负责容器化部署；README.md 是项目说明。这套结构不是死的，可以按项目需要调整，但”保持清晰和模块化”这条原则最好一直守着。

Python 项目中的 __init__.py 文件#

__init__.py 在 Python 项目里最根本的作用，是把一个目录标识成 Python 包——只要目录里有它，解释器就把这个目录当成包，你才能用 import mypackage.module 这种包导入语法。可以写初始化代码，比如导入包内的子模块、设置包级别的变量或函数、配置日志等等：

mypackage/__init__.py
from .submodule1 import func1
from .submodule2 import func2


__all__ = ["func1", "func2"]

它还有个很实用的价值是简化导入路径。在 __init__.py 里提前把子模块里的东西导进来，使用方就不必关心具体的模块结构，直接从包顶层拿就行：

mypackage/__init__.py
from .submodule import MyClass


# Now you can do
from mypackage import MyClass

FastAPI 项目的开发处理过程#

在 FastAPI 项目里，CRUD 操作一般会收进一个专门的 crud 模块，由它去调用 SQLAlchemy 模型对象完成数据库读写。下面从模型开始，一步步把一个最小可用的用户管理系统串起来。

1. 定义模型 (models/user.py)#

from sqlalchemy import Column, Integer, String
from app.db.base_class import Base


class User(Base):
    __tablename__ = "users"


    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    full_name = Column(String, index=True)

2. 创建数据库会话 (db/session.py)#

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker


DATABASE_URL = "sqlite:///./test.db"  # 使用 SQLite 数据库作为示例
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

3. 定义 CRUD 操作 (crud/user.py)#

from sqlalchemy.orm import Session
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate


def get_user(db: Session, user_id: int):
    return db.query(User).filter(User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(User).filter(User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 10):
    return db.query(User).offset(skip).limit(limit).all()


def create_user(db: Session, user: UserCreate):
    db_user = User(
        email=user.email,
        hashed_password=user.hashed_password,  # 在实际应用中应该对密码进行哈希处理
        full_name=user.full_name
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def update_user(db: Session, user_id: int, user: UserUpdate):
    db_user = get_user(db, user_id)
    if db_user:
        db_user.email = user.email
        db_user.full_name = user.full_name
        db.commit()
        db.refresh(db_user)
        return db_user


def delete_user(db: Session, user_id: int):
    db_user = get_user(db, user_id)
    if db_user:
        db.delete(db_user)
        db.commit()
        return db_user

4. 定义数据模型 (schemas/user.py)#

from pydantic import BaseModel


class UserBase(BaseModel):
    email: str
    full_name: str = None


class UserCreate(UserBase):
    hashed_password: str


class UserUpdate(UserBase):
    pass


class User(UserBase):
    id: int


    class Config:
        orm_mode = True

5. 在 API 端点中使用 CRUD 操作 (api/v1/endpoints/users.py)#

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
from app import crud, models, schemas
from app.db.session import SessionLocal


router = APIRouter()


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@router.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@router.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@router.put("/users/{user_id}", response_model=schemas.User)
def update_user(user_id: int, user: schemas.UserUpdate, db: Session = Depends(get_db)):
    db_user = crud.update_user(db=db, user_id=user_id, user=user)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@router.delete("/users/{user_id}", response_model=schemas.User)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.delete_user(db=db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@router.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users

6. 注册路由 (main.py)#

from fastapi import FastAPI
from app.api.v1.endpoints import users


app = FastAPI()


app.include_router(users.router, prefix="/api/v1", tags=["users"])


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

7. 初始化数据库 (db/base.py)#

from app.db.session import engine
from app.models import user


user.Base.metadata.create_all(bind=engine)

8. 运行应用#

在项目根目录下运行：

Terminal window
uvicorn app.main:app --reload

到这里，从模型、会话、CRUD、Pydantic 校验到 API 端点就全都接通了，CRUD 层可以正常调模型对象去读写数据库，一个最简版的用户管理系统也就跑起来了。

实际 FastAPI 项目对基类的封装#

上面那种写法每个资源都要把 get、create、update、delete 重抄一遍，资源一多就全是重复代码。更实际的做法是抽一个通用的 CRUD 基类，把这些常规操作封装进去，再让具体的 CRUD 类去继承它——这样既减少重复，也让代码更好维护、更好复用。

1. 创建通用 CRUD 基类 (crud/base.py)#

from typing import Generic, Type, TypeVar, Optional, List
from pydantic import BaseModel
from sqlalchemy.orm import Session
from app.db.base_class import Base


ModelType = TypeVar("ModelType", bound=Base)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)


class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
    def __init__(self, model: Type[ModelType]):
        self.model = model


    def get(self, db: Session, id: int) -> Optional[ModelType]:
        return db.query(self.model).filter(self.model.id == id).first()


    def get_multi(self, db: Session, skip: int = 0, limit: int = 100) -> List[ModelType]:
        return db.query(self.model).offset(skip).limit(limit).all()


    def create(self, db: Session, obj_in: CreateSchemaType) -> ModelType:
        obj_in_data = obj_in.dict()
        db_obj = self.model(**obj_in_data)  # type: ignore
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj


    def update(self, db: Session, db_obj: ModelType, obj_in: UpdateSchemaType) -> ModelType:
        obj_data = db_obj.dict()
        update_data = obj_in.dict(skip_defaults=True)
        for field in obj_data:
            if field in update_data:
                setattr(db_obj, field, update_data[field])
        db.commit()
        db.refresh(db_obj)
        return db_obj


    def remove(self, db: Session, id: int) -> ModelType:
        obj = db.query(self.model).get(id)
        db.delete(obj)
        db.commit()
        return obj

这里用了泛型（Generic 加三个 TypeVar），把模型类型、创建用的 Schema 和更新用的 Schema 都参数化，所以这套基类对任何资源都通用。

2. 定义用户 CRUD 操作 (crud/user.py)#

有了基类，用户的 CRUD 类就只剩”额外的、专属于用户的”那部分逻辑了，比如按邮箱查人；常规的增删改查全继承自 CRUDBase：

from typing import Any
from sqlalchemy.orm import Session
from app.crud.base import CRUDBase
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate


class CRUDUser(CRUDBase[User, UserCreate, UserUpdate]):
    def get_by_email(self, db: Session, email: str) -> Any:
        return db.query(self.model).filter(self.model.email == email).first()


user = CRUDUser(User)

3. 定义数据模型 (schemas/user.py)#

Schema 这部分和前面一致，不用改：

from pydantic import BaseModel


class UserBase(BaseModel):
    email: str
    full_name: str = None


class UserCreate(UserBase):
    hashed_password: str


class UserUpdate(UserBase):
    pass


class User(UserBase):
    id: int


    class Config:
        orm_mode = True

4. 在 API 端点中使用 CRUD 操作 (api/v1/endpoints/users.py)#

端点里的调用也跟着收敛成 crud.user.xxx 的统一风格，读起来比一堆零散函数清爽不少：

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
from app import crud, schemas
from app.db.session import SessionLocal
from app.models.user import User


router = APIRouter()


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@router.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.user.get_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.user.create(db=db, obj_in=user)


@router.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.user.get(db, id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@router.put("/users/{user_id}", response_model=schemas.User)
def update_user(user_id: int, user: schemas.UserUpdate, db: Session = Depends(get_db)):
    db_user = crud.user.get(db=db, id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return crud.user.update(db=db, db_obj=db_user, obj_in=user)


@router.delete("/users/{user_id}", response_model=schemas.User)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.user.get(db=db, id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return crud.user.remove(db=db, id=user_id)


@router.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    users = crud.user.get_multi(db, skip=skip, limit=limit)
    return users

这样一来，常规的 CRUD 都沉到了基类里，像 CRUDUser 这样的具体类只负责继承基类、再补几个专属方法，重复代码大幅减少，可维护性和复用性都上来了。

如果还想再往上抽一层，可以连 API 端点本身也做成通用基类，把那五个标准 CRUD 路由的注册和处理逻辑封进去，具体资源只要继承一下就能拿到全套接口。

5. 创建通用 API 基类 (api/deps.py)#

from typing import Type, TypeVar, Generic, List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel
from app.crud.base import CRUDBase
from app.db.session import SessionLocal


ModelType = TypeVar("ModelType", bound=BaseModel)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)


class CRUDRouter(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
    def __init__(self, crud: CRUDBase[ModelType, CreateSchemaType, UpdateSchemaType]):
        self.crud = crud
        self.router = APIRouter()


        self.router.post("/", response_model=ModelType)(self.create_item)
        self.router.get("/{item_id}", response_model=ModelType)(self.read_item)
        self.router.put("/{item_id}", response_model=ModelType)(self.update_item)
        self.router.delete("/{item_id}", response_model=ModelType)(self.delete_item)
        self.router.get("/", response_model=List[ModelType])(self.read_items)


    def get_db(self):
        db = SessionLocal()
        try:
            yield db
        finally:
            db.close()


    async def create_item(self, item_in: CreateSchemaType, db: Session = Depends(self.get_db)):
        db_item = self.crud.create(db=db, obj_in=item_in)
        return db_item


    async def read_item(self, item_id: int, db: Session = Depends(self.get_db)):
        db_item = self.crud.get(db=db, id=item_id)
        if not db_item:
            raise HTTPException(status_code=404, detail="Item not found")
        return db_item


    async def update_item(self, item_id: int, item_in: UpdateSchemaType, db: Session = Depends(self.get_db)):
        db_item = self.crud.get(db=db, id=item_id)
        if not db_item:
            raise HTTPException(status_code=404, detail="Item not found")
        return self.crud.update(db=db, db_obj=db_item, obj_in=item_in)


    async def delete_item(self, item_id: int, db: Session = Depends(self.get_db)):
        db_item = self.crud.get(db=db, id=item_id)
        if not db_item:
            raise HTTPException(status_code=404, detail="Item not found")
        return self.crud.remove(db=db, id=item_id)


    async def read_items(self, skip: int = 0, limit: int = 10, db: Session = Depends(self.get_db)):
        items = self.crud.get_multi(db=db, skip=skip, limit=limit)
        return items

6. 使用通用 API 基类定义用户端点 (api/v1/endpoints/users.py)#

抽到这一步，定义一个资源的全套接口就压缩成了两行——传入对应的 CRUD 实例和三个 Schema，路由就自动挂好了：

from fastapi import APIRouter
from app.crud.user import user as user_crud
from app.schemas.user import User, UserCreate, UserUpdate
from app.api.deps import CRUDRouter


user_router = CRUDRouter[User, UserCreate, UserUpdate](user_crud)
router = user_router.router

SQLAlchemy 模型的基类定义#

前面模型文件里一直 from app.db.base_class import Base，这个 base_class 就是专门用来定义 SQLAlchemy 模型基类的文件。它的职责是给出一个所有 ORM 模型共同继承的 Base 类。

定义 Base 类 (db/base_class.py)#

这里用 @as_declarative() 装饰器声明基类，并顺手用 declared_attr 把表名约定成”类名小写”，省得每个模型都手写一遍 __tablename__：

from sqlalchemy.ext.declarative import as_declarative, declared_attr


@as_declarative()
class Base:
    id: int
    __name__: str


    @declared_attr
    def __tablename__(cls) -> str:
        return cls.__name__.lower()

models/user.py 类文件如下定义#

具体模型继承这个 Base 之后，正常声明字段即可（如果想用上面那条表名约定，这里的 __tablename__ 其实也可以省掉）：

from sqlalchemy import Column, Integer, String
from app.db.base_class import Base


class User(Base):
    __tablename__ = "users"


    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    full_name = Column(String, index=True)

分享到社交平台

将本文分享给你的朋友们

2026-03-28-FastAPI 工程实践随记
https://zhongye1.github.io/posts/2026-03-28-fastapi-工程实践随记/
作者
Zhongye
发布于
2026-03-29
版权声明
CC BY-NC-SA 4.0
2026-03-28-团队项目协作规范随记
2026-03-14-关于 MSYS2
相关文章 智能推荐
1
2026-03-01-Windows wsl安装docker
工具 2026-03-01
2
2026-01-07-机器学习相关算法
学习 2026-01-08
3
爬虫实战-爬取广州大学课程表
工具 爬虫实战-爬取广州大学课程表
4
关于WSL Docker清理
工具 Windows下释放Docker所占用的WSL磁盘空间
5
广州大学选课脚本
工具 广州大学选课脚本, 可多线程自动选全校性选课和体育选课的课程
随机文章 随便看看

评论

留下你的足迹吧~

Profile Image of the Author
Zhongye
南漂中
公告
新的博客站！旧站点传送门 👇
音乐
专辑封面

音乐

暂无播放

0:00 0:00
暂无歌词
分类
标签
站点统计
文章数
143
分类数
14
标签数
202
总字数
417,493
运行天数
0 天
最后更新
0 天前
总访问量
42296 次
访客数
29212 人
© 2026 Zhongye. All Rights Reserved. / RSS / Sitemap
Powered by Astro & Firefly

目录