#154 collapse duplicated sync/async response-shaping + align retry policy

datamaxi/_retry.py

@@ -0,0 +1,86 @@
+"""Shared transient-5xx retry policy for the sync and async transports.
+
+The sync client (``datamaxi.api.API``) mounts a ``urllib3.util.retry.Retry``
+on its ``requests.Session`` adapters, which already implements this policy
+for `requests`. The async client (``datamaxi.aio._core.AsyncAPI``) is built
+on ``httpx``, which has no equivalent adapter-level retry, so it hand-rolls
+its own loop. This module is the single description of that policy so the
+two loops can't drift onto different retry behavior:
+
+* GET-only — retries are only safe for idempotent requests.
+* Exponential backoff — mirrors ``Retry.get_backoff_time()``: no delay
+  before the first retry, then ``backoff_factor * 2 ** (n - 1)`` before the
+  n-th retry (n >= 2), capped at ``BACKOFF_MAX`` seconds.
+* Honors a ``Retry-After`` response header (seconds or HTTP-date form) when
+  present, taking priority over the computed backoff — mirrors
+  ``Retry.respect_retry_after_header=True``.
+"""
+
+import email.utils
+import re
+import time
+
+#: Matches urllib3's ``Retry.DEFAULT_BACKOFF_MAX``.
+BACKOFF_MAX = 120.0
+
+_RETRY_AFTER_SECONDS_RE = re.compile(r"^\s*[0-9]+\s*$")
+
+
+def is_retryable(method, status_code, attempt, max_retries, retry_statuses):
+    """Whether attempt number ``attempt`` (1 = first failure) may be retried.
+
+    Only idempotent GETs are retried, only for statuses in
+    ``retry_statuses``, and only while ``attempt <= max_retries``.
+    """
+    return (
+        str(method).upper() == "GET"
+        and status_code in retry_statuses
+        and attempt <= max_retries
+    )
+
+
+def parse_retry_after(value):
+    """Parse a ``Retry-After`` header value into seconds, or ``None``.
+
+    Accepts either the numeric-seconds form or an HTTP-date, matching
+    ``urllib3.util.retry.Retry.parse_retry_after``. Returns ``None`` if
+    ``value`` is ``None`` or not parseable; never returns a negative number.
+    """
+    if value is None:
+        return None
+
+    if _RETRY_AFTER_SECONDS_RE.match(value):
+        seconds = float(value)
+    else:
+        retry_date_tuple = email.utils.parsedate_tz(value)
+        if retry_date_tuple is None:
+            return None
+        if retry_date_tuple[9] is None:
+            retry_date_tuple = retry_date_tuple[:9] + (0,) + retry_date_tuple[10:]
+        retry_date = email.utils.mktime_tz(retry_date_tuple)
+        seconds = retry_date - time.time()
+
+    return max(0.0, seconds)
+
+
+def get_backoff_time(attempt, backoff_factor):
+    """Seconds to sleep before retry number ``attempt`` (1 = first retry).
+
+    Matches urllib3's ``Retry.get_backoff_time()``: zero delay before the
+    first retry, then exponential growth, capped at ``BACKOFF_MAX``.
+    """
+    if attempt <= 1:
+        return 0.0
+    return min(BACKOFF_MAX, backoff_factor * (2 ** (attempt - 1)))
+
+
+def get_retry_delay(attempt, backoff_factor, headers):
+    """Seconds to sleep before retry number ``attempt``, honoring ``Retry-After``.
+
+    ``headers`` is any mapping-like object exposing ``.get(...)``
+    (``requests``/``httpx`` headers are both case-insensitive mappings).
+    """
+    retry_after = parse_retry_after(headers.get("Retry-After"))
+    if retry_after is not None:
+        return retry_after
+    return get_backoff_time(attempt, backoff_factor)

datamaxi/aio/_core.py

@@ -2,7 +2,10 @@
 
 Reuses the sync client's endpoint resolution and error handling
 (``datamaxi._dispatch``) plus ``ResponseMeta``, so the sync and async clients
-can't drift on request building or error semantics.
+can't drift on request building or error semantics. The retry loop below
+also follows the shared policy described in ``datamaxi._retry`` (GET-only,
+exponential backoff, honors ``Retry-After``) so it can't drift from the
+``urllib3``-backed retry mounted on the sync ``requests.Session``.
 """
 
 import asyncio
@@ -11,6 +14,7 @@
 from datamaxi.__version__ import __version__
 from datamaxi.api import ResponseMeta
 from datamaxi._dispatch import resolve_endpoint, raise_for_error, extract_limit_usage
+from datamaxi._retry import is_retryable, get_retry_delay
 
 
 def _import_httpx():
@@ -28,8 +32,9 @@ class AsyncAPI:
     """Async transport built on ``httpx.AsyncClient``.
 
     Mirrors the sync ``API``: shared endpoint resolution, bounded retry of
