（二十）把并发读取接回 SQLite 书籍库

1. 先安排任务与资源的使用范围

这次任务读取三本书，最多同时进行两个操作，保留输入顺序。任何一项请求失败都先完成取消和关闭，不进入数据库导入；网络全部成功后，仍由第五章的记录检查决定字段能否建立 Book。

network_project/
├── pyproject.toml、uv.lock、requirements-dev.txt
├── src/
│   ├── labserver/         本机固定接口、延迟与状态观察
│   ├── bookdata/          第五章的记录检查与 SQLite
│   └── netreading/
│       ├── __init__.py
│       ├── __main__.py
│       ├── client.py      并发、超时、状态与正文解析
│       ├── service.py     校验整批记录，再进入数据库
│       └── cli.py         参数、日志与退出状态
├── bench_waits.py
└── tests/

用一次 AsyncClient 管理连接，用一次 TaskGroup 管理本批任务，在每项操作内取得信号量。整体时间范围还包含等待名额和退出收尾。书籍数量很多时，可以把这层调度换成第十一节的固定工作任务与有界队列。

2. 请求成功、解析与任务组共同推进

# 项目文件：network_project/src/netreading/client.py
# 每个任务在同一客户端上请求；路径随错误保留，取消继续向外传播。
import asyncio
import logging
import math

import httpx

logger = logging.getLogger(__name__)


class FetchError(RuntimeError):
    """带有请求路径的读取失败。"""


async def fetch_records(
    base_url: str,
    paths: list[str],
    concurrency: int = 2,
    deadline: float = 2,
    request_timeout: float = 1,
) -> list[object]:
    if concurrency < 1:
        raise ValueError("并发数应至少为 1")
    if not all(
        math.isfinite(value) and value > 0 for value in (deadline, request_timeout)
    ):
        raise ValueError("请求超时与整批时间应为正的有限秒数")
    if not paths:
        return []
    semaphore = asyncio.Semaphore(concurrency)
    limits = httpx.Limits(
        max_connections=concurrency, max_keepalive_connections=concurrency
    )
    # 总时间覆盖等待名额、网络操作和上下文退出；单次网络等待另设超时。
    async with asyncio.timeout(deadline):
        async with httpx.AsyncClient(
            base_url=base_url,
            timeout=httpx.Timeout(request_timeout),
            limits=limits,
            follow_redirects=True,
            trust_env=False,
        ) as client:

            async def read_one(path):
                async with semaphore:
                    try:
                        response = await client.get(path)
                        # 在任务内部检查状态和正文，一项失败即可触发任务组的收尾。
                        response.raise_for_status()
                        record = response.json()
                    except (httpx.HTTPError, ValueError) as cause:
                        raise FetchError(
                            f"{path}：{type(cause).__name__} {cause}"
                        ) from cause
                    logger.info("完成读取 path=%s", path)
                    return record

            # 先等待本批任务退出，之后才会离开客户端并关闭连接池。
            async with asyncio.TaskGroup() as group:
                tasks = [group.create_task(read_one(path)) for path in paths]
    # Task 的建立次序与 paths 相同，按这份列表收集而非按完成次序拼接。
    return [task.result() for task in tasks]

并发数为 0 会让所有任务都等不到名额，因此在发请求前反馈输入错误；无穷大和 NaN 也不拿来安排时间。空路径列表直接交回空结果。

网络调用、状态检查和 JSON 解析都在 read_one 中，所以这些失败会立即进入 TaskGroup 的异常传播。FetchError 保存路径，异常链继续保留底层错误；没有接住 CancelledError，取消能顺着原来的等待关系传出。

退出顺序是先等待任务组收尾，再关闭客户端连接，最后离开总时间范围。tasks 列表按 paths 建立，因此即使第二项先完成，返回数据仍按第一、第二、第三项排列。

3. 校验与数据库保存继续复用已有实现

# 项目文件：network_project/src/netreading/service.py
# asyncio.run 结束后才建立数据库连接；所有书籍字段先统一校验。
import asyncio
from pathlib import Path

