7.2 数据层实现：模型与 CRUD

本节完成项目的"地基"三个文件：database.py、models.py、crud.py。代码量较大，但每一段都是前面章节学过的知识——留意注释里的章节标注。

一、app/database.py

与 6.4 节几乎一致，唯一变化是数据库文件名：

# app/database.py
from typing import Annotated, Generator

from fastapi import Depends
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker

engine = create_engine(
    "sqlite:///library.db",
    connect_args={"check_same_thread": False},   # SQLite + FastAPI 标配（6.2节）
    echo=True,                                   # 上线前改 False
)

SessionLocal = sessionmaker(bind=engine, autoflush=False)


class Base(DeclarativeBase):
    pass


def get_db() -> Generator[Session, None, None]:
    with SessionLocal() as session:
        yield session


DbSession = Annotated[Session, Depends(get_db)]

二、app/models.py

六张表全在这里。这是全项目知识密度最高的文件：

# app/models.py
from datetime import date, datetime
from typing import Optional

from sqlalchemy import Column, ForeignKey, String, Table, Text, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.database import Base


# ============ 多对多中间表：图书-分类（4.3节） ============
book_categories = Table(
    "book_categories",
    Base.metadata,
    Column("book_id", ForeignKey("books.id"), primary_key=True),
    Column("category_id", ForeignKey("categories.id"), primary_key=True),
)


class Author(Base):
    __tablename__ = "authors"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), index=True)
    bio: Mapped[Optional[str]] = mapped_column(Text)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    # 一对多：作者 → 图书（4.1节）。作者删除连带删书
    books: Mapped[list["Book"]] = relationship(
        back_populates="author", cascade="all, delete-orphan"
    )

    def __repr__(self) -> str:
        return f"Author(id={self.id}, name={self.name!r})"


class Category(Base):
    __tablename__ = "categories"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)

    # 多对多：分类 ↔ 图书。共享关系，不配 delete-orphan（4.3节）
    books: Mapped[list["Book"]] = relationship(
        secondary=book_categories, back_populates="categories"
    )

    def __repr__(self) -> str:
        return f"Category(id={self.id}, name={self.name!r})"


class Book(Base):
    __tablename__ = "books"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), index=True)
    isbn: Mapped[str] = mapped_column(String(20), unique=True)
    summary: Mapped[Optional[str]] = mapped_column(Text)
    total_copies: Mapped[int] = mapped_column(default=1)
    available_copies: Mapped[int] = mapped_column(default=1)   # 冗余字段，借还时事务维护
    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"), index=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    author: Mapped["Author"] = relationship(back_populates="books")
    categories: Mapped[list["Category"]] = relationship(
        secondary=book_categories, back_populates="books"
    )
    borrows: Mapped[list["Borrow"]] = relationship(back_populates="book")

    def __repr__(self) -> str:
        return f"Book(id={self.id}, title={self.title!r})"


class Reader(Base):
    __tablename__ = "readers"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(120), unique=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    borrows: Mapped[list["Borrow"]] = relationship(back_populates="reader")


class Borrow(Base):
    """借阅记录 —— 图书与读者之间的"关联对象"（4.3节第五部分）"""
    __tablename__ = "borrows"

    id: Mapped[int] = mapped_column(primary_key=True)
    book_id: Mapped[int] = mapped_column(ForeignKey("books.id"), index=True)
    reader_id: Mapped[int] = mapped_column(ForeignKey("readers.id"), index=True)
    borrowed_at: Mapped[datetime] = mapped_column(server_default=func.now())
    due_date: Mapped[date]                          # 应还日期
    returned_at: Mapped[Optional[datetime]]         # NULL = 未归还

    book: Mapped["Book"] = relationship(back_populates="borrows")
    reader: Mapped["Reader"] = relationship(back_populates="borrows")

    def __repr__(self) -> str:
        return f"Borrow(id={self.id}, book_id={self.book_id}, returned={self.returned_at is not None})"