-    transient gateway 5xx, the same ``ClientError`` / ``ServerError`` contract,
-    and ``last_response`` metadata.
+    transient gateway 5xx on GET requests with exponential backoff (honoring
+    ``Retry-After`` — see ``datamaxi._retry``), the same ``ClientError`` /
+    ``ServerError`` contract, and ``last_response`` metadata.
     """
 
     def __init__(
@@ -69,15 +74,20 @@ async def send_request(self, method, url_path, payload=None):
         # str()-encode scalars so bools match the sync client's urlencode
         # output (e.g. include_source -> "True", not httpx's "true").
         params = {k: str(v) for k, v in (payload or {}).items() if v is not None}
-        for attempt in range(self.max_retries + 1):
+        attempt = 0
+        while True:
             response = await self._client.request(method, url_path, params=params)
-            if (
-                response.status_code in self.retry_statuses
-                and attempt < self.max_retries
+            attempt += 1
+            if not is_retryable(
+                method,
+                response.status_code,
+                attempt,
+                self.max_retries,
+                self.retry_statuses,
             ):
-                await asyncio.sleep(self.retry_backoff * (attempt + 1))
-                continue
-            break
+                break
+            delay = get_retry_delay(attempt, self.retry_backoff, response.headers)
+            await asyncio.sleep(delay)
 
         raise_for_error(response.status_code, response.text, response.headers)
 

datamaxi/aio/cex.py

@@ -6,6 +6,7 @@
 
 from datamaxi.aio._core import AsyncAPI, AsyncResource
 from datamaxi.lib.utils import check_required_parameter, check_required_parameters
+from datamaxi.resources.utils import raise_if_no_data, to_indexed_dataframe
 from datamaxi.lib.constants import (
     SPOT,
     FUTURES,
@@ -63,8 +64,7 @@ async def __call__(
             currency=currency,
             **{"from": from_unix, "to": to_unix},
         )
-        if res["data"] is None or len(res["data"]) == 0:
-            raise ValueError("no data found")
+        raise_if_no_data(res)
 
         if pandas:
             from datamaxi.resources.utils import convert_data_to_data_frame
@@ -124,11 +124,7 @@ async def get(
         )
 
         if pandas:
-            import pandas as pd
-
-            df = pd.DataFrame([res["data"]])
-            df = df.set_index("d")
-            return df
+            return to_indexed_dataframe([res["data"]], "d")
         return res
 
     async def exchanges(self, market: Market) -> List[str]:
@@ -184,11 +180,7 @@ async def __call__(
             "wallet_status", exchange=exchange, asset=asset
         )
         if pandas:
-            import pandas as pd
-
-            df = pd.DataFrame(res)
-            df = df.set_index("network")
-            return df
+            return to_indexed_dataframe(res, "network")
         return res
 
     async def exchanges(self) -> List[str]:

datamaxi/aio/premium.py

@@ -1,19 +1,25 @@
-"""Async premium resource — mirror of ``datamaxi.resources.premium``."""
+"""Async premium resource — mirror of ``datamaxi.resources.premium``.
+
+Param assembly and response shaping are shared with the sync resource via
+``build_premium_params`` / ``shape_premium_response`` (see #154); only the
+``await`` glue differs.
+"""
 
 from __future__ import annotations
 
 from typing import List, Union, Optional, TYPE_CHECKING
 
 from datamaxi.aio._core import AsyncResource
 from datamaxi.resources.responses import PremiumResponse
+from datamaxi.resources.premium import build_premium_params, shape_premium_response
 from datamaxi.lib.constants import Market, SortOrder
 
 if TYPE_CHECKING:
     import pandas as pd
 
 
 class AsyncPremium(AsyncResource):
-    async def __call__(  # noqa: C901
+    async def __call__(
         self,
         source_exchange: Optional[str] = None,
         target_exchange: Optional[str] = None,
@@ -38,94 +44,31 @@ async def __call__(  # noqa: C901
         query: Optional[str] = None,
         pandas: bool = True,
     ) -> Union[pd.DataFrame, PremiumResponse]:
-        params = {}
-
-        if source_exchange is not None:
-            params["source_exchange"] = source_exchange
-
-        if target_exchange is not None:
-            params["target_exchange"] = target_exchange
-
-        if asset is not None:
-            params["asset"] = asset
-
-        if source_quote is not None:
-            params["source_quote"] = source_quote
-
-        if target_quote is not None:
-            params["target_quote"] = target_quote
-
-        if sort is not None:
-            params["sort"] = sort
-
-        if key is not None:
-            params["key"] = key
-
-        if query is not None:
-            params["query"] = query
-
-        if page is not None:
-            params["page"] = page
-
-        if limit is not None:
-            params["limit"] = limit
-
-        if currency is not None:
-            params["currency"] = currency
-
-        if conversion_base is not None:
-            params["conversion_base"] = conversion_base
-
-        if min_sv is not None:
-            params["min_sv"] = min_sv
-
-        if min_tv is not None:
-            params["min_tv"] = min_tv
-
-        if source_market is not None:
-            params["source_market"] = source_market
-
-        if target_market is not None:
-            params["target_market"] = target_market
-
-        if only_transferable:
-            params["only_transferable"] = True
-
-        if network is not None:
-            params["network"] = network
-
-        if premium_type is not None:
-            params["premium_type"] = premium_type
-
-        if token_include is not None:
-            params["token_include"] = token_include
-
-        if token_exclude is not None:
-            params["token_exclude"] = token_exclude
-
+        params = build_premium_params(
+            source_exchange=source_exchange,
+            target_exchange=target_exchange,
+            asset=asset,
+            source_quote=source_quote,
+            target_quote=target_quote,
+            sort=sort,
+            key=key,
+            page=page,
+            limit=limit,
+            currency=currency,
+            conversion_base=conversion_base,
+            min_sv=min_sv,
+            min_tv=min_tv,
+            source_market=source_market,
+            target_market=target_market,
+            only_transferable=only_transferable,
+            network=network,
+            premium_type=premium_type,
+            token_include=token_include,
+            token_exclude=token_exclude,
+            query=query,
+        )
         res = await self.request_endpoint("premium", **params)
-        if res["data"] is None or len(res["data"]) == 0:
-            raise ValueError("no data found")
-
-        if pandas:
-            import pandas as pd
-
-            df = pd.DataFrame(
-                [
-                    {
-                        **item["detail"],
-                        "source_annualized_funding_rate": item.get(
-                            "source_annualized_funding_rate"
-                        ),
-                        "target_annualized_funding_rate": item.get(
-                            "target_annualized_funding_rate"
-                        ),
-                    }
-                    for item in res["data"]
-                ]
-            )
-            return df
-        return res
+        return shape_premium_response(res, pandas)
 
     async def exchanges(self) -> List[str]:
         return await self.request_endpoint("premium_exchanges")

datamaxi/api.py

@@ -7,7 +7,7 @@
 from .__version__ import __version__
 from datamaxi.lib.utils import cleanNoneValue
 from datamaxi.lib.utils import encoded_string
-from datamaxi._dispatch import resolve_endpoint, raise_for_error
+from datamaxi._dispatch import resolve_endpoint, raise_for_error, extract_limit_usage
 
 
 class API(object):
@@ -93,6 +93,11 @@ def _mount_retries(self, max_retries, retry_backoff, retry_statuses):
         ``_handle_exception`` still raises ``ServerError`` — preserving
         the existing error contract instead of leaking urllib3's
         ``MaxRetryError``.
+
+        This is the canonical retry policy; ``datamaxi._retry`` documents
+        the same GET-only/backoff/``Retry-After`` semantics for the async
+        (``httpx``) client, which has no urllib3-equivalent adapter to
+        mount this on directly.
         """
         retry = Retry(
             total=max_retries,
@@ -171,26 +176,12 @@ def send_request(self, http_method, url_path, payload=None):
         self.last_response = ResponseMeta(
             status_code=response.status_code,
             headers=response.headers,
-            limit_usage=self._extract_limit_usage(response.headers),
+            limit_usage=extract_limit_usage(response.headers),
             data=data,
         )
 
         return data
 
-    @staticmethod
-    def _extract_limit_usage(headers):
-        """Pull the ``x-ratelimit-*`` triplet out of the response headers."""
-        usage = {}
-        for key in headers.keys():
-            k = key.lower()
-            if (
-                k.startswith("x-ratelimit-limit")
-                or k.startswith("x-ratelimit-remaining")
-                or k.startswith("x-ratelimit-reset")
-            ):
-                usage[k] = headers[key]
-        return usage
-
     def _prepare_params(self, params):
         return encoded_string(cleanNoneValue(params))