| """ | |
| 全局 httpx 客户端 - 连接池复用 | |
| """ | |
| import httpx | |
| from typing import Optional | |
| DEFAULT_TIMEOUT = 60.0 | |
| MAX_CONNECTIONS = 10 | |
| MAX_KEEPALIVE_CONNECTIONS = 5 | |
| KEEPALIVE_EXPIRY = 30.0 | |
| _http_client: Optional[httpx.AsyncClient] = None | |
| def get_http_client() -> httpx.AsyncClient: | |
| """获取全局 httpx 客户端单例""" | |
| global _http_client | |
| if _http_client is None: | |
| limits = httpx.Limits( | |
| max_connections=MAX_CONNECTIONS, | |
| max_keepalive_connections=MAX_KEEPALIVE_CONNECTIONS, | |
| keepalive_expiry=KEEPALIVE_EXPIRY, | |
| ) | |
| _http_client = httpx.AsyncClient( | |
| timeout=DEFAULT_TIMEOUT, | |
| limits=limits, | |
| follow_redirects=True, | |
| ) | |
| return _http_client | |
| async def close_http_client(): | |
| """关闭全局客户端（应用关闭时调用）""" | |
| global _http_client | |
| if _http_client is not None: | |
| await _http_client.aclose() | |
| _http_client = None |