Archived
feat(core): requests 兼容异步垫片 + 同步/异步桥
- core/aiohttp_compat.py：以 requests 同形 API（post/get + Response.json/text/ status_code/raise_for_status + HTTPError/RequestException/Timeout）实现异步请求， 使既有 requests 风格代码可「加 await + async def」最小改动迁移；4xx/5xx 不抛、 仅 raise_for_status() 抛（与 requests 语义一致，保住风控判断逻辑） - core/async_bridge.py：run_sync(coro) 经常驻后台事件循环线程执行协程并阻塞等待， 供迁移期同步调用方（遗留 Flask 路由/线程池）复用异步实现 验证：pytest 16/16；ruff All checks passed；pyright 0 errors。
This commit is contained in:
snowy-yang committed
1 parent
9f56f68b66
commit
4962542e78
4 files changed
+301
No files matched your search
@@ -0,0 +1,162 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""requests 兼容的异步 HTTP 垫片（基于 aiohttp）。
|
||||
|
||||
目的：让既有大量 `requests.post/get + response.json()/raise_for_status()` 风格的
|
||||
代码**以最小改动**迁移到异步——把 `import requests` 换成
|
||||
`from core import aiohttp_compat as requests`，加 `await`、加 `async def` 即可，
|
||||
其余 `.text/.json()/.status_code/.raise_for_status()` 语义保持不变。
|
||||
|
||||
与 requests 的语义对齐点（很关键，迁移期不能变）：
|
||||
- 请求本身**不**因 4xx/5xx 抛异常；只有显式 `raise_for_status()` 才抛 `HTTPError`
|
||||
（上游风控 429/460 的重试/不重试判断依赖调用方读取 `status_code`，见 music_api.HTTPClient）。
|
||||
- 网络异常/超时抛 `RequestException`（含 `Timeout`）。
|
||||
- `Response` 一次性读全 body（与 requests 一致），暴露 text/json/content/headers/cookies/url。
|
||||
|
||||
连接按事件循环隔离（见 core.http_client 的同样设计）。
|
||||
"""
|
||||
import asyncio
|
||||
import json as _json
|
||||
from collections.abc import Mapping
|
||||
from typing import Any
|
||||
|
||||
import aiohttp
|
||||
|
||||
DEFAULT_TIMEOUT = 30
|
||||
DEFAULT_MAX_BYTES = 8 * 1024 * 1024
|
||||
|
||||
# ---------- 异常层次（对齐 requests 命名，便于原样 except） ----------
|
||||
|
||||
|
||||
class RequestException(Exception):
|
||||
"""请求异常基类（网络/协议层）。"""
|
||||
|
||||
|
||||
class HTTPError(RequestException):
|
||||
"""raise_for_status() 抛出的 HTTP 状态异常。"""
|
||||
|
||||
def __init__(self, message: str, response: 'Response | None' = None):
|
||||
super().__init__(message)
|
||||
self.response = response
|
||||
|
||||
|
||||
class Timeout(RequestException):
|
||||
"""请求超时。"""
|
||||
|
||||
|
||||
class ConnectionError(RequestException):
|
||||
"""连接错误。"""
|
||||
|
||||
|
||||
class Response:
|
||||
"""requests.Response 的异步子集：body 已读全，属性同步访问。"""
|
||||
|
||||
def __init__(self, status_code: int, url: str, content: bytes,
|
||||
headers: Mapping[str, str], cookies: Mapping[str, str], reason: str = ''):
|
||||
self.status_code = status_code
|
||||
self.url = url
|
||||
self.content = content
|
||||
self.headers = dict(headers)
|
||||
self.cookies = dict(cookies)
|
||||
self.reason = reason
|
||||
|
||||
@property
|
||||
def text(self) -> str:
|
||||
return self.content.decode('utf-8', errors='replace')
|
||||
|
||||
def json(self, **kw) -> Any:
|
||||
return _json.loads(self.text, **kw)
|
||||
|
||||
def raise_for_status(self) -> None:
|
||||
if 400 <= self.status_code:
|
||||
raise HTTPError(f"{self.status_code} {self.reason}".strip(), self)
|
||||
|
||||
|
||||
# ---------- 会话（按 loop 隔离的 aiohttp session） ----------
|
||||
|
||||
|
||||
class _SessionState:
|
||||
def __init__(self):
|
||||
self._sessions: dict[int, aiohttp.ClientSession] = {}
|
||||
self._lock: asyncio.Lock | None = None
|
||||
|
||||
def _get_lock(self) -> asyncio.Lock:
|
||||
if self._lock is None:
|
||||
self._lock = asyncio.Lock()
|
||||
return self._lock
|
||||
|
||||
async def session(self) -> aiohttp.ClientSession:
|
||||
key = id(asyncio.get_running_loop())
|
||||
sess = self._sessions.get(key)
|
||||
if sess is not None and not sess.closed:
|
||||
return sess
|
||||
async with self._get_lock():
|
||||
sess = self._sessions.get(key)
|
||||
if sess is None or sess.closed:
|
||||
sess = aiohttp.ClientSession(
|
||||
connector=aiohttp.TCPConnector(limit=100, limit_per_host=32, ttl_dns_cache=300),
|
||||
)
|
||||
self._sessions[key] = sess
|
||||
return sess
|
||||
|
||||
async def close_all(self) -> None:
|
||||
for sess in self._sessions.values():
|
||||
if not sess.closed:
|
||||
await sess.close()
|
||||
self._sessions.clear()
|
||||
|
||||
|
||||
_state = _SessionState()
|
||||
|
||||
|
||||
async def close() -> None:
|
||||
"""关闭所有连接（应用关闭时调用）。"""
|
||||
await _state.close_all()
|
||||
|
||||
|
||||
async def request(
|
||||
method: str,
|
||||
url: str,
|
||||
*,
|
||||
params: Any = None,
|
||||
data: Any = None,
|
||||
json: Any = None,
|
||||
headers: Mapping[str, str] | None = None,
|
||||
cookies: Mapping[str, str] | None = None,
|
||||
timeout: float | None = DEFAULT_TIMEOUT,
|
||||
allow_redirects: bool = True,
|
||||
max_bytes: int = DEFAULT_MAX_BYTES,
|
||||
) -> Response:
|
||||
"""发起单次异步请求，返回 Response（不因 4xx/5xx 抛异常，同 requests）。"""
|
||||
sess = await _state.session()
|
||||
req_timeout = aiohttp.ClientTimeout(total=timeout) if timeout else None
|
||||
try:
|
||||
async with sess.request(
|
||||
method, url,
|
||||
params=params, data=data, json=json,
|
||||
headers=dict(headers) if headers else None,
|
||||
cookies=dict(cookies) if cookies else None,
|
||||
timeout=req_timeout, allow_redirects=allow_redirects,
|
||||
) as resp:
|
||||
body = await resp.read()
|
||||
if len(body) > max_bytes:
|
||||
raise RequestException(f"响应体超过上限 {max_bytes} 字节")
|
||||
return Response(
|
||||
status_code=resp.status,
|
||||
url=str(resp.url),
|
||||
content=body,
|
||||
headers=dict(resp.headers),
|
||||
cookies={k: v.value for k, v in resp.cookies.items()},
|
||||
reason=resp.reason or '',
|
||||
)
|
||||
except aiohttp.ClientError as e:
|
||||
raise ConnectionError(str(e)) from e
|
||||
except TimeoutError as e:
|
||||
raise Timeout('请求超时') from e
|
||||
|
||||
|
||||
async def get(url: str, **kw) -> Response:
|
||||
return await request('GET', url, **kw)
|
||||
|
||||
|
||||
async def post(url: str, **kw) -> Response:
|
||||
return await request('POST', url, **kw)
|
||||
@@ -0,0 +1,56 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""同步 ↔ 异步桥（迁移期用）。
|
||||
|
||||
场景：部分调用方仍是同步代码（遗留 Flask 路由、线程池下载器），而服务层已异步。
|
||||
`run_sync(coro)` 把协程提交到**专用后台事件循环线程**执行并阻塞等待结果，
|
||||
使同步代码得以调用异步实现，无需把调用链一次性全部改异步。
|
||||
|
||||
- 后台 loop 常驻单例（daemon 线程），避免每个请求新建 loop 的开销与
|
||||
aiohttp 连接池失效；
|
||||
- 在已有事件循环中调用 `run_sync` 会死锁，故检测到运行中的 loop 时直接报错，
|
||||
提示改用 `await`。
|
||||
"""
|
||||
import asyncio
|
||||
import threading
|
||||
from collections.abc import Coroutine
|
||||
from typing import Any
|
||||
|
||||
_loop: asyncio.AbstractEventLoop | None = None
|
||||
_loop_lock = threading.Lock()
|
||||
|
||||
|
||||
def _get_loop() -> asyncio.AbstractEventLoop:
|
||||
global _loop
|
||||
if _loop is not None and not _loop.is_closed():
|
||||
return _loop
|
||||
with _loop_lock:
|
||||
if _loop is None or _loop.is_closed():
|
||||
loop = asyncio.new_event_loop()
|
||||
t = threading.Thread(target=loop.run_forever, name='async-bridge', daemon=True)
|
||||
t.start()
|
||||
_loop = loop
|
||||
return _loop
|
||||
|
||||
|
||||
def run_sync[T](coro: Coroutine[Any, Any, T], timeout: float | None = None) -> T:
|
||||
"""在同步代码中运行协程并等待结果。
|
||||
|
||||
Raises:
|
||||
RuntimeError: 在已有事件循环内调用（应直接 await 而非桥接）。
|
||||
"""
|
||||
try:
|
||||
asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
pass
|
||||
else:
|
||||
raise RuntimeError('run_sync 不能在事件循环内调用，请直接 await 协程')
|
||||
fut = asyncio.run_coroutine_threadsafe(coro, _get_loop())
|
||||
return fut.result(timeout)
|
||||
|
||||
|
||||
def shutdown() -> None:
|
||||
"""停止后台 loop（应用退出时调用）。"""
|
||||
global _loop
|
||||
if _loop is not None and not _loop.is_closed():
|
||||
_loop.call_soon_threadsafe(_loop.stop)
|
||||
_loop = None
|
||||
@@ -34,6 +34,7 @@ ignore = [
|
||||
# ASYNC109：timeout 参数是有意 API——直接转发给 aiohttp 的 ClientTimeout，
|
||||
# 不是「用 sleep 当超时」的反模式。
|
||||
"core/http_client.py" = ["ASYNC109"]
|
||||
"core/aiohttp_compat.py" = ["ASYNC109"]
|
||||
|
||||
[tool.ruff.lint.isort]
|
||||
known-first-party = ["core", "web", "music_api", "auth_system"]
|
||||
|
||||
@@ -0,0 +1,82 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""兼容垫片与同步桥测试。"""
|
||||
import asyncio
|
||||
|
||||
import pytest
|
||||
from aiohttp import web
|
||||
|
||||
from core import aiohttp_compat as req
|
||||
from core import async_bridge
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
async def server():
|
||||
app = web.Application()
|
||||
|
||||
async def ok(request):
|
||||
return web.json_response({'code': 200, 'data': 'hi'})
|
||||
|
||||
async def notfound(request):
|
||||
return web.json_response({'code': 404}, status=404)
|
||||
|
||||
async def echo_post(request):
|
||||
body = await request.text()
|
||||
return web.json_response({'got': body, 'ua': request.headers.get('User-Agent')})
|
||||
|
||||
app.router.add_get('/ok', ok)
|
||||
app.router.add_get('/nf', notfound)
|
||||
app.router.add_post('/echo', echo_post)
|
||||
|
||||
runner = web.AppRunner(app)
|
||||
await runner.setup()
|
||||
site = web.TCPSite(runner, '127.0.0.1', 0)
|
||||
await site.start()
|
||||
port = site._server.sockets[0].getsockname()[1]
|
||||
try:
|
||||
yield f'http://127.0.0.1:{port}'
|
||||
finally:
|
||||
await runner.cleanup()
|
||||
|
||||
|
||||
async def test_get_json(server):
|
||||
r = await req.get(server + '/ok')
|
||||
assert r.status_code == 200
|
||||
assert r.json() == {'code': 200, 'data': 'hi'}
|
||||
assert 'code' in r.text
|
||||
r.raise_for_status() # 200 不抛
|
||||
|
||||
|
||||
async def test_4xx_does_not_raise_until_raise_for_status(server):
|
||||
r = await req.get(server + '/nf')
|
||||
assert r.status_code == 404 # 请求本身不抛（同 requests）
|
||||
assert r.json() == {'code': 404}
|
||||
with pytest.raises(req.HTTPError):
|
||||
r.raise_for_status()
|
||||
|
||||
|
||||
async def test_post_headers_forwarded(server):
|
||||
r = await req.post(server + '/echo', data={'params': 'x'}, headers={'User-Agent': 'ua-test'})
|
||||
body = r.json()
|
||||
assert body['ua'] == 'ua-test'
|
||||
assert 'params' in body['got']
|
||||
|
||||
|
||||
async def test_network_error_raises_request_exception():
|
||||
with pytest.raises(req.RequestException):
|
||||
await req.get('http://127.0.0.1:1/nope', timeout=1)
|
||||
|
||||
|
||||
def test_bridge_runs_coro_from_sync():
|
||||
async def coro():
|
||||
await asyncio.sleep(0.01)
|
||||
return 42
|
||||
|
||||
assert async_bridge.run_sync(coro()) == 42
|
||||
|
||||
|
||||
async def test_bridge_rejects_inside_loop():
|
||||
async def coro():
|
||||
return 1
|
||||
|
||||
with pytest.raises(RuntimeError):
|
||||
async_bridge.run_sync(coro())
|
||||
Reference in new issue
Block a user