7.3 接口层实现：路由与测试

本节完成 routers 三个文件和 main.py，跑通整个系统，最后用 pytest 写接口测试。

一、app/routers/authors.py

APIRouter 是"迷你 app"，把同一资源的路由聚在一个文件里，最后统一挂到主应用：

# app/routers/authors.py
from fastapi import APIRouter, HTTPException

from app import crud, schemas
from app.database import DbSession

router = APIRouter(prefix="/authors", tags=["作者"])


@router.post("", response_model=schemas.AuthorOut, status_code=201)
def create_author(data: schemas.AuthorCreate, db: DbSession):
    author = crud.create_author(db, name=data.name, bio=data.bio)
    db.commit()                      # 事务边界在路由层（7.2节的约定）
    return author


@router.get("", response_model=list[schemas.AuthorOut])
def list_authors(db: DbSession):
    return crud.list_authors(db)


@router.get("/{author_id}", response_model=schemas.AuthorDetail)
def get_author(author_id: int, db: DbSession):
    author = crud.get_author_with_books(db, author_id)
    if author is None:
        raise HTTPException(404, "作者不存在")
    return author

• prefix="/authors"：本文件所有路径自动带上前缀
• tags=["作者"]：/docs 文档里自动按标签分组

二、app/routers/books.py

# app/routers/books.py
from fastapi import APIRouter, HTTPException, Query

from app import crud, models, schemas
from app.database import DbSession

router = APIRouter(tags=["图书与分类"])


# ---------- 分类 ----------
@router.post("/categories", response_model=schemas.CategoryOut, status_code=201)
def create_category(data: schemas.CategoryCreate, db: DbSession):
    if crud.get_category_by_name(db, data.name):
        raise HTTPException(409, "分类已存在")
    category = crud.create_category(db, data.name)
    db.commit()
    return category


@router.get("/categories", response_model=list[schemas.CategoryWithCount])
def list_categories(db: DbSession):
    rows = crud.list_categories_with_count(db)
    return [{"id": r.id, "name": r.name, "book_count": r.book_count} for r in rows]


# ---------- 图书 ----------
@router.post("/books", response_model=schemas.BookOut, status_code=201)
def create_book(data: schemas.BookCreate, db: DbSession):
    if crud.get_book_by_isbn(db, data.isbn):
        raise HTTPException(409, "ISBN 已存在")
    if db.get(models.Author, data.author_id) is None:
        raise HTTPException(404, "作者不存在")

    categories = [
        c for cid in data.category_ids
        if (c := db.get(models.Category, cid)) is not None
    ]
    book = crud.create_book(
        db,
        title=data.title, isbn=data.isbn, summary=data.summary,
        total_copies=data.total_copies, author_id=data.author_id,
        categories=categories,
    )
    db.commit()
    return crud.get_book(db, book.id)        # 重查一次，带上预加载的关系


@router.get("/books", response_model=schemas.BookPage)
def search_books(
    db: DbSession,
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=10, ge=1, le=100),
    keyword: str | None = None,
    category_id: int | None = None,
    author_id: int | None = None,
):
    total, items = crud.search_books(
        db, page=page, page_size=page_size,
        keyword=keyword, category_id=category_id, author_id=author_id,
    )
    return {"total": total, "page": page, "page_size": page_size, "items": items}


@router.get("/books/{book_id}", response_model=schemas.BookOut)
def get_book(book_id: int, db: DbSession):
    book = crud.get_book(db, book_id)
    if book is None:
        raise HTTPException(404, "图书不存在")
    return book


@router.patch("/books/{book_id}", response_model=schemas.BookOut)
def update_book(book_id: int, data: schemas.BookUpdate, db: DbSession):
    book = crud.get_book(db, book_id)
    if book is None:
        raise HTTPException(404, "图书不存在")

    updates = data.model_dump(exclude_unset=True)         # PATCH 语义（6.3节）
    if "category_ids" in updates:
        ids = updates.pop("category_ids")
        book.categories = [
            c for cid in ids if (c := db.get(models.Category, cid)) is not None
        ]
    for key, value in updates.items():
        setattr(book, key, value)
    db.commit()
    return crud.get_book(db, book_id)


@router.delete("/books/{book_id}", status_code=204)
def delete_book(book_id: int, db: DbSession):
    book = db.get(models.Book, book_id)
    if book is None:
        raise HTTPException(404, "图书不存在")
    db.delete(book)
    db.commit()

三、app/routers/borrows.py（业务最重的一支）

# app/routers/borrows.py
from fastapi import APIRouter, HTTPException

from app import crud, models, schemas
from app.database import DbSession

router = APIRouter(tags=["读者与借阅"])


@router.post("/readers", response_model=schemas.ReaderOut, status_code=201)
def create_reader(data: schemas.ReaderCreate, db: DbSession):
    reader = models.Reader(**data.model_dump())
    db.add(reader)
    db.commit()
    return reader


@router.post("/borrows", response_model=schemas.BorrowOut, status_code=201)
def borrow_book(data: schemas.BorrowCreate, db: DbSession):
    """借书：库存检查 + 减库存 + 建记录，一个事务"""
    book = db.get(models.Book, data.book_id)
    if book is None:
        raise HTTPException(404, "图书不存在")
    if db.get(models.Reader, data.reader_id) is None:
        raise HTTPException(404, "读者不存在")
    if book.available_copies <= 0:
        raise HTTPException(409, "该书已无可借库存")

    borrow = crud.borrow_book(db, book=book, reader_id=data.reader_id, days=data.days)
    db.commit()          # 减库存 + 借阅记录一起生效（5.2节：事务原子性）
    return crud.get_borrow(db, borrow.id)


