diff --git a/core/repo.py b/core/repo.py new file mode 100644 index 0000000..1c264e3 --- /dev/null +++ b/core/repo.py @@ -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)) diff --git a/tests/test_repo.py b/tests/test_repo.py new file mode 100644 index 0000000..5d1db3c --- /dev/null +++ b/tests/test_repo.py @@ -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' diff --git a/web/app.py b/web/app.py index 5e62338..4057354 100644 --- a/web/app.py +++ b/web/app.py @@ -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) diff --git a/web/routers/site.py b/web/routers/site.py new file mode 100644 index 0000000..5e6d825 --- /dev/null +++ b/web/routers/site.py @@ -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)