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

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

简单记录一下基于fastapi构建py后端服务的一些东西

目录结构#

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/: API 路由和视图，分版本管理。
    • models/: 数据库模型。
    • schemas/: 数据模型，用于请求和响应的验证。
    • crud/: 数据库操作（CRUD：创建、读取、更新、删除）。
    • db/: 数据库相关设置和会话管理。
    • tests/: 测试代码。
    • utils/: 工具函数和公用模块。
  • .env: 环境变量文件，用于存储敏感信息，如数据库连接字符串。
  • alembic/: 数据库迁移工具 Alembic 的配置目录。
  • requirements.txt: 项目依赖列表。
  • Dockerfile: Docker 配置文件，用于容器化部署。
  • README.md: 项目说明文件。

这个结构可以根据项目需求进行调整，但保持清晰和模块化是良好的实践。

Python 项目中的 __init__.py 文件#

在 Python 项目中，__init__.py 文件的主要作用是将目录标识为一个 Python 包。它使得目录中的模块可以被导入和使用。在一些情况下，__init__.py 可以不仅仅是一个空文件，还可以包含一些初始化代码。

__init__.py 的意义：

  1. 将目录标识为包:

    • 任何包含 __init__.py 的目录都会被 Python 解释器认为是一个包，这样你就可以使用包导入语法，如 import mypackage.module。

  2. 初始化代码:

    • 可以在 __init__.py 中包含一些初始化代码，如导入包内的子模块、设置包级别的变量或函数、配置日志记录等。
mypackage/__init__.py
from .submodule1 import func1
from .submodule2 import func2


__all__ = ["func1", "func2"]
  1. 简化导入:
    • 通过在 __init__.py 中导入子模块，可以简化包的导入路径，使得用户可以直接从包中导入函数或类，而不必知道具体的模块结构。
mypackage/__init__.py
from .submodule import MyClass


# Now you can do
from mypackage import MyClass

对于 Python 3.3 及以上版本，__init__.py 文件不是强制性的，即使没有 __init__.py 文件，Python 解释器也可以识别包。然而，添加 __init__.py 文件仍然是一个良好的习惯，可以避免某些情况下的意外行为，并且明确表示该目录是一个包。

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 层就可以调用模型对象来进行数据库操作了。上述代码展示了如何定义模型、数据库会话、CRUD 操作、数据模型和 API 端点，并将它们结合在一起，实现一个简单的用户管理系统。

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

可以通过创建一个通用的 CRUD 基类来封装常规的 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

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

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)#

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)#

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 基类中封装常规的 CRUD 操作，而特定的 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)#

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 模型的基类定义#

app.db.base_class 通常是用于定义 SQLAlchemy 模型基类的文件。在这个文件中，我们会定义一个基本的 Base 类，这个类是所有 SQLAlchemy 模型的基类。

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

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 类文件如下定义#

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://firefly.cuteleaf.cn/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
2025-12-07-杂谈-复杂异构系统监控与可视化项目设计
项目 2025-12-07
4
爬虫实战-爬取广州大学课程表
工具 爬虫实战-爬取广州大学课程表
5
关于WSL Docker清理
工具 Windows下释放Docker所占用的WSL磁盘空间
随机文章 随便看看

评论

留下你的足迹吧~

Profile Image of the Author
Zhongye
南漂中
公告
新的博客站！旧站点传送门 zhongye1.github.io/Arknight-notes
音乐
专辑封面

音乐

暂无播放

0:00 0:00
暂无歌词
分类
标签
站点统计
文章数
142
分类数
14
标签数
214
总字数
339,690
运行天数
0 天
最后更新
0 天前
© 2026 Zhongye. All Rights Reserved. / RSS / Sitemap
Powered by Astro & Firefly

目录