From 29229a10ab215f101c8f0d0605e0e1fc9677fa8d Mon Sep 17 00:00:00 2001 From: Martin Kersner Date: Sat, 4 Jul 2026 20:56:23 +0900 Subject: [PATCH] refactor: share response-shaping + align async retry policy (#154) Extract transport-agnostic response-shaping into datamaxi.resources.utils (assemble_params/raise_if_no_data/to_indexed_dataframe) and premium-specific build_premium_params/shape_premium_response in resources/premium.py, imported directly by aio/premium.py and aio/cex.py. Only await/transport glue is left per-client now; sync/async can't drift on shaping. Add datamaxi/_retry.py: pure GET-only/exponential-backoff/Retry-After policy (mirrors urllib3 Retry semantics). Rewire aio/_core.py's hand-rolled loop (previously linear backoff, all methods, ignored Retry-After) onto it. Sync keeps its urllib3 Retry unchanged (well-tested, no need to touch); _retry.py is the shared description so the two can't diverge again. Also collapse api.py's duplicate _extract_limit_usage onto the one already shared via _dispatch.extract_limit_usage (used by aio already). No public API/behavior change. Extended test_endpoint_param_coverage.py's AST extractor to recognize the new params = build_x_params(x=x, ...) call-site shape alongside the existing params["x"] = ... pattern it already looked for. --- datamaxi/_retry.py | 86 ++++++++++ datamaxi/aio/_core.py | 30 ++-- datamaxi/aio/cex.py | 16 +- datamaxi/aio/premium.py | 121 ++++---------- datamaxi/api.py | 23 +-- datamaxi/resources/cex_candle.py | 5 +- datamaxi/resources/cex_ticker.py | 7 +- datamaxi/resources/cex_wallet_status.py | 7 +- datamaxi/resources/premium.py | 203 +++++++++++++----------- datamaxi/resources/utils.py | 48 +++++- tests/test_endpoint_param_coverage.py | 25 ++- tests/test_premium_shaping_shared.py | 18 +++ tests/test_resources_utils.py | 64 ++++++++ tests/test_retry_policy.py | 183 +++++++++++++++++++++ 14 files changed, 602 insertions(+), 234 deletions(-) create mode 100644 datamaxi/_retry.py create mode 100644 tests/test_premium_shaping_shared.py create mode 100644 tests/test_resources_utils.py create mode 100644 tests/test_retry_policy.py diff --git a/datamaxi/_retry.py b/datamaxi/_retry.py new file mode 100644 index 0000000..da2d101 --- /dev/null +++ b/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) diff --git a/datamaxi/aio/_core.py b/datamaxi/aio/_core.py index 798690d..2e13df3 100644 --- a/datamaxi/aio/_core.py +++ b/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) diff --git a/datamaxi/aio/cex.py b/datamaxi/aio/cex.py index 90f38db..dc6bc8d 100644 --- a/datamaxi/aio/cex.py +++ b/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]: diff --git a/datamaxi/aio/premium.py b/datamaxi/aio/premium.py index d412c0c..f021d99 100644 --- a/datamaxi/aio/premium.py +++ b/datamaxi/aio/premium.py @@ -1,4 +1,9 @@ -"""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 @@ -6,6 +11,7 @@ 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: @@ -13,7 +19,7 @@ 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") diff --git a/datamaxi/api.py b/datamaxi/api.py index 5fd1a0b..9aab299 100644 --- a/datamaxi/api.py +++ b/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)) diff --git a/datamaxi/resources/cex_candle.py b/datamaxi/resources/cex_candle.py index ebb3402..133768f 100644 --- a/datamaxi/resources/cex_candle.py +++ b/datamaxi/resources/cex_candle.py @@ -4,7 +4,7 @@ from datamaxi.api import Resource from datamaxi.lib.utils import check_required_parameter from datamaxi.lib.utils import check_required_parameters -from datamaxi.resources.utils import convert_data_to_data_frame +from datamaxi.resources.utils import convert_data_to_data_frame, raise_if_no_data from datamaxi.resources.responses import CandleResponse from datamaxi.lib.constants import SPOT, FUTURES, INTERVAL_1D, USD, Market, Interval @@ -79,8 +79,7 @@ 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: return convert_data_to_data_frame(res["data"]) diff --git a/datamaxi/resources/cex_ticker.py b/datamaxi/resources/cex_ticker.py index 9b4978e..8e90df4 100644 --- a/datamaxi/resources/cex_ticker.py +++ b/datamaxi/resources/cex_ticker.py @@ -3,6 +3,7 @@ from typing import Any, List, Union, Optional, TYPE_CHECKING from datamaxi.api import Resource from datamaxi.lib.utils import check_required_parameters +from datamaxi.resources.utils import to_indexed_dataframe from datamaxi.resources.responses import TickerResponse from datamaxi.lib.constants import SPOT, FUTURES, Market @@ -74,11 +75,7 @@ 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") else: return res diff --git a/datamaxi/resources/cex_wallet_status.py b/datamaxi/resources/cex_wallet_status.py index 5f2894f..ca33ef0 100644 --- a/datamaxi/resources/cex_wallet_status.py +++ b/datamaxi/resources/cex_wallet_status.py @@ -3,6 +3,7 @@ from typing import Any, List, Union, TYPE_CHECKING from datamaxi.api import Resource from datamaxi.resources.responses import WalletStatusRow +from datamaxi.resources.utils import to_indexed_dataframe from datamaxi.lib.utils import check_required_parameters from datamaxi.lib.utils import check_required_parameter @@ -51,11 +52,7 @@ def __call__( res = self.request_endpoint("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 diff --git a/datamaxi/resources/premium.py b/datamaxi/resources/premium.py index 3196c2b..9ace3f5 100644 --- a/datamaxi/resources/premium.py +++ b/datamaxi/resources/premium.py @@ -1,14 +1,101 @@ from __future__ import annotations -from typing import Any, List, Union, Optional, TYPE_CHECKING +from typing import Any, Dict, List, Union, Optional, TYPE_CHECKING from datamaxi.api import Resource from datamaxi.resources.responses import PremiumResponse +from datamaxi.resources.utils import assemble_params, raise_if_no_data from datamaxi.lib.constants import Market, SortOrder if TYPE_CHECKING: import pandas as pd +def build_premium_params( + source_exchange: Optional[str] = None, + target_exchange: Optional[str] = None, + asset: Optional[str] = None, + source_quote: Optional[str] = None, + target_quote: Optional[str] = None, + sort: Optional[SortOrder] = None, + key: Optional[str] = None, + page: int = 1, + limit: int = 100, + currency: Optional[str] = None, + conversion_base: Optional[str] = None, + min_sv: Optional[str] = None, + min_tv: Optional[str] = None, + source_market: Optional[Market] = None, + target_market: Optional[Market] = None, + only_transferable: bool = False, + network: Optional[str] = None, + premium_type: Optional[str] = None, + token_include: Optional[str] = None, + token_exclude: Optional[str] = None, + query: Optional[str] = None, +) -> Dict[str, Any]: + """Assemble the ``premium`` endpoint's query params from caller args. + + Transport-agnostic (no request is made here) so the sync and async + ``Premium.__call__`` methods share the exact same param-building logic — + see #154. + """ + return assemble_params( + ("source_exchange", source_exchange), + ("target_exchange", target_exchange), + ("asset", asset), + ("source_quote", source_quote), + ("target_quote", target_quote), + ("sort", sort), + ("key", key), + ("query", query), + ("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", True if only_transferable else None), + ("network", network), + ("premium_type", premium_type), + ("token_include", token_include), + ("token_exclude", token_exclude), + ) + + +def shape_premium_response( + res: PremiumResponse, pandas: bool +) -> Union[pd.DataFrame, PremiumResponse]: + """Turn a raw ``premium`` response into the DataFrame or typed dict shape. + + Shared by the sync and async ``Premium.__call__`` so the "no data" + check and DataFrame construction can't drift between the two — see + #154. + """ + raise_if_no_data(res) + + if not pandas: + return res + + import pandas as pd + + return 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"] + ] + ) + + class Premium(Resource): """Client to fetch premium data from DataMaxi+ API.""" @@ -24,7 +111,7 @@ def __init__(self, api_key=None, **kwargs: Any): self.__module__ = __name__ self.__qualname__ = self.__class__.__qualname__ - def __call__( # noqa: C901 + def __call__( self, source_exchange: Optional[str] = None, target_exchange: Optional[str] = None, @@ -82,95 +169,31 @@ def __call__( # noqa: C901 Returns: Premium data in pandas DataFrame """ - 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 = 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 - else: - return res + return shape_premium_response(res, pandas) def exchanges(self) -> List[str]: """Fetch supported exchanges for premium data. diff --git a/datamaxi/resources/utils.py b/datamaxi/resources/utils.py index 09b2f40..1444801 100644 --- a/datamaxi/resources/utils.py +++ b/datamaxi/resources/utils.py @@ -1,11 +1,57 @@ +"""Transport-agnostic response-shaping helpers shared by the sync and async +resources (see #154). + +Kept alongside ``datamaxi._dispatch`` (which shares *request*-building) as +the shared *response*-shaping layer: param assembly, the "no data" envelope +check, and the DataFrame-vs-raw-dict conversion decision. Pure functions +only (deferred ``pandas`` import, no ``requests``/``httpx``), so both +``datamaxi.resources.*`` and ``datamaxi.aio.*`` can import this leaf module +without any import-cycle risk. +""" + from __future__ import annotations -from typing import List, TYPE_CHECKING +from typing import Any, Dict, List, Tuple, TYPE_CHECKING if TYPE_CHECKING: import pandas as pd +def assemble_params(*pairs: Tuple[str, Any]) -> Dict[str, Any]: + """Build a query-param dict from ``(name, value)`` pairs, dropping ``None``. + + Mirrors the repeated ``if x is not None: params["x"] = x`` blocks that + were hand-copied across sync/async resource methods. For "only include + when truthy" flags (e.g. ``only_transferable``), pass ``value if value + else None`` at the call site. + """ + return {name: value for name, value in pairs if value is not None} + + +def raise_if_no_data(res: Dict[str, Any], check_length: bool = True) -> None: + """Raise ``ValueError("no data found")`` for an empty ``{"data": ...}`` envelope. + + ``check_length`` matches call sites that also treat an empty + list/dict as "no data" (e.g. candle, premium), vs. ones that only + check for ``None`` (e.g. announcements, token updates). + """ + if res["data"] is None or (check_length and len(res["data"]) == 0): + raise ValueError("no data found") + + +def to_indexed_dataframe(rows: List, index_col: str) -> pd.DataFrame: + """``pd.DataFrame(rows).set_index(index_col)`` — the ticker/wallet-status shape. + + ``rows`` is already a list of dicts (wrap a single dict as ``[rows]`` at + the call site, as the ticker endpoint's bare-object response does). + """ + import pandas as pd + + df = pd.DataFrame(rows) + df = df.set_index(index_col) + return df + + def convert_data_to_data_frame( data: List, columns_to_replace: List[str] = [], diff --git a/tests/test_endpoint_param_coverage.py b/tests/test_endpoint_param_coverage.py index 660cc92..073d622 100644 --- a/tests/test_endpoint_param_coverage.py +++ b/tests/test_endpoint_param_coverage.py @@ -8,8 +8,9 @@ This test statically extracts, per ``op_id``, the union of keyword arguments forwarded at every ``request_endpoint("op_id", ...)`` call site (handling the -``**{"from": ...}`` dict-splat and the ``**params`` local-dict patterns), then -asserts that every registry param is either forwarded, globally ignored +``**{"from": ...}`` dict-splat, the ``**params`` local-dict pattern, and the +``params = build_x_params(x=x, ...)`` extracted-builder pattern — see #154), +then asserts that every registry param is either forwarded, globally ignored (pagination), or explicitly allow-listed below with a rationale. Regenerating ``_endpoints.py`` (``make python`` upstream) that adds a new param @@ -71,7 +72,17 @@ def _enclosing_func(parents, node): def _subscript_string_keys(func_node, name): - """Collect literal keys of ``name[] = ...`` assignments in ``func``.""" + """Collect the wire param names that populate the local dict ``name``. + + Handles two ways a resource method builds its ``params`` dict before + ``request_endpoint(op, **params)``: + + - ``name["x"] = ...`` assignments inline in the method. + - ``name = build_x_params(x=x, y=y, ...)`` — a call to an extracted, + transport-agnostic param-builder shared with the async mirror (see + #154); the keyword-argument names at the call site are the wire param + names it forwards (builders are written to keep them 1:1). + """ keys = set() if func_node is None: return keys @@ -87,6 +98,14 @@ def _subscript_string_keys(func_node, name): and isinstance(tgt.slice.value, str) ): keys.add(tgt.slice.value) + elif ( + isinstance(tgt, ast.Name) + and tgt.id == name + and isinstance(n.value, ast.Call) + ): + for kw in n.value.keywords: + if kw.arg is not None: + keys.add(kw.arg) return keys diff --git a/tests/test_premium_shaping_shared.py b/tests/test_premium_shaping_shared.py new file mode 100644 index 0000000..b5512bb --- /dev/null +++ b/tests/test_premium_shaping_shared.py @@ -0,0 +1,18 @@ +"""Anti-drift check: the async premium resource reuses the sync module's +param-builder / response-shaper (see #154) rather than a hand-copied one. +""" + +import pytest + +httpx = pytest.importorskip("httpx") + +from datamaxi.aio import premium as aio_premium # noqa: E402 +from datamaxi.resources import premium as sync_premium # noqa: E402 + + +def test_async_premium_reuses_sync_param_builder(): + assert aio_premium.build_premium_params is sync_premium.build_premium_params + + +def test_async_premium_reuses_sync_response_shaper(): + assert aio_premium.shape_premium_response is sync_premium.shape_premium_response diff --git a/tests/test_resources_utils.py b/tests/test_resources_utils.py new file mode 100644 index 0000000..29aec70 --- /dev/null +++ b/tests/test_resources_utils.py @@ -0,0 +1,64 @@ +"""Tests for the shared response-shaping helpers (``datamaxi.resources.utils``, +see #154) — param assembly, the "no data" check, and DataFrame shaping used +by both the sync and async resources. +""" + +import pandas as pd +import pytest + +from datamaxi.resources.utils import ( + assemble_params, + raise_if_no_data, + to_indexed_dataframe, +) + + +def test_assemble_params_drops_none_and_keeps_order(): + params = assemble_params( + ("a", 1), + ("b", None), + ("c", "x"), + ) + assert params == {"a": 1, "c": "x"} + assert list(params) == ["a", "c"] + + +def test_assemble_params_only_truthy_flag_pattern(): + # Call-site pattern for "include only if truthy" flags like + # only_transferable: pass `True if flag else None`. + included = assemble_params(("only_transferable", True if True else None)) + excluded = assemble_params(("only_transferable", True if False else None)) + assert included == {"only_transferable": True} + assert excluded == {} + + +def test_raise_if_no_data_raises_on_none(): + with pytest.raises(ValueError): + raise_if_no_data({"data": None}) + + +def test_raise_if_no_data_raises_on_empty_by_default(): + with pytest.raises(ValueError): + raise_if_no_data({"data": []}) + + +def test_raise_if_no_data_allows_empty_when_length_check_disabled(): + raise_if_no_data({"data": []}, check_length=False) # no raise + + +def test_raise_if_no_data_passes_with_data(): + raise_if_no_data({"data": [{"x": 1}]}) + + +def test_to_indexed_dataframe_single_row(): + df = to_indexed_dataframe([{"d": "123", "p": "1.5"}], "d") + assert isinstance(df, pd.DataFrame) + assert df.index.name == "d" + assert list(df.index) == ["123"] + + +def test_to_indexed_dataframe_multi_row(): + df = to_indexed_dataframe( + [{"network": "BSC", "x": 1}, {"network": "ETH", "x": 2}], "network" + ) + assert list(df.index) == ["BSC", "ETH"] diff --git a/tests/test_retry_policy.py b/tests/test_retry_policy.py new file mode 100644 index 0000000..d8d3786 --- /dev/null +++ b/tests/test_retry_policy.py @@ -0,0 +1,183 @@ +"""Tests for the shared retry policy (``datamaxi._retry``, see #154). + +Covers the pure policy functions directly, plus the async transport's use of +them: GET-only, exponential backoff, and honoring ``Retry-After`` — aligning +the ``httpx``-based async client with the ``urllib3``-backed sync retry +policy exercised in ``tests/test_retry.py``. +""" + +import asyncio + +import pytest + +from datamaxi._retry import ( + is_retryable, + parse_retry_after, + get_backoff_time, + get_retry_delay, +) + +httpx = pytest.importorskip("httpx") + +from datamaxi.aio._core import AsyncAPI # noqa: E402 + +BASE_URL = "https://api.datamaxiplus.com" + + +# --- pure policy functions --------------------------------------------------- +def test_is_retryable_get_only(): + assert is_retryable("GET", 503, 1, 3, (502, 503, 504)) + assert is_retryable("get", 503, 1, 3, (502, 503, 504)) # case-insensitive + assert not is_retryable("POST", 503, 1, 3, (502, 503, 504)) + + +def test_is_retryable_only_for_listed_statuses(): + assert not is_retryable("GET", 200, 1, 3, (502, 503, 504)) + + +def test_is_retryable_respects_max_retries(): + assert is_retryable("GET", 503, 3, 3, (503,)) + assert not is_retryable("GET", 503, 4, 3, (503,)) + + +def test_get_backoff_time_is_exponential_with_zero_first_retry(): + # Matches urllib3.Retry.get_backoff_time(): no delay before the first + # retry, then backoff_factor * 2 ** (n - 1). + assert get_backoff_time(1, 0.5) == 0.0 + assert get_backoff_time(2, 0.5) == 1.0 + assert get_backoff_time(3, 0.5) == 2.0 + assert get_backoff_time(4, 0.5) == 4.0 + + +def test_get_backoff_time_caps_at_backoff_max(): + assert get_backoff_time(20, 10.0) == 120.0 + + +def test_parse_retry_after_seconds_form(): + assert parse_retry_after("120") == 120.0 + + +def test_parse_retry_after_missing_returns_none(): + assert parse_retry_after(None) is None + + +def test_parse_retry_after_never_negative(): + # An HTTP-date in the past yields a would-be-negative delta. + assert parse_retry_after("Mon, 01 Jan 2001 00:00:00 GMT") == 0.0 + + +def test_get_retry_delay_prefers_retry_after_header(): + assert get_retry_delay(4, 0.5, {"Retry-After": "7"}) == 7.0 + + +def test_get_retry_delay_falls_back_to_backoff(): + assert get_retry_delay(3, 0.5, {}) == 2.0 + + +# --- async transport wiring --------------------------------------------------- +def _run(coro): + return asyncio.run(coro) + + +def test_async_get_retries_transient_5xx_with_exponential_backoff(monkeypatch): + sleeps = [] + + async def fake_sleep(seconds): + sleeps.append(seconds) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + calls = {"n": 0} + + def handler(request): + calls["n"] += 1 + if calls["n"] < 4: + return httpx.Response(503, json={"error": "busy"}) + return httpx.Response(200, json={"ok": True}) + + async def run(): + api = AsyncAPI( + api_key="k", + base_url=BASE_URL, + max_retries=3, + retry_backoff=0.5, + transport=httpx.MockTransport(handler), + ) + try: + return await api.send_request("GET", "/x") + finally: + await api.aclose() + + data = _run(run()) + assert data == {"ok": True} + assert calls["n"] == 4 + # No delay before the 1st retry, then exponential growth (factor * 2**(n-1)). + assert sleeps == [0.0, 1.0, 2.0] + + +def test_async_post_is_not_retried(monkeypatch): + async def fake_sleep(seconds): + raise AssertionError("POST must not be retried") + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + calls = {"n": 0} + + def handler(request): + calls["n"] += 1 + return httpx.Response(503, json={"error": "busy"}) + + async def run(): + api = AsyncAPI( + api_key="k", + base_url=BASE_URL, + max_retries=3, + transport=httpx.MockTransport(handler), + ) + try: + await api.send_request("POST", "/x") + finally: + await api.aclose() + + from datamaxi.error import ServerError + + with pytest.raises(ServerError): + _run(run()) + assert calls["n"] == 1 + + +def test_async_honors_retry_after_header(monkeypatch): + sleeps = [] + + async def fake_sleep(seconds): + sleeps.append(seconds) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + calls = {"n": 0} + + def handler(request): + calls["n"] += 1 + if calls["n"] < 2: + return httpx.Response( + 503, json={"error": "busy"}, headers={"Retry-After": "3"} + ) + return httpx.Response(200, json={"ok": True}) + + async def run(): + api = AsyncAPI( + api_key="k", + base_url=BASE_URL, + max_retries=3, + retry_backoff=0.5, + transport=httpx.MockTransport(handler), + ) + try: + return await api.send_request("GET", "/x") + finally: + await api.aclose() + + data = _run(run()) + assert data == {"ok": True} + # Retry-After (3s) takes priority over the computed backoff (0s). + assert sleeps == [3.0]