Archived
feat(db-async): 站点数据异步读写仓储 + FastAPI 生命周期
- core/repo.py：经由 core.db(aiosqlite) 读写 auth_system 的**同一份** site.db （settings/announcements/users 只读/ledger 只读），字段口径对齐 auth_system； 读并发、写串行、WAL 读写同时进行 - web/routers/site.py：/api/v1/site/announcements、/api/v1/site/stats（异步多表并发读） - web/app.py：lifespan 关闭时释放 aiohttp 连接池（含按 loop 隔离的 session） 写路径说明：迁移期写操作仍归 auth_system 的 _DB_LOCK 事务口径，避免两套写 路径互相踩踏；本层先只做读。 验证：pytest 22/22（新增 6 项含「异步读 + 同步写者并发 WAL」）； /api/v1/site/* 实测读到共享库（3 用户/1 管理员）；ruff/pyright 干净。
This commit is contained in:
snowy-yang committed
1 parent
ce7ea66d6a
commit
0f2e72db44
4 files changed
+219
No files matched your search
@@ -0,0 +1,61 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""站点数据异步读写仓储（FastAPI 栈用）。
|
||||
|
||||
直接读写 auth_system 建立的同一份 SQLite（同一 DB_PATH / 同一 schema），
|
||||
但走 core.db 的 aiosqlite 异步层——读并发、写串行、WAL 读写同时进行。
|
||||
|
||||
设计约束：本模块**不改动** auth_system 既有 schema，只做读写；字段口径与
|
||||
auth_system 保持一致（键名、默认值）。迁移期两条栈共存、共库。
|
||||
"""
|
||||
from typing import Any
|
||||
|
||||
from core import db
|
||||
|
||||
# ---------- settings ----------
|
||||
|
||||
async def get_setting(key: str, default: str = '') -> str:
|
||||
"""读单个站点设置（对应 auth_system.get_setting）。"""
|
||||
row = await db.fetch_one('SELECT value FROM settings WHERE key = ?', (key,))
|
||||
return row['value'] if row is not None else default
|
||||
|
||||
|
||||
async def set_setting(key: str, value: str) -> None:
|
||||
"""写站点设置（upsert）。"""
|
||||
await db.execute(
|
||||
'INSERT INTO settings (key, value) VALUES (?, ?) '
|
||||
'ON CONFLICT(key) DO UPDATE SET value = excluded.value',
|
||||
(key, value),
|
||||
)
|
||||
|
||||
|
||||
async def all_settings() -> dict[str, str]:
|
||||
rows = await db.fetch_all('SELECT key, value FROM settings')
|
||||
return {r['key']: r['value'] for r in rows}
|
||||
|
||||
|
||||
# ---------- announcements ----------
|
||||
|
||||
async def list_announcements(limit: int = 20) -> list[dict[str, Any]]:
|
||||
rows = await db.fetch_all(
|
||||
'SELECT * FROM announcements ORDER BY rowid DESC LIMIT ?', (limit,))
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
# ---------- users（只读统计；写操作请走 auth_system 的事务口径） ----------
|
||||
|
||||
async def count_users(role: str | None = None) -> int:
|
||||
if role:
|
||||
return int(await db.fetch_val('SELECT COUNT(*) FROM users WHERE role = ?', (role,), 0))
|
||||
return int(await db.fetch_val('SELECT COUNT(*) FROM users', (), 0))
|
||||
|
||||
|
||||
async def get_user_by_name(username: str) -> dict[str, Any] | None:
|
||||
row = await db.fetch_one('SELECT * FROM users WHERE username = ?', (username,))
|
||||
return dict(row) if row is not None else None
|
||||
|
||||
|
||||
# ---------- billing ledger（只读对账） ----------
|
||||
|
||||
async def ledger_open_count() -> int:
|
||||
return int(await db.fetch_val(
|
||||
"SELECT COUNT(*) FROM billing_ledger WHERE status = 'open'", (), 0))
|
||||
@@ -0,0 +1,96 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""异步仓储层测试：读写共享库、并发读、与 auth_system schema 兼容。"""
|
||||
import asyncio
|
||||
import sqlite3
|
||||
|
||||
import pytest
|
||||
|
||||
import core.db as db
|
||||
import core.repo as repo
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def site_db(tmp_path, monkeypatch):
|
||||
"""建一个与 auth_system 口径一致的迷你库，指给 core.db。"""
|
||||
path = tmp_path / 'site.db'
|
||||
conn = sqlite3.connect(str(path))
|
||||
conn.executescript("""
|
||||
CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT UNIQUE, role TEXT DEFAULT 'user');
|
||||
CREATE TABLE settings (key TEXT PRIMARY KEY, value TEXT DEFAULT '');
|
||||
CREATE TABLE announcements (id INTEGER PRIMARY KEY, title TEXT DEFAULT '', content TEXT DEFAULT '',
|
||||
popup_mode TEXT DEFAULT 'off', updated_at TEXT DEFAULT '');
|
||||
CREATE TABLE billing_ledger (task_id TEXT PRIMARY KEY, user_id INTEGER, status TEXT DEFAULT 'open');
|
||||
INSERT INTO users (username, role) VALUES ('alice','user'), ('bob','user'), ('qfbsh','admin');
|
||||
INSERT INTO settings (key, value) VALUES ('server_nickname','云村测试');
|
||||
INSERT INTO announcements (id, title, content) VALUES (1, '公告', '正文');
|
||||
INSERT INTO billing_ledger (task_id, user_id, status) VALUES ('t1', 1, 'open');
|
||||
""")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
monkeypatch.setattr(db, 'DATA_DIR', tmp_path)
|
||||
monkeypatch.setattr(db, 'DB_PATH', path)
|
||||
return path
|
||||
|
||||
|
||||
async def test_get_setting(site_db):
|
||||
assert await repo.get_setting('server_nickname') == '云村测试'
|
||||
assert await repo.get_setting('missing', 'fallback') == 'fallback'
|
||||
|
||||
|
||||
async def test_set_setting_upsert(site_db):
|
||||
await repo.set_setting('a', '1')
|
||||
assert await repo.get_setting('a') == '1'
|
||||
await repo.set_setting('a', '2')
|
||||
assert await repo.get_setting('a') == '2'
|
||||
|
||||
|
||||
async def test_counts(site_db):
|
||||
assert await repo.count_users() == 3
|
||||
assert await repo.count_users('admin') == 1
|
||||
assert await repo.ledger_open_count() == 1
|
||||
|
||||
|
||||
async def test_list_announcements(site_db):
|
||||
rows = await repo.list_announcements()
|
||||
assert rows and rows[0]['title'] == '公告'
|
||||
|
||||
|
||||
async def test_concurrent_reads(site_db):
|
||||
"""并发读同一库不互相阻塞、结果一致。"""
|
||||
results = await asyncio.gather(*[repo.count_users() for _ in range(20)])
|
||||
assert results == [3] * 20
|
||||
|
||||
|
||||
async def test_read_write_concurrent_with_sync_writer(site_db):
|
||||
"""异步读与『另一个同步写者』（模拟遗留栈写库）可同时进行（WAL）。"""
|
||||
stop = asyncio.Event()
|
||||
errors = []
|
||||
|
||||
def sync_writer():
|
||||
# 模拟 auth_system 的同步写：独立连接持续写 settings
|
||||
conn = sqlite3.connect(str(site_db), timeout=15)
|
||||
conn.execute('PRAGMA journal_mode=WAL')
|
||||
try:
|
||||
for i in range(80):
|
||||
conn.execute("INSERT INTO settings (key, value) VALUES (?, 'x') "
|
||||
"ON CONFLICT(key) DO UPDATE SET value='x'", (f'k{i}',))
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
stop.set()
|
||||
|
||||
async def reader():
|
||||
while not stop.is_set():
|
||||
try:
|
||||
await repo.get_setting('server_nickname')
|
||||
except Exception as e:
|
||||
errors.append(e)
|
||||
await asyncio.sleep(0)
|
||||
|
||||
# sync_writer 放线程池，reader 留在事件循环
|
||||
loop = asyncio.get_running_loop()
|
||||
writer_task = loop.run_in_executor(None, sync_writer)
|
||||
await asyncio.gather(writer_task, *[reader() for _ in range(3)])
|
||||
|
||||
assert not errors, f'读期间异常: {errors[:3]}'
|
||||
assert await repo.get_setting('k79') == 'x'
|
||||
+17
@@ -13,12 +13,27 @@
|
||||
当前仅用于开发期联调。
|
||||
"""
|
||||
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
from fastapi import FastAPI
|
||||
|
||||
from core.config import APP_VERSION
|
||||
from web.routers import site as site_router
|
||||
from web.routers import system as system_router
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def _lifespan(app: FastAPI):
|
||||
"""应用生命周期：关闭时释放 aiohttp 连接池（含按事件循环隔离的 session）。"""
|
||||
yield
|
||||
try:
|
||||
from core import aiohttp_compat, http_client
|
||||
await http_client.client.close()
|
||||
await aiohttp_compat.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def create_app(serve_flask_fallback: bool = False) -> FastAPI:
|
||||
"""创建 FastAPI 应用。"""
|
||||
app = FastAPI(
|
||||
@@ -27,10 +42,12 @@ def create_app(serve_flask_fallback: bool = False) -> FastAPI:
|
||||
docs_url='/api/v1/docs',
|
||||
redoc_url=None,
|
||||
openapi_url='/api/v1/openapi.json',
|
||||
lifespan=_lifespan,
|
||||
)
|
||||
|
||||
# 已迁移路由：统一挂 /api/v1（迁移期与 Flask 根路径并存，便于对比）
|
||||
app.include_router(system_router.router, prefix='/api/v1', tags=['system'])
|
||||
app.include_router(site_router.router, prefix='/api/v1', tags=['site'])
|
||||
|
||||
if serve_flask_fallback:
|
||||
_mount_flask_fallback(app)
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""站点数据端点（FastAPI 栈）：走 core.db 异步 SQLite（读并发/写串行/WAL）。
|
||||
|
||||
这些是**新增**的 `/api/v1` 只读端点，用于验证异步读写路径与共享同一份站点库；
|
||||
不替代、不影响 Flask 侧既有端点。鉴权端点（需登录/管理员）待会话互通层就绪后再迁移。
|
||||
|
||||
注意：本模块只做读（与 announcements/settings 这些低风险资源对应），
|
||||
写操作仍归 auth_system 的事务口径，避免迁移期两套写路径互相踩踏。
|
||||
"""
|
||||
from fastapi import APIRouter
|
||||
|
||||
from core import repo
|
||||
from web.response import error, success
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get('/site/announcements')
|
||||
async def list_announcements(limit: int = 20):
|
||||
"""公告列表（异步读；对应 settings/announcements 表）。"""
|
||||
try:
|
||||
items = await repo.list_announcements(max(1, min(limit, 100)))
|
||||
return success(items, '公告获取成功')
|
||||
except Exception:
|
||||
return error('服务暂时不可用，请稍后再试', 500)
|
||||
|
||||
|
||||
@router.get('/site/stats')
|
||||
async def site_stats():
|
||||
"""站点规模概览（异步并发读多张表）。"""
|
||||
try:
|
||||
import asyncio
|
||||
|
||||
total, admins, open_ledger = await asyncio.gather(
|
||||
repo.count_users(),
|
||||
repo.count_users('admin'),
|
||||
repo.ledger_open_count(),
|
||||
)
|
||||
return success({
|
||||
'users': total,
|
||||
'admins': admins,
|
||||
'open_ledger': open_ledger,
|
||||
}, '站点统计获取成功')
|
||||
except Exception:
|
||||
return error('服务暂时不可用，请稍后再试', 500)
|
||||
Reference in new issue
Block a user