This repository has been archived on 2026-10-02. You can view files and clone it. You cannot open issues or pull requests or push a commit.
Files
snowy-yang 96af56cc27 refactor: auth_system 全面异步化(aiosqlite),拆掉 web/services 层的 to_thread 桥接
拆掉两类垫片的最后一处:auth_system 原为同步 sqlite3 栈,web 层与 services 的
worker 只能靠 asyncio.to_thread 逐个桥接才不阻塞事件循环——每次 DB 访问都白付一次
线程池调度,且「同步调异步」的边界散落在 100+ 个调用点。B1 已把 worker 变成协程,
本批把访问层本身也迁到原生异步,两条线在此合拢。

改动(按子系统):
- auth_system:db.py 照 core/db.py 现有范式重建为原生异步访问层(aiosqlite + WAL)
  ——读路径 _q 每次独立连接、无锁、可并发;写路径 _x / write_transaction 用
  asyncio.Lock 串行化,写者不阻塞读者。users / cards / credits / ledger / logs /
  settings / mail / geo / backup 九模块随之全部 async/await;security.py 是纯计算
  (pbkdf2/normalize)不动。__init__ 去掉 _DB_LOCK 与 get_conn 导出。
- web:app / deps / billing / tasks_registry 与 routers/{admin_site,auth,music,
  register,user} 直接 await,拆掉包裹的 to_thread。deps.secret_key 改 async,
  连带 _parse_cookie 也 async(会话解析要读密钥)。
- services:_taskcore / tasks / downloads / podcast 的账本调用与查询直接 await。
  4 处 start_task 调用点(download.py)确认已 await。

设计取舍(为什么这么做,而不是直接包一层):
- 锁按**事件循环**隔离(_LOOP_LOCKS 以 id(loop) 为键):asyncio.Lock 绑定创建它的
  loop,应用主 loop 与测试各自的 loop 并存,跨 loop 复用会抛 "bound to a different
  event loop"。
- WAL 切换用**独立锁** _wal_lock 而非 _write_lock:execute() 已持写锁,再在锁内
  connect→_configure 会自锁死(asyncio.Lock 非重入)。切换 + 标记在专用锁内串行并
  二次复查,否则并发首连会同时执行 PRAGMA journal_mode=WAL 争排他锁报
  database is locked(core/db.py 同款教训,实测复现)。
- force_wal 保留:journal_mode 持久化在库文件头,库文件被删重建时新文件不继承,
  必须重设。exists() 判定须在 aiosqlite.connect **之前**(connect 会创建空文件)。
- connect() 用 DB_PATH.parent 兜底建目录,优于 DATA_DIR:生产两者相同,但测试
  monkeypatch 只换 DB_PATH 时兜底仍能命中正确目录。

留在 to_thread 上的仅剩**正当的阻塞卸载**,不是同步门面:PBKDF2 的 10 万次迭代
(登录链路每次都要跑)、smtplib 发信、备份文件的 copy2 / os.replace、secret.key
的文件 IO。原有 aria2 / downloader / zip 侧的 to_thread 未动。

不变量逐条保持(迁移只改执行模型,不改语义):
- ledger_settle 幂等:账本非 open 直接 return 0,重启后重放不重复退款;
- 终态先到先得:_mark_task_terminal 对已 done/error/canceled 的任务直接 return,
  堵住 canceled→error 的非法迁移边;
- settle 必须在任务锁**外**调用(ledger_settle 要抢 SQLite 写锁,busy 最长 15s,
  持任务锁会让全站进度/提交/取消随库停摆);
- reserve_and_open 的预扣与开账同事务,消除原两步之间的崩溃丢账窗口。

路径真源红线:DATA_DIR / DB_PATH / BACKUP_DIR / SECRET_KEY_PATH 仍是 auth_system.db
的**模块全局**(用与 core.db 相同的表达式,但**不是**它的别名)。测试只 patch
auth_system.db.*,若改成 core.db 的别名会静默把测试写到生产库。已核 data/site.db
读写前后 sha256 完全一致。
顺带清掉 get_conn:迁移后它只剩「转调 connect()」一层空壳、全仓零消费者(测试
直接开裸连接、备份用 db.connect()),保留等于留一条「拿裸连接自己写」的口子。

测试:19 个文件同步迁移——10 个文件的 web.deps.secret_key 替身改 async;
test_db_conn.py 的多线程首连用例按异步语义拆成 2 条(单 loop gather(16) 覆盖批量
并发首连、WAL 就绪后 8 线程各自 loop 建连覆盖跨 loop);test_billing_scope.py 去掉
多余的 @pytest.mark.anyio;被 await 且**正常返回**的 monkeypatch 替身一律改 async
(仅 raise 的同步替身仍可用)。

门禁:ruff 全过 / pyright 0-0-0(standard 模式含 tests 与顶层模块)/ pytest
384 项全绿(0 失败 0 error 0 skipped;383→384 因 test_db_conn 拆用例)。
data/site.db 读写前后 sha256 一致(a19f5db…6487),无 -wal/-shm 边车残留。

下一步 B3:删 core/sync_http.py、core/async_bridge.py、core/http_errors.py 及
 tests/test_async_bridge.py——三者在生产侧已零消费者(http_errors 仅 sync_http
 自用,requests 已非声明依赖)。
2026-09-28 20:13:19 +08:00