from bookdata.library import Library
from bookdata.records import validate_books

from .client import fetch_records


def import_remote(
    base_url: str,
    paths: list[str],
    database: Path,
    concurrency: int = 2,
    deadline: float = 2,
    request_timeout: float = 1,
) -> int:
    records = asyncio.run(
        fetch_records(base_url, paths, concurrency, deadline, request_timeout)
    )
    books = validate_books(records)
    with Library(database) as library:
        return library.import_books(books)

请求成功还要检查布尔状态、标题与标签。数据库连接由最后负责写入的调用层在当前线程中取得；工作任务不同时拿着同一份 SQLite 连接修改。已有书的导入仍保留编号和阅读日志，数据库约束失败仍由一次导入事务整体回滚。

第二项状态失败时取消同组等待，另一种路径则在字段校验失败后停止，两者都未开始写库

4. 命令入口记录具体请求与退出原因

# 项目文件：network_project/src/netreading/cli.py
# 参数交给服务层，异常组递归展开为简短反馈，日志保存完整的调用关系。
import argparse
import logging
from pathlib import Path
import sqlite3
import sys

from .service import import_remote

logger = logging.getLogger(__name__)
DEFAULT_PATHS = [
    "/api/books/1?delay=0.06",
    "/api/books/2?delay=0.02",
    "/api/books/3?delay=0.04",
]


def describe(error):
    if isinstance(error, BaseExceptionGroup):
        return "; ".join(describe(child) for child in error.exceptions)
    return str(error)


def main(argv=None):
    parser = argparse.ArgumentParser(description="并发读取并导入书目")
    parser.add_argument("paths", nargs="*", default=DEFAULT_PATHS)
    parser.add_argument("--base-url", default="http://127.0.0.1:8765")
    parser.add_argument("--db", type=Path, default=Path("data/network.sqlite3"))
    parser.add_argument("--concurrency", type=int, default=2)
    parser.add_argument("--deadline", type=float, default=2)
    parser.add_argument("--request-timeout", type=float, default=1)
    args = parser.parse_args(argv)
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s %(message)s")
    try:
        count = import_remote(
            args.base_url,
            args.paths,
            args.db,
            args.concurrency,
            args.deadline,
            args.request_timeout,
        )
    except (ExceptionGroup, TimeoutError, OSError, ValueError, sqlite3.Error) as error:
        logger.exception("本次书目导入失败")
        print("错误：" + (describe(error) or type(error).__name__), file=sys.stderr)
        return 1
    print("已导入", count, "本")
    return 0

nargs="*" 接收零个或多个路径参数，没有输入时使用三条默认路径。describe 用第三章的递归处理异常组，沿成员找到具体路径与错误；日志则保留完整异常链。超时、请求失败与校验错误都返回非零状态，方便前一章的自动检查发现失败。

# 项目文件：network_project/src/netreading/__init__.py
# 导入包时不建立事件循环、连接或数据库。
"""并发读取与书籍库导入。"""

# 项目文件：network_project/src/netreading/__main__.py
# 包入口与 reading-fetch 命令共用 main 的进程退出状态。
from .cli import main

if __name__ == "__main__":
    raise SystemExit(main())

5. 用多次启动检查数据的去向

先在一个终端运行本地服务，另一个终端进入同一个项目环境：

# 服务终端：等待客户端请求，Ctrl+C 结束。
uv run python -m labserver --port 8765

# 客户端终端：并发取得书目，再由一次数据库导入保存。
uv run reading-fetch --concurrency 2
uv run python -m bookdata --db data/network.sqlite3 list
# 独立导出已保存的书目，观察文件格式仍与第五章相同。
uv run python -m bookdata --db data/network.sqlite3 export-json data/network.json
# 选择一个失败路径，查看请求失败的日志与进程退出结果。
uv run reading-fetch /api/books/1 /api/status/503

正常导入显示三本书：Python 未读、NumPy 已读、Pygame 未读，阅读时长均为 0。失败路径返回 503 后，不会把这次仅成功取到的那一本当成完整导入。已经存在的书目保持原来的状态。