Spaces:
Paused
Paused
| """ | |
| HTTP Client Pool — CrowData | |
| Centralized HTTP client with connection pooling, retries, and timeouts. | |
| """ | |
| import asyncio | |
| import logging | |
| from typing import Optional | |
| import httpx | |
| from app.config import get_settings | |
| logger = logging.getLogger(__name__) | |
| _settings = get_settings() | |
| # Global connection pools | |
| _http_clients: dict[str, httpx.AsyncClient] = {} | |
| _client_locks: dict[str, asyncio.Lock] = {} | |
| class HTTPClientPool: | |
| """Manages pooled HTTP clients per domain.""" | |
| def __init__(self): | |
| self._clients: dict[str, httpx.AsyncClient] = {} | |
| self._locks: dict[str, asyncio.Lock] = {} | |
| def _get_lock(self, key: str) -> asyncio.Lock: | |
| if key not in self._locks: | |
| self._locks[key] = asyncio.Lock() | |
| return self._locks[key] | |
| async def get_client(self, base_url: str = None, timeout: float = 30.0) -> httpx.AsyncClient: | |
| """Get or create a client for the given base_url.""" | |
| key = base_url or "default" | |
| lock = self._get_lock(key) | |
| async with lock: | |
| if key not in self._clients or self._clients[key].is_closed: | |
| client_kwargs = { | |
| "timeout": httpx.Timeout(timeout, connect=10.0), | |
| "limits": httpx.Limits( | |
| max_keepalive_connections=20, | |
| max_connections=100, | |
| keepalive_expiry=30.0, | |
| ), | |
| "headers": { | |
| "User-Agent": "CrowData/1.0 (+https://crowdata.ar)", | |
| "Accept": "application/json, text/html, */*", | |
| "Accept-Language": "es-AR,es;q=0.9,en;q=0.8", | |
| }, | |
| "follow_redirects": True, | |
| } | |
| if base_url: | |
| client_kwargs["base_url"] = base_url | |
| self._clients[key] = httpx.AsyncClient(**client_kwargs) | |
| logger.debug(f"Created new HTTP client for {key or 'default'}") | |
| return self._clients[key] | |
| async def close_all(self): | |
| """Close all clients gracefully.""" | |
| for key, client in self._clients.items(): | |
| try: | |
| await client.aclose() | |
| logger.debug(f"Closed HTTP client for {key}") | |
| except Exception as e: | |
| logger.warning(f"Error closing client {key}: {e}") | |
| self._clients.clear() | |
| async def get(self, url: str, base_url: str = None, **kwargs) -> httpx.Response: | |
| """Convenience method for GET request.""" | |
| client = await self.get_client(base_url) | |
| return await client.get(url, **kwargs) | |
| async def post(self, url: str, base_url: str = None, **kwargs) -> httpx.Response: | |
| """Convenience method for POST request.""" | |
| client = await self.get_client(base_url) | |
| return await client.post(url, **kwargs) | |
| # Global pool instance | |
| _pool = HTTPClientPool() | |
| async def get_http_client(base_url: str = None, timeout: float = 30.0) -> httpx.AsyncClient: | |
| """Get HTTP client from pool.""" | |
| return await _pool.get_client(base_url, timeout) | |
| async def close_http_pool(): | |
| """Close all pooled connections.""" | |
| await _pool.close_all() | |
| # Convenience functions | |
| async def http_get(url: str, *, base_url: str = None, **kwargs) -> httpx.Response: | |
| """GET request using pooled client.""" | |
| if url is None: | |
| logger.error(f"http_get called with None URL! base_url={base_url}, kwargs={kwargs}") | |
| raise ValueError("http_get called with None URL") | |
| if not isinstance(url, str): | |
| logger.error(f"http_get called with non-string URL: {type(url)} = {url!r}") | |
| raise ValueError(f"http_get expects str URL, got {type(url).__name__}") | |
| client = await _pool.get_client(base_url) | |
| return await client.get(url, **kwargs) | |
| async def http_post(url: str, *, base_url: str = None, **kwargs) -> httpx.Response: | |
| """POST request using pooled client.""" | |
| if url is None: | |
| logger.error(f"http_post called with None URL! base_url={base_url}, kwargs={kwargs}") | |
| raise ValueError("http_post called with None URL") | |
| if not isinstance(url, str): | |
| logger.error(f"http_post called with non-string URL: {type(url)} = {url!r}") | |
| raise ValueError(f"http_post expects str URL, got {type(url).__name__}") | |
| client = await _pool.get_client(base_url) | |
| return await client.post(url, **kwargs) |