diff --git a/core/aiohttp_compat.py b/core/aiohttp_compat.py new file mode 100644 index 0000000..b02c796 --- /dev/null +++ b/core/aiohttp_compat.py @@ -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) diff --git a/core/async_bridge.py b/core/async_bridge.py new file mode 100644 index 0000000..f858e63 --- /dev/null +++ b/core/async_bridge.py @@ -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 diff --git a/pyproject.toml b/pyproject.toml index 42b934c..29b7b58 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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"] diff --git a/tests/test_aiohttp_compat.py b/tests/test_aiohttp_compat.py new file mode 100644 index 0000000..4a4a94f --- /dev/null +++ b/tests/test_aiohttp_compat.py @@ -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())