Spaces:
Paused
Paused
File size: 4,409 Bytes
4223796 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 | """
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) |