99 lines
2.8 KiB
Python
99 lines
2.8 KiB
Python
"""基础 HTTP 客户端:token 注入 + 401 自动重登。
|
|
|
|
两个外部系统(RIIN / SDS)共享这套行为,仅以下差异由子类配置:
|
|
- base_url
|
|
- token 写入的请求头名(RIIN: Authorization,SDS: access-token)
|
|
- 固定请求头(RIIN 带 X-Time-Zone / X-Requested-With,SDS 无)
|
|
- 登录实现(子类实现 _login)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
|
|
class TokenHttpClient(ABC):
|
|
"""带 token 鉴权的 HTTP 客户端基类(组合式,非强制耦合)。"""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
*,
|
|
token_header: str,
|
|
fixed_headers: dict[str, str] | None = None,
|
|
login_markers: tuple[str, ...] = (),
|
|
timeout: float = 30.0,
|
|
) -> None:
|
|
self._token: str | None = None
|
|
self._token_header = token_header
|
|
self._login_markers = login_markers
|
|
self._http = httpx.Client(
|
|
base_url=base_url,
|
|
headers=fixed_headers or {},
|
|
timeout=timeout,
|
|
)
|
|
|
|
@abstractmethod
|
|
def _login(self) -> str:
|
|
"""执行登录并返回新 token。子类实现具体认证逻辑。"""
|
|
|
|
# ----- token 生命周期 -----
|
|
|
|
def _ensure_token(self) -> str:
|
|
if not self._token:
|
|
self._token = self._login()
|
|
return self._token
|
|
|
|
def _clear_token(self) -> None:
|
|
self._token = None
|
|
|
|
def _is_login_call(self, url: str) -> bool:
|
|
return any(marker in url for marker in self._login_markers)
|
|
|
|
# ----- 请求 -----
|
|
|
|
def _send(
|
|
self,
|
|
method: str,
|
|
url: str,
|
|
*,
|
|
params: dict[str, Any] | None = None,
|
|
body: dict[str, Any] | None = None,
|
|
_retry: bool = True,
|
|
) -> Any:
|
|
headers: dict[str, str] = {}
|
|
if not self._is_login_call(url):
|
|
headers[self._token_header] = self._ensure_token()
|
|
|
|
response = self._http.request(method, url, params=params, json=body, headers=headers)
|
|
|
|
if response.status_code == 401 and _retry and not self._is_login_call(url):
|
|
self._clear_token()
|
|
return self._send(method, url, params=params, body=body, _retry=False)
|
|
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def get(self, url: str, params: dict[str, Any] | None = None) -> Any:
|
|
return self._send("GET", url, params=params)
|
|
|
|
def post(
|
|
self,
|
|
url: str,
|
|
body: dict[str, Any] | None = None,
|
|
params: dict[str, Any] | None = None,
|
|
) -> Any:
|
|
return self._send("POST", url, body=body, params=params)
|
|
|
|
def close(self) -> None:
|
|
self._http.close()
|
|
|
|
def __enter__(self) -> "TokenHttpClient":
|
|
return self
|
|
|
|
def __exit__(self, *exc: Any) -> None:
|
|
self.close()
|