Archived
snowy-yang
9f56f68b66
异步基础层（core/）：
- db.py aiosqlite + WAL：读路径无锁并发、写路径 asyncio.Lock 串行，
busy_timeout 自旋——实现「读写同时进行」（旧实现用 RLock 全读串行）
- http_client.py aiohttp 连接池 + 有界重试（5xx 重试 / 4xx 不重试）+
响应体上限 + 流式接口；按事件循环隔离 session（支持迁移期双 loop）
工具链：
- pyproject.toml ruff(E/F/W/I/B/UP/C4/ASYNC/S/RUF) + pyright(basic) + pytest(asyncio auto)
- requirements-fastapi.txt FastAPI/aiohttp/aiosqlite/aiofiles 等目标栈
- requirements-dev.txt 开发依赖（pytest/ruff/pyright）
清理：删除无用脚本 _csscheck.py 与产物标记 _css_ok.txt/_pkg_ok.txt，同步 .gitignore
验证：pytest 10/10（含「写 200 条同时 4 读协程零异常」并发用例）；
ruff All checks passed；pyright 0 errors。
38 lines
1.3 KiB
Python
38 lines
1.3 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""响应工具：与 Flask 侧 APIResponse 的 JSON 契约逐字段对齐。
|
||
|
||
Flask 的 APIResponse.success/error 返回 (dict, status) 元组，由 Flask 序列化。
|
||
FastAPI 侧改为直接返回 JSONResponse，但**包体结构完全一致**：
|
||
success: {"status": sc, "success": true, "message": msg, ["data": ...]}
|
||
error: {"status": sc, "success": false, "message": msg, ["error_code": ...]}
|
||
这样前端无需任何改动即可同时对接两个框架。
|
||
"""
|
||
|
||
from typing import Any
|
||
|
||
from fastapi.responses import JSONResponse
|
||
|
||
|
||
def success(data: Any = None, message: str = 'success', status_code: int = 200) -> JSONResponse:
|
||
"""成功响应（对应 Flask APIResponse.success）"""
|
||
body = {
|
||
'status': status_code,
|
||
'success': True,
|
||
'message': message,
|
||
}
|
||
if data is not None:
|
||
body['data'] = data
|
||
return JSONResponse(content=body, status_code=status_code)
|
||
|
||
|
||
def error(message: str, status_code: int = 400, error_code: str | None = None) -> JSONResponse:
|
||
"""错误响应（对应 Flask APIResponse.error）"""
|
||
body = {
|
||
'status': status_code,
|
||
'success': False,
|
||
'message': message,
|
||
}
|
||
if error_code:
|
||
body['error_code'] = error_code
|
||
return JSONResponse(content=body, status_code=status_code)
|