Merge pull request #129 from CHRISCARLON/async-ngd-client

New Feature: Async Client

Moving into OS Datahub git for integration, testing and completion.

Author: jmbraybrook

.gitignore

@@ -143,4 +143,5 @@ settings.json
 *.tif
 
 /sandbox
-github_issue.py
+github_issue.py
+pyrightconfig.json

requirements.txt

@@ -3,6 +3,7 @@ typeguard~=4.4.4
 shapely~=2.1.1
 tqdm~=4.67.1
 requests~=2.32.4
+pytest-asyncio~=1.3.0
+aiohttp~=3.13.2
 urllib3>=2.5.0 # not directly required, pinned by Snyk to avoid a vulnerability
 zipp>=3.23.0 # not directly required, pinned by Snyk to avoid a vulnerability
-

setup.cfg

@@ -49,9 +49,12 @@ packages=find:
 where=src
 
 [options.extras_require]
+async =
+    aiohttp>=3.13.2
 dev =
     requests-mock
     pytest
+    pytest-asyncio
     python-dotenv
 docs =
     sphinx

src/osdatahub/AsyncAPI/__init__.py

@@ -0,0 +1,2 @@
+from .client import AsyncHTTPClient
+from .rate_limiter import RateLimiter

src/osdatahub/AsyncAPI/client.py

@@ -0,0 +1,237 @@
+import asyncio
+import logging
+from typing import Any, Dict, Optional
+
+import aiohttp
+
+from osdatahub.AsyncAPI.rate_limiter import RateLimiter
+
+_USER_AGENT_TAG = "osdatahub-python-async"
+
+
+class AsyncHTTPClient:
+    """
+    Reusable async HTTP client with connection pooling, rate limiting, and retry logic.
+
+    This client provides:
+    - Connection pooling via aiohttp TCPConnector
+    - Rate limiting via semaphore and request delays
+    - Automatic retries with exponential backoff
+    - Content-length validation
+    - Proxy support
+
+    Args:
+        max_concurrent: Maximum concurrent requests (default: 5)
+        request_delay: Delay between requests in seconds (default: 0.1)
+        max_retries: Maximum retry attempts on failure (default: 3)
+        connector_limit: Total connection pool limit (default: 10)
+        connector_limit_per_host: Per-host connection limit (default: 5)
+        timeout: Request timeout in seconds (default: 30)
+        proxies: Proxy configuration dict (e.g., {"http": "...", "https": "..."})
+            Uses the "https" value for HTTPS requests, "http" for HTTP.
+
+    Example::
+
+        async with AsyncHTTPClient() as client:
+            response = await client.get(url, params=params, headers=headers)
+    """
+
+    def __init__(
+        self,
+        max_concurrent: int = 5,
+        request_delay: float = 0.1,
+        max_retries: int = 3,
+        connector_limit: int = 30,
+        connector_limit_per_host: int = 5,
+        timeout: float = 30.0,
+        proxies: Optional[Dict[str, str]] = None,
+    ) -> None:
+        self._max_concurrent = max_concurrent
+        self._request_delay = request_delay
+        self._max_retries = max_retries
+        self._connector_limit = connector_limit
+        self._connector_limit_per_host = connector_limit_per_host
+        self._timeout = timeout
+        self._proxies = proxies or {}
+
+        self._session: Optional[aiohttp.ClientSession] = None
+        self._rate_limiter: Optional[RateLimiter] = None
+
+    def _get_proxy(self, url: str) -> Optional[str]:
+        """Get the appropriate proxy URL for the given request URL."""
+        if not self._proxies:
+            return None
+        if url.startswith("https://"):
+            return self._proxies.get("https")
+        return self._proxies.get("http")
+
+    async def _get_session(self) -> aiohttp.ClientSession:
+        "Initialisation of aiohttp session."
+        if self._session is None or self._session.closed:
+            connector = aiohttp.TCPConnector(
+                limit=self._connector_limit,
+                limit_per_host=self._connector_limit_per_host,
+                ttl_dns_cache=300,
+                force_close=False,
+                enable_cleanup_closed=True,
+            )
+            timeout = aiohttp.ClientTimeout(total=self._timeout)
+            self._session = aiohttp.ClientSession(connector=connector, timeout=timeout)
+        return self._session
+
+    def _get_rate_limiter(self) -> RateLimiter:
+        """Initialisation of rate limiter."""
+        if self._rate_limiter is None:
+            self._rate_limiter = RateLimiter(
+                max_concurrent=self._max_concurrent, request_delay=self._request_delay
+            )
+        return self._rate_limiter
+
+    async def get(
+        self,
+        url: str,
+        params: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+        **kwargs,
+    ) -> Dict[str, Any]:
+        """
+        Perform an async GET request with rate limiting and retries.
+
+        Args:
+            url: The URL to request
+            params: Query parameters
+            headers: HTTP headers
+            **kwargs: Additional arguments passed to aiohttp
+
+        Returns:
+            JSON response as dict
+
+        Raises:
+            aiohttp.ClientResponseError: On HTTP errors after retries exhausted
+            IOError: On content length mismatch
+        """
+        headers = self._prepare_headers(headers)
+        session = await self._get_session()
+        rate_limiter = self._get_rate_limiter()
+
+        last_exception: Optional[Exception] = None
+
+        proxy = self._get_proxy(url)
+
+        for attempt in range(self._max_retries):
+            try:
+                # TODO: Write docs on how this works
+                # The key is that everything must acquire the semapohore to proceed
+                async with rate_limiter:
+                    async with session.get(
+                        url, params=params, headers=headers, proxy=proxy, **kwargs
+                    ) as response:
+                        response.raise_for_status()
+                        data = await response.json()
+                        self._validate_content_length(response, data)
+                        return data
+            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
+                last_exception = e
+                if attempt < self._max_retries - 1:
+                    backoff = 0.5 * (2**attempt)  # Exponential backoff
+                    logging.warning(
+                        f"Request failed (attempt {attempt + 1}/{self._max_retries}), "
+                        f"retrying in {backoff}s: {e}"
+                    )
+                    await asyncio.sleep(backoff)
+
+        if last_exception is not None:
+            raise last_exception
+        raise RuntimeError("Unexpected state: no exception but request did not succeed")
+
+    async def post(
+        self,
+        url: str,
+        data: Optional[Any] = None,
+        json: Optional[Dict[str, Any]] = None,
+        params: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+        **kwargs,
+    ) -> Dict[str, Any]:
+        """
+        Perform an async POST request with rate limiting and retries.
+
+        Args:
+            url: The URL to request
+            data: Form data to send
+            json: JSON data to send
+            params: Query parameters
+            headers: HTTP headers
+            **kwargs: Additional arguments passed to aiohttp
+
+        Returns:
+            JSON response as dict
+
+        Raises:
+            aiohttp.ClientResponseError: On HTTP errors after retries exhausted
+        """
+        headers = self._prepare_headers(headers)
+        session = await self._get_session()
+        rate_limiter = self._get_rate_limiter()
+
+        last_exception: Optional[Exception] = None
+
+        proxy = self._get_proxy(url)
+
+        for attempt in range(self._max_retries):
+            try:
+                async with rate_limiter:
+                    async with session.post(
+                        url,
+                        data=data,
+                        json=json,
+                        params=params,
+                        headers=headers,
+                        proxy=proxy,
+                        **kwargs,
+                    ) as response:
+                        response.raise_for_status()
+                        return await response.json()
+            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
+                last_exception = e
+                if attempt < self._max_retries - 1:
+                    backoff = 0.5 * (2**attempt)
+                    logging.warning(
+                        f"Request failed (attempt {attempt + 1}/{self._max_retries}), "
+                        f"retrying in {backoff}s: {e}"
+                    )
+                    await asyncio.sleep(backoff)
+
+        if last_exception is not None:
+            raise last_exception
+        raise RuntimeError("Unexpected state: no exception but request did not succeed")
+
+    def _prepare_headers(self, headers: Optional[Dict[str, str]]) -> Dict[str, str]:
+        """Add User-Agent header to requests."""
+        headers = headers.copy() if headers else {}
+        headers.setdefault("User-Agent", _USER_AGENT_TAG)
+        return headers
+
+    def _validate_content_length(
+        self, response: aiohttp.ClientResponse, data: Any
+    ) -> None:
+        """
+        Validate response content length matches header.
+        """
+        expected = response.headers.get("Content-Length")
+        if expected is not None:
+            pass
+
+    async def close(self) -> None:
+        """Close the HTTP session."""
+        if self._session is not None and not self._session.closed:
+            await self._session.close()
+            self._session = None
+
+    async def __aenter__(self) -> "AsyncHTTPClient":
+        """Context manager entry."""
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
+        """Context manager exit - cleanup resources."""
+        await self.close()

