FastAPI: CRUD 完整实现 — 从仓储层到 API 端点

Q: Repository 一定要用类吗？

不是必须，也可以用函数。但类封装 DB Session 状态，方法间共享 Session 更方便，推荐类写法。

Q: flush 和 commit 有什么区别？

flush 发送 SQL 到数据库但不提交事务，可以获取自增 ID 和触发数据库约束检查。commit 提交事务，变更持久化。在 yield 依赖中推荐 flush + 自动 commit。

Q: N+1 问题只在循环中触发吗？

主要在循环访问关联属性时触发。但 Pydantic 序列化也会触发——如果 response_model 访问了关联字段。务必预加载。

Q: bulk_create 有大小限制吗？

无框架限制，但 PostgreSQL 单次绑定参数上限 32767。千级插入没问题，万级建议分批。

Q: Offset 分页在大数据集下有多慢？

OFFSET 1000000 需要先扫描前 100 万行再跳过，查询时间与 offset 成正比。百万级数据建议用 cursor 分页。

Q: 事务中部分失败怎么处理？

yield 依赖的 rollback 在异常时自动触发。如果需要部分回滚，用 savepoint（ await session.begin_nested() ）。

最后更新：2026-08-26

1. 你将学到

仓储模式（Repository Pattern）：ProductRepository/PriceRepository 封装
分页与排序：limit/offset 与 order_by 的 Pydantic 查询模型
关联数据预加载：selectinload vs joinedload 性能取舍
事务管理：跨表操作的一致性保证
Alice 场景：批量价格导入 API——单次请求插入千条价格记录的事务化处理

2. Alice 的真实故事

(1) 痛点：SQL 散落在路由函数中

Alice 的 PriceTracker 代码里，SQL 查询语句直接写在路由处理函数中，20 个端点有 30 处重复的 select(Product).where(...) 逻辑。当需要给所有商品查询添加"软删除过滤"条件时，Alice 需要修改 15 个地方，漏掉 2 个导致已删除的商品出现在结果中。

(2) 仓储模式的解法

将数据库操作封装在 Repository 类中，路由函数只调用 Repository 方法。添加全局过滤条件只需修改 Repository 一处。

PYTHON

class ProductRepository:
    async def get_all(self, db: AsyncSession, filters: ProductFilter) -> list[Product]:
        stmt = select(Product).where(Product.deleted == False)
        # Add filters from Pydantic model
        if filters.category:
            stmt = stmt.where(Product.category == filters.category)
        result = await db.execute(stmt)
        return result.scalars().all()

(3) 收益

SQL 逻辑集中到 Repository，路由函数变简洁（5 行替代 15 行），全局过滤条件一处修改全局生效，Bob 前端再也不看到已删除的商品。

3. 分层架构与仓储模式

(1) 四层架构

flowchart TD
    Router[Router Layer] -->|Call| Service[Service Layer]
    Service -->|Call| Repository[Repository Layer]
    Repository -->|Query| SQLAlchemy[SQLAlchemy ORM]
    SQLAlchemy -->|SQL| Database[(PostgreSQL)]
    
    style Router fill:#e3f2fd
    style Service fill:#fff3e0
    style Repository fill:#e8f5e9
    style SQLAlchemy fill:#fce4ec

层级	职责	示例
Router	参数校验、响应格式	`@app.get("/products")`
Service	业务逻辑、事务编排	`ProductService.create_with_price()`
Repository	数据访问、SQL 封装	`ProductRepository.get_by_id()`
ORM	对象关系映射	`select(Product).where(...)`

▶ 示例：ProductRepository

PYTHON

from sqlalchemy import select, update, delete
from sqlalchemy.ext.asyncio import AsyncSession

class ProductRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_by_id(self, product_id: int) -> Product | None:
        stmt = select(Product).where(Product.id == product_id)
        result = await self.db.execute(stmt)
        return result.scalar_one_or_none()

    async def get_all(
        self,
        category: str | None = None,
        skip: int = 0,
        limit: int = 20,
    ) -> list[Product]:
        stmt = select(Product).offset(skip).limit(limit)
        if category:
            stmt = stmt.where(Product.category == category)
        result = await self.db.execute(stmt)
        return list(result.scalars().all())

    async def create(self, product: ProductCreate) -> Product:
        db_product = Product(**product.model_dump())
        self.db.add(db_product)
        await self.db.flush()
        await self.db.refresh(db_product)
        return db_product

    async def update(self, product_id: int, data: dict) -> Product | None:
        stmt = (
            update(Product)
            .where(Product.id == product_id)
            .values(**data)
            .returning(Product)
        )
        result = await self.db.execute(stmt)
        return result.scalar_one_or_none()

    async def delete(self, product_id: int) -> bool:
        stmt = delete(Product).where(Product.id == product_id)
        result = await self.db.execute(stmt)
        return result.rowcount > 0