@router.post("/borrows/{borrow_id}/return", response_model=schemas.BorrowOut)
def return_book(borrow_id: int, db: DbSession):
    borrow = crud.get_borrow(db, borrow_id)
    if borrow is None:
        raise HTTPException(404, "借阅记录不存在")
    if borrow.returned_at is not None:
        raise HTTPException(409, "这本书已经还过了")

    crud.return_book(db, borrow)
    db.commit()
    return borrow


@router.get("/borrows", response_model=list[schemas.BorrowOut])
def list_borrows(db: DbSession, only_open: bool = False, only_overdue: bool = False):
    return crud.list_borrows(db, only_open=only_open, only_overdue=only_overdue)

📌 留给你的练习：create_reader 目前没做邮箱查重——重复邮箱会撞唯一约束抛 500。请仿照图书的 ISBN 查重，在 crud 层加一个 get_reader_by_email，路由里查到重复返回 409。（做完可以用 7.3 末尾的测试思路给它补一个测试用例。）

四、app/main.py：组装

# app/main.py
from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.database import Base, engine
from app.routers import authors, books, borrows


@asynccontextmanager
async def lifespan(app: FastAPI):
    Base.metadata.create_all(bind=engine)     # 正式项目换成 Alembic（5.3节）
    yield
    engine.dispose()


app = FastAPI(
    title="图书管理系统 API",
    description="SQLAlchemy + FastAPI 教程实战项目",
    lifespan=lifespan,
)

app.include_router(authors.router)
app.include_router(books.router)
app.include_router(borrows.router)


@app.get("/", tags=["根"])
def root():
    return {"message": "图书管理系统 API，文档见 /docs"}

五、启动与冒烟测试

在 library_api 目录下：

fastapi dev app/main.py

打开 /docs 按业务流程走一遍：

1. POST /authors：{"name": "刘慈欣", "bio": "科幻作家"}
2. POST /categories：{"name": "科幻"}
3. POST /books：{"title": "三体", "isbn": "9787536692930", "total_copies": 2, "author_id": 1, "category_ids": [1]}
4. POST /readers：{"name": "小明", "email": "xm@example.com"}
5. POST /borrows 借两次 → 第三次应返回 409 无库存 ✔
6. GET /books/1 → available_copies 应为 0 ✔
7. POST /borrows/1/return 还书 → 再查库存应为 1；重复还 → 409 ✔
8. GET /borrows?only_open=true 查未还记录

六、接口测试：pytest + TestClient

手点 docs 只能测一次，自动化测试才能反复保障质量。用 6.2 节的"依赖覆盖"让测试跑在内存数据库上：

# tests/test_api.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.database import Base, get_db
from app.main import app

# ---------- 测试专用：内存数据库 ----------
test_engine = create_engine(
    "sqlite://",                                  # 纯内存
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,                         # 保证所有会话共用同一个内存库
)
TestSession = sessionmaker(bind=test_engine)


def override_get_db():
    with TestSession() as session:
        yield session


app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)


@pytest.fixture(autouse=True)
def fresh_db():
    """每个测试用例前重建全部表，保证互不干扰"""
    Base.metadata.create_all(bind=test_engine)
    yield
    Base.metadata.drop_all(bind=test_engine)


def _setup_book(copies: int = 1) -> int:
    """造数据的辅助函数，返回 book_id"""
    author = client.post("/authors", json={"name": "刘慈欣"}).json()
    client.post("/categories", json={"name": "科幻"})
    book = client.post("/books", json={
        "title": "三体", "isbn": "9787536692930",
        "total_copies": copies, "author_id": author["id"], "category_ids": [1],
    }).json()
    return book["id"]


def test_create_and_get_book():
    book_id = _setup_book()
    resp = client.get(f"/books/{book_id}")
    assert resp.status_code == 200
    data = resp.json()
    assert data["title"] == "三体"
    assert data["author"]["name"] == "刘慈欣"
    assert data["categories"][0]["name"] == "科幻"


def test_duplicate_isbn_409():
    _setup_book()
    resp = client.post("/books", json={
        "title": "三体二", "isbn": "9787536692930",
        "total_copies": 1, "author_id": 1, "category_ids": [],
    })
    assert resp.status_code == 409


def test_borrow_and_stock():
    book_id = _setup_book(copies=1)
    client.post("/readers", json={"name": "小明", "email": "xm@example.com"})

    # 第一次借：成功，库存归零
    resp = client.post("/borrows", json={"book_id": book_id, "reader_id": 1})
    assert resp.status_code == 201
    assert client.get(f"/books/{book_id}").json()["available_copies"] == 0

    # 第二次借：无库存
    resp = client.post("/borrows", json={"book_id": book_id, "reader_id": 1})
    assert resp.status_code == 409

    # 还书：库存回补；重复还报 409
    assert client.post("/borrows/1/return").status_code == 200
    assert client.get(f"/books/{book_id}").json()["available_copies"] == 1
    assert client.post("/borrows/1/return").status_code == 409


def test_search_books():
    _setup_book()
    assert client.get("/books?keyword=三体").json()["total"] == 1
    assert client.get("/books?keyword=不存在的书").json()["total"] == 0
    assert client.get("/books?category_id=1").json()["total"] == 1

运行：

pip install pytest
pytest tests/ -v

看到 4 个 PASSED，整个系统的核心流程就有了自动化保障。

📝 本节小结

• APIRouter + include_router 按资源拆分路由，tags 自动整理文档
• 事务边界在路由层：一个接口一次 commit，业务规则（库存检查）在 commit 前完成
• 测试三板斧：内存 SQLite + dependency_overrides + 每用例重建表
• 测试覆盖了正常流、409 冲突、库存一致性——这正是事务设计的验收

最后一节收尾 → 7.4 最佳实践与下一步