Python 现代开发学习笔记

(二十一)初始化数据并运行完整项目

1. 导入之前采集的书目

配套项目的 data/books.csv 沿用第九章采集、第十一章分析的六本书。初始化时为 alice 建立这些书籍,为 bob 建立空书架;重复执行不会重置已有账号密码,也不会重复插入同名书。

# 项目文件:api_project/src/bookapi/cli.py
# 在同一事务中准备演示账户并导入 CSV;每次写入前先查找同一账户的同名书籍。
import argparse
import csv
import json
from pathlib import Path

import uvicorn
from sqlalchemy import select
from werkzeug.security import generate_password_hash

from .config import Settings
from .database import open_database
from .models import Book, User
from .schemas import BookCreate


def initialize(settings: Settings, csv_path: Path):
    engine, factory = open_database(settings)
    try:
        with factory.begin() as session:
            for name in ["alice", "bob"]:
                user = session.scalar(select(User).where(User.username == name))
                if user is None:
                    user = User(
                        username=name,
                        password_hash=generate_password_hash(f"learn-{name}"),
                    )
                    session.add(user)
                    session.flush()
                if name != "alice":
                    continue
                with csv_path.open(encoding="utf-8-sig", newline="") as stream:
                    for row in csv.DictReader(stream):
                        data = BookCreate(
                            title=row["title"],
                            finished=bool(int(row["finished"])),
                            tags=json.loads(row["tags"]),
                        )
                        exists = session.scalar(
                            select(Book.id).where(
                                Book.owner_id == user.id, Book.title == data.title
                            )
                        )
                        if exists is None:
                            session.add(Book(owner_id=user.id, **data.model_dump()))
    finally:
        engine.dispose()

这里复用已学过的 CSV、JSON、Path 和 SQLAlchemy 查询。bool(int(row["finished"])) 把 CSV 中的 "0"、"1" 先变成数字,再变成布尔值;BookCreate 再执行与 API 相同的字段规则。factory.begin() 正常结束统一提交,最后 dispose() 释放本次初始化使用的引擎。

# 续写同一文件:api_project/src/bookapi/cli.py
def main():
    parser = argparse.ArgumentParser(description="初始化或运行书籍 API")
    commands = parser.add_subparsers(dest="command", required=True)
    init = commands.add_parser("init")
    init.add_argument("--csv", type=Path, default=Path("data/books.csv"))
    serve = commands.add_parser("serve")
    serve.add_argument("--host", default="127.0.0.1")
    serve.add_argument("--port", type=int, default=8768)
    args = parser.parse_args()
    if args.command == "init":
        initialize(Settings(), args.csv)
        print("已准备 alice、bob 与采集书目。")
    else:
        # 导入字符串加 factory=True,先调用工厂,再运行 ASGI 应用。
        uvicorn.run(
            "bookapi.app:create_app", factory=True, host=args.host, port=args.port
        )

uvicorn.run("bookapi.app:create_app", factory=True, ...) 从模块找到工厂,调用后运行返回的 ASGI 应用。子命令 init 只初始化,serve 启动监听。完整包还提供 python -m bookapi 入口,调用同一 main。

# 在 api_project 目录初始化账号和六本采集书目。
uv run api-books init
# 启动 API;Ctrl+C 停止,数据库文件继续保留。
uv run api-books serve

默认地址为 http://127.0.0.1:8768/docs,演示账号为 alice / learn-alice 和 bob / learn-bob。也可以先调用注册接口建立自己的账号。

数据库位置来自 BOOKS_DB。例如在 PowerShell 中用另一份数据文件运行:

# 同一个终端中,init 和 serve 都读取这个设置。
$env:BOOKS_DB = "data/another.sqlite3"
uv run api-books init
uv run api-books serve --port 8770
初始化读取六本真实采集书目并按归属及书名查重,正常提交后服务打开同一路径,停止重启后继续读取相同账号和记录

笔记正文与全部题解 · Ctrl / ⌘ K