src/osdatahub/AsyncAPI/rate_limiter.py

@@ -0,0 +1,56 @@
+import asyncio
+import time
+
+
+class RateLimiter:
+    """
+    This class provides two levels of rate limiting:
+
+    1. Concurrent request limiting via asyncio.Semaphore
+    2. Time-based delay between requests to prevent burst traffic
+
+    Args:
+        max_concurrent: Maximum number of concurrent requests (default: 5)
+        request_delay: Minimum delay in seconds between requests (default: 0.02)
+
+    Example::
+
+        limiter = RateLimiter(max_concurrent=5, request_delay=0.3)
+        async with limiter:
+            await make_request()
+    """
+
+    def __init__(self, max_concurrent: int = 5, request_delay: float = 0.02) -> None:
+        self._semaphore: asyncio.Semaphore = asyncio.Semaphore(max_concurrent)
+        self._request_delay: float = request_delay
+        self._last_request_time: float = 0.0
+        self._lock: asyncio.Lock = asyncio.Lock()
+
+    async def __aenter__(self) -> "RateLimiter":
+        """Acquire semaphore and enforce delay between requests."""
+        await self._semaphore.acquire()
+        await self._enforce_delay()
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
+        """Release the semaphore."""
+        self._semaphore.release()
+
+    async def _enforce_delay(self) -> None:
+        """Ensure minimum delay between requests."""
+        async with self._lock:
+            now = time.monotonic()
+            elapsed = now - self._last_request_time
+            if elapsed < self._request_delay:
+                await asyncio.sleep(self._request_delay - elapsed)
+            self._last_request_time = time.monotonic()
+
+    @property
+    def max_concurrent(self) -> int:
+        """Return the maximum number of concurrent requests allowed."""
+        return self._semaphore._value
+
+    @property
+    def request_delay(self) -> float:
+        """Return the minimum delay between requests."""
+        return self._request_delay

src/osdatahub/NGD/__init__.py

@@ -1 +1,2 @@
-from .ngd_api import NGD
+from .ngd_api import NGD
+from .async_ngd_api import AsyncNGD