（十七）综合示例：可替换存储器的书架

1. 任务、接口与状态划分

任务（自拟）：在上一章菜单上，维护书名、已读状态和标签。支持加入、完成、添加标签、显示清单、保存快照、重新载入与退出。启动时用 --store memory 或 --store dict 选择存储器；两种方式提供同样的 load、save 接口。

数据保存在本次运行的存储器中。save 用当前书架替换存储快照，reload 用已保存快照替换当前书架；退出结束本次运行。这样先看清对象之间的保存关系，下一章再把快照写进文件。

层次	保存的数据与负责的动作
`Book`	一条记录的书名、状态与独立标签列表
`BookShelf`	当前书架，负责加入、完成和标签操作
`MemoryStore`	用书籍对象列表保存独立快照
`DictStore`	用书名索引的字典保存状态与标签元组
命令行入口	解析命令，选择存储器，显示方法返回的结果

存储器读取和写入时建立独立的数据快照。否则，书架修改一个标签列表，就可能连尚未执行 save 的存储内容也一起改变。

目录保存为 bookshelf_project，在它的目录下启动包：

bookshelf_project/
└── bookshelf/
    ├── __init__.py
    ├── __main__.py
    ├── models.py
    ├── interfaces.py
    ├── stores.py
    ├── shelf.py
    ├── views.py
    ├── commands.py
    └── cli.py

导入从入口走向业务与数据：__main__ → cli → commands / shelf / stores，书架与存储器再使用 interfaces、models，显示函数放在 views。计算和存储模块不反过来导入命令行。

2. 书籍记录与存储协议

# 项目文件：bookshelf_project/bookshelf/models.py
# 每本书初始化时由 list 创建自己的标签；完成状态由实例方法更新。
from dataclasses import dataclass, field


@dataclass
class Book:
    title: str
    finished: bool = False
    tags: list[str] = field(default_factory=list)

    def finish(self) -> None:
        self.finished = True

# 项目文件：bookshelf_project/bookshelf/interfaces.py
# 协议只说明书架需要的两个动作，具体存储器可以不继承它。
from typing import Protocol

from .models import Book


class BookStore(Protocol):
    def load(self) -> list[Book]:
        ...

    def save(self, books: list[Book]) -> None:
        ...

除了这些方法的形状，两个实现还共同约定：载入交回独立记录，保存接收一份新快照。这份数据所有权约定要由实现完成，类型注解不会自动复制数据。

3. 两种存储器的具体实现

# 项目文件：bookshelf_project/bookshelf/stores.py
# deepcopy 连同 Book 实例内部的标签列表一起复制，避免只复制外层列表。
from copy import deepcopy

from .models import Book


class MemoryStore:
    def __init__(self):
        self._books: list[Book] = []

    def load(self) -> list[Book]:
        return deepcopy(self._books)

    def save(self, books: list[Book]) -> None:
        self._books = deepcopy(books)


class DictStore:
    def __init__(self):
        # 书名作键，值为“已读状态、标签元组”。
        self._rows: dict[str, tuple[bool, tuple[str, ...]]] = {}

    def load(self) -> list[Book]:
        # 从存储表示重新建立 Book 与可修改的标签列表。
        return [
            Book(title, finished, list(tags))
            for title, (finished, tags) in self._rows.items()
        ]

    def save(self, books: list[Book]) -> None:
        # 转成独立的元组记录，同时保留书架的加入顺序。
        self._rows = {
            book.title: (book.finished, tuple(book.tags))
            for book in books
        }

内存存储器用 deepcopy 复制对象与其中的列表，字典存储器则主动把记录转换成布尔值与字符串元组，再在载入时建立新实例。两者内部表示不同，书架却只调用同一份协议中的方法。对象复制

4. 书架对象的记录操作

# 项目文件：bookshelf_project/bookshelf/shelf.py
# 书架持有存储器，所有修改先发生在当前记录中。
from collections.abc import Iterator
from copy import deepcopy

from .interfaces import BookStore
from .models import Book


class BookShelf:
    def __init__(self, store: BookStore):
        self._store = store
        self.reload()

    def add(self, title: str) -> str:
        if title in self._books:
            return f"已经存在 {title}"
        self._books[title] = Book(title)
        return f"已加入 {title}"

    def finish(self, title: str) -> str:
        if title not in self._books:
            return f"没有找到 {title}"
        self._books[title].finish()
        return f"已标记 {title}"

    def add_tag(self, title: str, tag: str) -> str:
        if title not in self._books:
            return f"没有找到 {title}"
        tags = self._books[title].tags
        if tag not in tags:
            tags.append(tag)
        return f"已添加标签 {title}: {tag}"

    def save(self) -> None:
        self._store.save(list(self._books.values()))

    def reload(self) -> None:
        self._books = {book.title: book for book in self._store.load()}

    def snapshot(self) -> list[Book]:
        return deepcopy(list(self._books.values()))

    def __len__(self) -> int:
        return len(self._books)

    def __iter__(self) -> Iterator[Book]:
        # 对外遍历的是一份快照，修改拿到的 Book 不会绕过书架操作。
        return iter(self.snapshot())

    def __contains__(self, title: str) -> bool:
        return title in self._books

这里 self.reload() 先从存储器载入初始数据，并建立 _books。记录操作从字典定位 Book，再调用它自己的 finish()；书架不再到处重复“把完成字段设成真”。

__iter__ 返回快照的迭代器，因此两次遍历各有进度，对外得到的记录也与内部记录分开。快照会复制所有记录，占用额外空间；本例选择它，是为了让公开读取不直接交出内部可变对象。

5. 命令分派与入口文件