三、app/schemas.py

按"资源 × 进出方向"组织。注意嵌套 Schema 的用法（6.3 节）：

# app/schemas.py
from datetime import date, datetime

from pydantic import BaseModel, ConfigDict, EmailStr, Field


# ================= 作者 =================
class AuthorCreate(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    bio: str | None = None


class AuthorOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str
    bio: str | None


# ================= 分类 =================
class CategoryCreate(BaseModel):
    name: str = Field(min_length=1, max_length=50)


class CategoryOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str


class CategoryWithCount(CategoryOut):
    """分类 + 藏书量（聚合查询结果）"""
    book_count: int


# ================= 图书 =================
class BookCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    isbn: str = Field(min_length=10, max_length=20)
    summary: str | None = None
    total_copies: int = Field(default=1, ge=1)
    author_id: int
    category_ids: list[int] = []


class BookUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    summary: str | None = None
    category_ids: list[int] | None = None


class BookOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    title: str
    isbn: str
    summary: str | None
    total_copies: int
    available_copies: int
    author: AuthorOut                    # 嵌套：自动转 book.author
    categories: list[CategoryOut]        # 嵌套：自动转 book.categories


class BookPage(BaseModel):
    total: int
    page: int
    page_size: int
    items: list[BookOut]


class AuthorDetail(AuthorOut):
    """作者详情：带全部作品"""
    class _BookBrief(BaseModel):
        model_config = ConfigDict(from_attributes=True)
        id: int
        title: str

    books: list[_BookBrief] = []


# ================= 读者与借阅 =================
class ReaderCreate(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    email: EmailStr


class ReaderOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str
    email: str


class BorrowCreate(BaseModel):
    book_id: int
    reader_id: int
    days: int = Field(default=30, ge=1, le=90, description="借期天数")


class BorrowOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    borrowed_at: datetime
    due_date: date
    returned_at: datetime | None
    book: "BookBrief"
    reader: ReaderOut


class BookBrief(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    title: str

四、app/crud.py（核心）

所有数据库操作集中在这里。函数只操作不 commit（flush 拿 id），提交权在路由层——这是 5.2 节"谁代表完整业务谁 commit"原则的落地：

# app/crud.py
from datetime import date, datetime, timedelta

from sqlalchemy import func, select
from sqlalchemy.orm import Session, selectinload, joinedload

from app import models


# ================= 作者 =================
def create_author(db: Session, name: str, bio: str | None) -> models.Author:
    author = models.Author(name=name, bio=bio)
    db.add(author)
    db.flush()
    return author


def get_author_with_books(db: Session, author_id: int) -> models.Author | None:
    stmt = (
        select(models.Author)
        .options(selectinload(models.Author.books))      # 预加载防 N+1（4.4节）
        .where(models.Author.id == author_id)
    )
    return db.scalar(stmt)


def list_authors(db: Session) -> list[models.Author]:
    return list(db.scalars(select(models.Author).order_by(models.Author.id)))


# ================= 分类 =================
def create_category(db: Session, name: str) -> models.Category:
    category = models.Category(name=name)
    db.add(category)
    db.flush()
    return category


def get_category_by_name(db: Session, name: str) -> models.Category | None:
    return db.scalar(select(models.Category).where(models.Category.name == name))


def list_categories_with_count(db: Session):
    """分类列表 + 各分类藏书量（outerjoin + group_by，5.1节）"""
    stmt = (
        select(
            models.Category.id,
            models.Category.name,
            func.count(models.book_categories.c.book_id).label("book_count"),
        )
        .outerjoin(models.book_categories)
        .group_by(models.Category.id)
        .order_by(models.Category.id)
    )
    return db.execute(stmt).all()


# ================= 图书 =================
def get_book(db: Session, book_id: int, with_relations: bool = True) -> models.Book | None:
    if not with_relations:
        return db.get(models.Book, book_id)
    stmt = (
        select(models.Book)
        .options(
            joinedload(models.Book.author),              # 单对象用 joinedload（4.4节）
            selectinload(models.Book.categories),        # 列表用 selectinload
        )
        .where(models.Book.id == book_id)
    )
    return db.scalar(stmt)


def get_book_by_isbn(db: Session, isbn: str) -> models.Book | None:
    return db.scalar(select(models.Book).where(models.Book.isbn == isbn))


def create_book(db: Session, *, title: str, isbn: str, summary: str | None,
                total_copies: int, author_id: int,
                categories: list[models.Category]) -> models.Book:
    book = models.Book(
        title=title,
        isbn=isbn,
        summary=summary,
        total_copies=total_copies,
        available_copies=total_copies,       # 初始全部可借
        author_id=author_id,
        categories=categories,               # 多对多直接赋列表（4.3节）
    )
    db.add(book)
    db.flush()
    return book


def search_books(db: Session, *, page: int, page_size: int,
                 keyword: str | None = None,
                 category_id: int | None = None,
                 author_id: int | None = None):
    """图书搜索：动态条件 + 分页（3.3节）"""
    stmt = select(models.Book)

    if keyword:
        stmt = stmt.where(models.Book.title.contains(keyword))
    if author_id:
        stmt = stmt.where(models.Book.author_id == author_id)
    if category_id:
        # 多对多关系过滤：any（4.3节）
        stmt = stmt.where(models.Book.categories.any(models.Category.id == category_id))

    total = db.scalar(select(func.count()).select_from(stmt.subquery()))

    items = list(db.scalars(
        stmt.options(
            joinedload(models.Book.author),
            selectinload(models.Book.categories),
        )
        .order_by(models.Book.id.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
    ))
    return total, items


# ================= 借阅（含业务规则的事务操作，5.2节） =================
def borrow_book(db: Session, *, book: models.Book, reader_id: int,
                days: int) -> models.Borrow:
    """借书：减库存 + 建记录。调用方保证在同一事务中 commit。"""
    book.available_copies -= 1               # 脏数据追踪自动生成 UPDATE（3.4节）
    borrow = models.Borrow(
        book_id=book.id,
        reader_id=reader_id,
        due_date=date.today() + timedelta(days=days),
    )
    db.add(borrow)
    db.flush()
    return borrow


def return_book(db: Session, borrow: models.Borrow) -> models.Borrow:
    """还书：记归还时间 + 回补库存"""
    borrow.returned_at = datetime.now()
    borrow.book.available_copies += 1
    return borrow


def get_borrow(db: Session, borrow_id: int) -> models.Borrow | None:
    stmt = (
        select(models.Borrow)
        .options(joinedload(models.Borrow.book), joinedload(models.Borrow.reader))
        .where(models.Borrow.id == borrow_id)
    )
    return db.scalar(stmt)


def list_borrows(db: Session, *, only_open: bool = False, only_overdue: bool = False):
    stmt = select(models.Borrow).options(
        joinedload(models.Borrow.book), joinedload(models.Borrow.reader)
    )
    if only_open or only_overdue:
        stmt = stmt.where(models.Borrow.returned_at.is_(None))   # IS NULL（3.3节）
    if only_overdue:
        stmt = stmt.where(models.Borrow.due_date < date.today())
    return list(db.scalars(stmt.order_by(models.Borrow.id.desc())))

💡 读代码建议：search_books 和 borrow_book 是最值得精读的两个函数——前者是"动态查询 + 预加载 + 分页"的集大成模板，后者展示了业务规则如何依赖事务保证一致性。

📝 本节小结

• models.py 一次性用上了一对多、多对多、关联对象三种关系
• 所有"查了要访问关系"的函数都显式预加载（joinedload/selectinload）
• crud 函数不 commit（最多 flush 拿 id），事务边界留给路由层
• 借书/还书对 available_copies 的维护必须和借阅记录同事务

下一节写路由层并测试 → 7.3 接口层实现：路由与测试