Skip to content

⚙️ DogeAPIClient

The DogeAPIClient is the low-level engine powering all API interactions.
It wraps the httpx.Client and provides retry logic, timeouts, and request abstraction.

🔌 Class Reference

DogeAPIClient

Python
DogeAPIClient(base_url: str = BASE_URL, timeout: float = TIMEOUT, session: Optional[Client] = None, max_retries: int = 5, backoff_factor: float = 1.5, max_backoff: float = 60.0, **httpx_kwargs)

DOGE API client with built-in retry logic for unstable endpoints.

METHOD DESCRIPTION
rest_request

Perform a REST request with retry logic for 429 and 5xx errors.
Source code in src/pydoge_api/client.py
Python
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def __init__(
    self,
    base_url: str = BASE_URL,
    timeout: float = TIMEOUT,
    session: Optional[httpx.Client] = None,
    max_retries: int = 5,
    backoff_factor: float = 1.5,
    max_backoff: float = 60.0,
    **httpx_kwargs,
):
    self.base_url = base_url
    self.timeout = timeout
    self.max_retries = max_retries
    self.backoff_factor = backoff_factor
    self.max_backoff = max_backoff

    if session:
        if not isinstance(session, httpx.Client):
            raise TypeError("Custom session must be an instance of httpx.Client")
        self.client = session
        self._owns_session = False
    else:
        self.client = httpx.Client(base_url=base_url, timeout=timeout, **httpx_kwargs)
        self._owns_session = True

rest_request

Python
rest_request(method: str, url: str, **kwargs) -> Response

Perform a REST request with retry logic for 429 and 5xx errors.

PARAMETER DESCRIPTION

method

HTTP method (GET, POST, etc.)

TYPE: str

url

Full target URL.

TYPE: str

**kwargs

Passed to httpx.request() (params, json, data, headers, etc.)

TYPE: dict DEFAULT: {}

RETURNS DESCRIPTION
Response

Response object, or raises DogeAPIRequestError.
Source code in src/pydoge_api/client.py
Python
 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
def rest_request(self, method: str, url: str, **kwargs) -> httpx.Response:
    """
    Perform a REST request with retry logic for 429 and 5xx errors.

    Parameters
    ----------
    method : str
        HTTP method (GET, POST, etc.)
    url : str
        Full target URL.
    **kwargs : dict
        Passed to httpx.request() (params, json, data, headers, etc.)

    Returns
    -------
    httpx.Response
        Response object, or raises DogeAPIRequestError.
    """
    retriable = {429, 500, 502, 503, 504}
    retries = 0

    while retries <= self.max_retries:
        try:
            logger.debug(f"→ {method} {url} params={kwargs.get('params') or {}}")
            response = self.client.request(method, url, **kwargs)
            logger.debug(f"← {response.status_code} {method} {url} ({len(response.content)} bytes)")
            if response.status_code < 400:
                return response
            if response.status_code not in retriable:
                # Non-retriable client/server error (400, 401, 403, 404, ...):
                # surface a clear, typed error instead of returning the body.
                raise DogeAPIRequestError(
                    method, url, response.status_code, response.text[:200] or "Request failed"
                )
        except httpx.RequestError as e:
            logger.warning(f"⚠️ Network error during {method} {url}: {e}")
            err_response = getattr(e, "response", None)
            if err_response is None:
                raise
            response = err_response

        # Retry triggered
        retry_after = response.headers.get("Retry-After")
        if retry_after:
            try:
                wait = float(retry_after)
            except ValueError:
                wait = self.backoff_factor * (retries + 1)
        else:
            jitter = random.uniform(0, 0.3)
            wait = self.backoff_factor * (2**retries) + jitter

        # Cap the wait so large retry counts can't produce multi-minute sleeps.
        wait = min(wait, self.max_backoff)

        logger.warning(
            f"🔁 Retry {retries + 1}/{self.max_retries} for {method} {url} "
            f"→ HTTP {response.status_code}. Waiting {wait:.2f}s..."
        )
        time.sleep(wait)
        retries += 1

    logger.error(f"❌ {method} {url} failed after {self.max_retries} retries.")
    raise DogeAPIRequestError(method, url, response.status_code, "Max retries exceeded")