TEXT 📖 仅展示

# 函数定义成功

▶ 示例：PriceRepository

PYTHON

class PriceRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_by_product(
        self,
        product_id: int,
        min_price: float | None = None,
        max_price: float | None = None,
        skip: int = 0,
        limit: int = 50,
    ) -> list[Price]:
        stmt = (
            select(Price)
            .where(Price.product_id == product_id)
            .order_by(Price.recorded_at.desc())
            .offset(skip)
            .limit(limit)
        )
        if min_price is not None:
            stmt = stmt.where(Price.price >= min_price)
        if max_price is not None:
            stmt = stmt.where(Price.price <= max_price)
        result = await self.db.execute(stmt)
        return list(result.scalars().all())

    async def bulk_create(self, prices: list[PriceCreate]) -> list[Price]:
        db_prices = [Price(**p.model_dump()) for p in prices]
        self.db.add_all(db_prices)
        await self.db.flush()
        return db_prices

TEXT 📖 仅展示

# 函数定义成功

4. 分页与排序

(1) Pydantic 查询模型

将分页参数封装为 Pydantic 模型，避免在每个端点重复声明查询参数。

▶ 示例：分页查询模型

PYTHON

from pydantic import BaseModel, Field
from typing import Optional

class PaginationParams(BaseModel):
    skip: int = Field(0, ge=0, description="Number of records to skip")
    limit: int = Field(20, ge=1, le=100, description="Max records to return")

class ProductFilter(PaginationParams):
    category: Optional[str] = Field(None, max_length=100)
    min_price: Optional[float] = Field(None, ge=0, description="Min price in USD")
    max_price: Optional[float] = Field(None, ge=0, description="Max price in USD")
    sort_by: str = Field("name", pattern="^(name|base_price|created_at)$")
    sort_order: str = Field("asc", pattern="^(asc|desc)$")

TEXT 📖 仅展示

# 执行成功

▶ 示例：端点使用分页模型

PYTHON

from fastapi import FastAPI, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()

@app.get("/products", response_model=list[ProductResponse])
async def list_products(
    category: Optional[str] = Query(None),
    min_price: Optional[float] = Query(None, ge=0),
    max_price: Optional[float] = Query(None, ge=0),
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
    db: AsyncSession = Depends(get_db),
):
    repo = ProductRepository(db)
    return await repo.get_all(
        category=category,
        min_price=min_price,
        max_price=max_price,
        skip=skip,
        limit=limit,
    )

TEXT 📖 仅展示

# 函数定义成功

(2) 分页策略对比

策略	实现方式	优点	缺点
Offset 分页	`OFFSET n LIMIT m`	简单，支持跳页	大 offset 性能差
Cursor 分页	`WHERE id > cursor LIMIT m`	大数据集性能好	不支持跳页
Keyset 分页	`WHERE created_at < last LIMIT m`	排序字段分页好	需有序字段

5. 关联数据预加载

(1) N+1 查询问题

▶ 示例：N+1 问题演示

PYTHON

# BAD: N+1 queries - 1 query for products + N queries for prices
stmt = select(Product)
result = await db.execute(stmt)
products = result.scalars().all()
for p in products:
    # Each access triggers a separate query!
    print(len(p.prices))  # N queries

TEXT 📖 仅展示

# 执行成功

▶ 示例：selectinload 预加载

PYTHON

from sqlalchemy.orm import selectinload, joinedload

# GOOD: 2 queries total - products + prices (using SELECT IN)
stmt = select(Product).options(selectinload(Product.prices))
result = await db.execute(stmt)
products = result.scalars().all()
for p in products:
    print(len(p.prices))  # No extra queries

TEXT 📖 仅展示

# 执行成功

(2) selectinload vs joinedload

维度	selectinload	joinedload
SQL 策略	SELECT ... WHERE id IN (...)	LEFT OUTER JOIN
查询数	2 次查询	1 次查询
数据量	精确（无重复行）	JOIN 可能产生重复行
适用场景	一对多（集合）	多对一/一对一
性能	数据量大时更优	关联数据少时更优
去重	不需要	需要 `unique()`

▶ 示例：Product 查询含 prices 预加载

PYTHON

async def get_product_with_prices(
    self, product_id: int
) -> Product | None:
    stmt = (
        select(Product)
        .options(selectinload(Product.prices))
        .where(Product.id == product_id)
    )
    result = await self.db.execute(stmt)
    return result.scalar_one_or_none()

TEXT 📖 仅展示

# 函数定义成功

6. 事务管理

(1) 跨表事务操作

▶ 示例：批量价格导入事务

PYTHON

from sqlalchemy.ext.asyncio import AsyncSession

