(二十一)初始化数据并运行完整项目
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