"""结业编码题参考实现：书架（bookshelf）。
规格见 exercises/final/quiz_final.md 第三部分。验证：
uv run pytest exercises/solutions/final/test_task_bookshelf.py -q
"""
from __future__ import annotations
import sqlite3
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import Protocol
from pydantic import BaseModel, ConfigDict, Field, field_validator
# ---- 领域 ----
class BookshelfError(Exception):
"""书架错误基类。"""
class BookNotFoundError(BookshelfError):
"""按 id 找不到书。"""
class DuplicateIsbnError(BookshelfError):
"""ISBN 重复。"""
@dataclass(frozen=True)
class Book:
"""领域模型：一本书（不可变）。"""
id: int | None
isbn: str
title: str
tags: tuple[str, ...]
added_at: datetime
# ---- 传输/校验模型 ----
class BookIn(BaseModel):
"""外部输入：ISBN 13 位数字、标题 1–200 字、标签 ≤5 个且小写去重。"""
model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")
isbn: str = Field(pattern=r"^\d{13}$")
title: str = Field(min_length=1, max_length=200)
tags: list[str] = Field(default_factory=list, max_length=5)
@field_validator("tags")
@classmethod
def _norm(cls, tags: list[str]) -> list[str]:
return list(dict.fromkeys(t.strip().lower() for t in tags if t.strip()))
# ---- 仓储 ----
class BookRepository(Protocol):
"""存储接口。"""
def add(self, book: Book) -> Book: ...
def get(self, book_id: int) -> Book | None: ...
def find_by_isbn(self, isbn: str) -> Book | None: ...
def list(self) -> list[Book]: ...
class InMemoryBookRepository:
"""测试用内存实现。"""
def __init__(self) -> None:
self._books: dict[int, Book] = {}
def add(self, book: Book) -> Book:
new_id = len(self._books) + 1
stored = Book(new_id, book.isbn, book.title, book.tags, book.added_at)
self._books[new_id] = stored
return stored
def get(self, book_id: int) -> Book | None:
return self._books.get(book_id)
def find_by_isbn(self, isbn: str) -> Book | None:
return next((b for b in self._books.values() if b.isbn == isbn), None)
def list(self) -> list[Book]:
return sorted(self._books.values(), key=lambda b: b.id or 0)
class SqliteBookRepository:
"""SQLite 实现：参数化查询 + with conn 事务。"""
def __init__(self, path: Path) -> None:
self.conn = sqlite3.connect(path)
self.conn.row_factory = sqlite3.Row
with self.conn:
self.conn.execute(
"CREATE TABLE IF NOT EXISTS books (id INTEGER PRIMARY KEY,"
" isbn TEXT UNIQUE, title TEXT, tags TEXT, added_at TEXT)"
)
def add(self, book: Book) -> Book:
with self.conn:
cur = self.conn.execute(
"INSERT INTO books (isbn, title, tags, added_at) VALUES (?, ?, ?, ?)",
(book.isbn, book.title, ",".join(book.tags), book.added_at.isoformat()),
)
return Book(
int(cur.lastrowid or 0), book.isbn, book.title, book.tags, book.added_at
)
def get(self, book_id: int) -> Book | None:
row = self.conn.execute(
"SELECT * FROM books WHERE id = ?", (book_id,)
).fetchone()
return self._to_book(row) if row else None
def find_by_isbn(self, isbn: str) -> Book | None:
row = self.conn.execute(
"SELECT * FROM books WHERE isbn = ?", (isbn,)
).fetchone()
return self._to_book(row) if row else None
def list(self) -> list[Book]:
rows = self.conn.execute("SELECT * FROM books ORDER BY id").fetchall()
return [self._to_book(r) for r in rows]
@staticmethod
def _to_book(row: sqlite3.Row) -> Book:
tags = tuple(t for t in str(row["tags"]).split(",") if t)
return Book(
row["id"],
row["isbn"],
row["title"],
tags,
datetime.fromisoformat(row["added_at"]),
)
# ---- 服务 ----
class Bookshelf:
"""用例：加书（校验 + 查重）、取书、按标签列出。"""
def __init__(self, repo: BookRepository) -> None:
self.repo = repo
def add(self, data: dict) -> Book:
"""dict -> BookIn 校验 -> 查重 -> 存储。校验失败抛 pydantic.ValidationError。"""
payload = BookIn.model_validate(data)
if self.repo.find_by_isbn(payload.isbn) is not None:
raise DuplicateIsbnError(f"ISBN {payload.isbn} 已存在")
book = Book(
None, payload.isbn, payload.title, tuple(payload.tags), datetime.now(UTC)
)
return self.repo.add(book)
def get(self, book_id: int) -> Book:
book = self.repo.get(book_id)
if book is None:
raise BookNotFoundError(f"书 {book_id} 不存在")
return book
def by_tag(self, tag: str) -> list[Book]:
tag = tag.strip().lower()
return [b for b in self.repo.list() if tag in b.tags]