145 lines
6.4 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
"""FastAPI 应用工厂。
本应用即唯一服务端:全部端点在 web/routers/ 下按业务域注册。
`api_prefix` 用于把整套路由挂到指定前缀下(便于契约测试与反向代理)。
"""
import asyncio
from contextlib import asynccontextmanager
from pathlib import Path
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
import auth_system
from core.config import APP_VERSION
from core.security import init_secret_key
from web.errors import install_error_handlers
from web.middleware import ResponseHygieneMiddleware
from web.routers import account as account_router
from web.routers import admin as admin_router
from web.routers import admin_site as admin_site_router
from web.routers import auth as auth_router
from web.routers import auth_status as auth_status_router
from web.routers import delivery as delivery_router
from web.routers import download as download_router
from web.routers import index as index_router
from web.routers import music as music_router
from web.routers import podcast as podcast_router
from web.routers import register as register_router
from web.routers import site as site_router
from web.routers import stream as stream_router
from web.routers import system as system_router
from web.routers import tasks as tasks_router
from web.routers import user as user_router
def _make_lifespan(start_monitor: bool):
"""构造 lifespan(是否启动任务监控线程由 create_app 参数决定)。"""
@asynccontextmanager
async def _lifespan(app: FastAPI):
"""应用生命周期:建表 → 启动任务监控 → 关闭时释放连接池。
启动必须建表。全新部署时宿主机的 data/ 往往是个空目录(compose 用
`./data:/app/data` 挂载,会盖掉镜像内的任何内容),此时若只等首次查询
让 sqlite3 自动建库,得到的是一张 **0 表空库**——随后所有涉库端点报
`no such table: users`(实测登录端点直接 500,且错误被吞成"请稍后重试",
极难定位)。init_db 全程 `CREATE TABLE IF NOT EXISTS` + 幂等迁移,
对已有库重复执行无副作用,故放启动期最稳妥。
任务监控线程也必须在此启动:Flask 单体时代由模块级 `threading.Thread(
target=_task_monitor_loop).start()` 拉起(基线 main.py:4594),移除 Flask
迁到 FastAPI 时该启动点丢失,此后监控循环从未运行——心跳超时不再暂停
离线设备、超时任务不再取消、过期任务与下载缓存不再回收、账本孤儿对账
不再执行(预扣未退款的账永久悬挂)。详见 services.tasks.start_task_monitor。
"""
await auth_system.init_db()
if start_monitor:
from services.tasks import start_task_monitor
await asyncio.to_thread(start_task_monitor)
yield
try:
from core import http_client
await http_client.client.close()
except Exception:
pass
return _lifespan
def _init_audio_secret() -> None:
"""注入音频签名密钥(与 auth_system 同源,令直链 token 与服务端一致)。
create_app 是**同步**的应用工厂入口(可能在无事件循环的模块导入期调用),
而 get_secret_key 已迁移为原生异步——此处用 asyncio.run 起一个临时 loop 完成
一次性的密钥读取(启动期一次性边界,非运行时同步门面)。若已处于事件循环中
无法 run,则跳过注入(由调用方另行初始化)。
"""
try:
import auth_system
init_secret_key(asyncio.run(auth_system.get_secret_key()))
except Exception:
pass
def create_app(api_prefix: str = '', *, start_monitor: bool = True) -> FastAPI:
"""创建 FastAPI 应用。
api_prefix:路由挂载前缀。
- ''(默认):直接服务前端使用的根路径(生产形态)
- '/api/v1':挂在子前缀下(便于契约测试与反向代理)
start_monitor:是否随应用启动任务监控线程。生产恒为 True;
测试若以 `with TestClient(app)` 方式进入 lifespan,可传 False
以避免后台线程清扫用例的临时目录。
"""
_init_audio_secret()
app = FastAPI(
title='云村小助手API服务',
version=APP_VERSION,
docs_url='/api/v1/docs',
redoc_url=None,
openapi_url='/api/v1/openapi.json',
lifespan=_make_lifespan(start_monitor),
)
# 响应侧统一处理:安全头 + 音频直链签名保护 + 版本指纹抹除
app.add_middleware(ResponseHygieneMiddleware)
# React 构建产物(vite build → frontend/dist,hash 指纹资源挂 /assets)
_dist_dir = Path(__file__).resolve().parent.parent / 'frontend' / 'dist'
if (_dist_dir / 'assets').is_dir():
app.mount(
'/assets',
StaticFiles(directory=str(_dist_dir / 'assets')),
name='assets',
)
# 业务路由:统一挂到 api_prefix(默认空前缀,即前端使用的根路径)
app.include_router(system_router.router, prefix=api_prefix, tags=['system'])
app.include_router(site_router.router, prefix=api_prefix, tags=['site'])
app.include_router(music_router.router, prefix=api_prefix, tags=['music'])
app.include_router(account_router.router, prefix=api_prefix, tags=['account'])
app.include_router(podcast_router.router, prefix=api_prefix, tags=['podcast'])
app.include_router(tasks_router.router, prefix=api_prefix, tags=['tasks'])
app.include_router(delivery_router.router, prefix=api_prefix, tags=['delivery'])
app.include_router(stream_router.router, prefix=api_prefix, tags=['stream'])
app.include_router(auth_status_router.router, prefix=api_prefix, tags=['auth'])
app.include_router(auth_router.router, prefix=api_prefix, tags=['auth'])
app.include_router(user_router.router, prefix=api_prefix, tags=['user'])
app.include_router(admin_router.router, prefix=api_prefix, tags=['admin'])
app.include_router(register_router.router, prefix=api_prefix, tags=['auth'])
app.include_router(admin_site_router.router, prefix=api_prefix, tags=['admin'])
app.include_router(download_router.router, prefix=api_prefix, tags=['download'])
# 首页(根路径,不挂前缀)
app.include_router(index_router.router, tags=['index'])
install_error_handlers(app)
return app