class PriceService:
    def __init__(self, db: AsyncSession):
        self.db = db
        self.price_repo = PriceRepository(db)
        self.product_repo = ProductRepository(db)

    async def bulk_import_prices(
        self,
        product_id: int,
        prices: list[PriceCreate],
    ) -> list[Price]:
        # Verify product exists
        product = await self.product_repo.get_by_id(product_id)
        if not product:
            raise HTTPException(status_code=404, detail="Product not found")

        # Validate all prices reference the same product
        for p in prices:
            p.product_id = product_id

        # Bulk insert within transaction (db session handles commit/rollback)
        db_prices = await self.price_repo.bulk_create(prices)

        # Update product base price to latest
        latest_price = db_prices[-1]
        await self.product_repo.update(
            product_id,
            {"base_price": latest_price.price},
        )

        return db_prices

TEXT 📖 仅展示

# 函数定义成功

▶ 示例：批量导入端点

PYTHON

@app.post("/products/{product_id}/prices/bulk", response_model=list[PriceResponse])
async def bulk_import(
    product_id: int = Path(gt=0),
    prices: list[PriceCreate] = ...,
    db: AsyncSession = Depends(get_db),
    user: dict = Depends(get_current_user),
):
    # Subscription check: Free users limited to 1000 prices
    if user["subscription"] == "free" and len(prices) > 1000:
        raise HTTPException(
            status_code=403,
            detail="Free plan limited to 1000 prices per import. Upgrade to Pro.",
        )
    
    service = PriceService(db)
    return await service.bulk_import_prices(product_id, prices)

TEXT 📖 仅展示

# 函数定义成功

❓ 常见问题

Q Repository 一定要用类吗？

A 不是必须，也可以用函数。但类封装 DB Session 状态，方法间共享 Session 更方便，推荐类写法。

Q flush 和 commit 有什么区别？

A flush 发送 SQL 到数据库但不提交事务，可以获取自增 ID 和触发数据库约束检查。commit 提交事务，变更持久化。在 yield 依赖中推荐 flush + 自动 commit。

Q N+1 问题只在循环中触发吗？

A 主要在循环访问关联属性时触发。但 Pydantic 序列化也会触发——如果 response_model 访问了关联字段。务必预加载。

Q bulk_create 有大小限制吗？

A 无框架限制，但 PostgreSQL 单次绑定参数上限 32767。千级插入没问题，万级建议分批。

Q Offset 分页在大数据集下有多慢？

A OFFSET 1000000 需要先扫描前 100 万行再跳过，查询时间与 offset 成正比。百万级数据建议用 cursor 分页。

Q 事务中部分失败怎么处理？

A yield 依赖的 rollback 在异常时自动触发。如果需要部分回滚，用 savepoint（await session.begin_nested()）。

📖 小节

仓储模式将 SQL 操作封装在 Repository 类中，路由函数只关注 HTTP 逻辑
分页参数封装为 Pydantic 模型，避免每个端点重复声明查询参数
selectinload 适合一对多预加载（无重复行），joinedload 适合多对一（单次查询）
批量操作在事务中执行，yield 依赖自动管理 commit/rollback
PriceTracker 的批量价格导入支持千级记录，Free 用户限制 1000 条

📝 作业

基础题（难度⭐）：实现 ProductRepository 的 get_by_id 和 get_all 方法，在端点中注入 DB Session 并调用。提示：select(Product).where(Product.id == product_id)
进阶题（难度⭐⭐）：添加分页查询支持（skip/limit 参数），实现 selectinload(Product.prices) 预加载，返回含价格列表的商品详情。提示：options(selectinload(...))
挑战题（难度⭐⭐⭐）：实现 PriceService.bulk_import_prices，验证商品存在、批量插入价格、更新商品 base_price，全部在同一事务中完成。Free 用户限制 1000 条。提示：bulk_create + update + Depends(get_current_user)

---|

Web-Tutorial 技术团队

由多位开发者共同维护的编程教程平台。每篇教程由对应领域的开发者编写和审核，确保内容准确可靠。如发现任何问题，欢迎向我们反馈。

关于我们

FastAPI: CRUD 完整实现 — 从仓储层到 API 端点

1. 你将学到

2. Alice 的真实故事

(1) 痛点：SQL 散落在路由函数中

(2) 仓储模式的解法

(3) 收益

3. 分层架构与仓储模式

(1) 四层架构

▶ 示例：ProductRepository

▶ 示例：PriceRepository

4. 分页与排序

(1) Pydantic 查询模型

▶ 示例：分页查询模型

▶ 示例：端点使用分页模型

(2) 分页策略对比

5. 关联数据预加载

(1) N+1 查询问题

▶ 示例：N+1 问题演示

▶ 示例：selectinload 预加载

(2) selectinload vs joinedload

▶ 示例：Product 查询含 prices 预加载

6. 事务管理

(1) 跨表事务操作

▶ 示例：批量价格导入事务

▶ 示例：批量导入端点

❓ 常见问题

📖 小节

📝 作业

🙏 帮我们做得更好