命令中的书名可以包含空格；tag 用第一个 | 分开书名和标签，例如 tag Python 入门|基础。把显示处理单独放进函数，再由入口决定调用哪个操作：

# 项目文件：bookshelf_project/bookshelf/views.py
# 显示函数读取快照，将记录转成面向用户的文字。
from .shelf import BookShelf


def show_books(shelf: BookShelf) -> None:
    if not shelf:
        print("清单为空")
        return
    for index, book in enumerate(shelf, start=1):
        status = "已读" if book.finished else "未读"
        # join 用逗号连接标签，只在有标签时补上显示内容。
        suffix = " [" + ", ".join(book.tags) + "]" if book.tags else ""
        print(f"{index} {book.title} {status}{suffix}")

再把一行命令对应到具体操作。返回值决定外面的读取循环是否继续：

# 项目文件：bookshelf_project/bookshelf/commands.py
# 命令处理不选择存储器，接收的始终是一份已经建立好的书架。
from .shelf import BookShelf
from .views import show_books


def execute(shelf: BookShelf, line: str) -> bool:
    parts = line.strip().split(maxsplit=1)
    if not parts:
        print("请输入命令")
        return True
    command = parts[0]
    argument = parts[1] if len(parts) == 2 else ""
    if command in {"add", "done", "tag"}:
        if not argument:
            print("请在命令后填写书名")
        elif command == "add":
            print(shelf.add(argument))
        elif command == "done":
            print(shelf.finish(argument))
        else:
            fields = argument.split("|", maxsplit=1)
            if len(fields) != 2 or not all(field.strip() for field in fields):
                print("标签命令格式：tag 书名|标签")
            else:
                print(shelf.add_tag(fields[0].strip(), fields[1].strip()))
    elif command in {"list", "save", "reload", "quit"}:
        if argument:
            print("这条命令不需要书名")
        elif command == "list":
            show_books(shelf)
        elif command == "save":
            shelf.save()
            print("已保存快照")
        elif command == "reload":
            shelf.reload()
            print("已重新载入")
        else:
            print("已退出")
            return False
    else:
        print("未知命令")
    return True

最后由启动入口选择存储器，再循环交入文本：

# 项目文件：bookshelf_project/bookshelf/cli.py
# 入口选择实现，书架方法负责状态，命令函数负责分派。
import argparse

from .commands import execute
from .shelf import BookShelf
from .stores import DictStore, MemoryStore


def main() -> None:
    parser = argparse.ArgumentParser(description="书架菜单")
    parser.add_argument("--store", choices=("memory", "dict"), default="memory")
    args = parser.parse_args()
    store = MemoryStore() if args.store == "memory" else DictStore()
    shelf = BookShelf(store)
    while execute(shelf, input()):
        pass  # 一轮命令已经在 execute 中处理，继续读下一行。

execute 返回是否继续读取命令，收到 quit 时返回 False，外面的循环结束。它不把多个职责塞进一个长的 main；显示、命令解析和书架修改仍然有各自的入口。

argparse 的 choices 指定可选择的文本，解析器会在启动时检查 --store。选定后只在入口建立对应存储器，再把它交给书架。后面的记录操作不用按存储种类重复写分支。

# 项目文件：bookshelf_project/bookshelf/__init__.py
# 包初始化只保留说明，不主动读取命令或修改书架。
"""可替换存储器的书架示例。"""

# 项目文件：bookshelf_project/bookshelf/__main__.py
# 与上一章相同，通过 python -m bookshelf 启动包入口。
from .cli import main

if __name__ == "__main__":
    main()

6. 运行、逐步核对与工作量

在 bookshelf_project 目录中使用 Python 3.12 环境运行：

# 两种启动方式使用相同的命令菜单，只替换存储器实现。
python -m bookshelf --store memory
python -m bookshelf --store dict

每次启动一份程序，输入下面这一组命令：

add Python 入门
done Python 入门
tag Python 入门|基础
save
add NumPy
list
reload
list
quit

两种方式都输出：

已加入 Python 入门
已标记 Python 入门
已添加标签 Python 入门: 基础
已保存快照
已加入 NumPy
1 Python 入门 已读 [基础]
2 NumPy 未读
已重新载入
1 Python 入门 已读 [基础]
已退出

save 时快照只含 Python；之后 NumPy 先留在当前书架。reload 重新创建当前记录，因此 NumPy 消失，Python 的状态与标签按已保存快照恢复。

也可以绕开命令行，直接检查对象协作：

# 项目文件：bookshelf_project/use_shelf.py
# 两份书架共用一个存储器，但分别持有当前记录。
from bookshelf.shelf import BookShelf
from bookshelf.stores import DictStore

store = DictStore()
first = BookShelf(store)
first.add("Python")
first.add_tag("Python", "基础")
first.save()
second = BookShelf(store)
outside = second.snapshot()
outside[0].tags.append("外部修改")
print(second.snapshot()[0].tags)
first.finish("Python")
print(second.snapshot()[0].finished)
first.save()
second.reload()
print(second.snapshot()[0].finished)

运行 python use_shelf.py 得到 ['基础']、False、True。先修改读取副本，第二份书架不变；再修改第一份书架，第二份也还未变化；经过保存与重新载入后，第二份才读到新状态。

设共有 n 本书、t 个标签。通过书名字典定位记录的平均工作量为 $O(1)$ ，检查一本书的重复标签与其标签数有关。保存、载入和完整遍历快照都要复制记录与标签，工作量及额外空间按 $O(n+t)$ 计算；文字本身的字符数量另随内容增长。与第三章的流式统计不同，这个书架要保留每本书的可查询状态，因此数据规模会直接进入存储空间。