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 

DogeAPIClient(
    base_url: str = "https://api.doge.gov",
    timeout: float = 10.0,
    session: Optional[Client] = None,
    max_retries: int = 5,
    backoff_factor: float = 1.5,
    **httpx_kwargs
)

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

Methods:

  • rest_request –

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

Source code in src/pydoge_api/client.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def __init__(
    self,
    base_url: str = "https://api.doge.gov",
    timeout: float = 10.0,
    session: Optional[httpx.Client] = None,
    max_retries: int = 5,
    backoff_factor: float = 1.5,
    **httpx_kwargs
):
    self.base_url = base_url
    self.timeout = timeout
    self.max_retries = max_retries
    self.backoff_factor = backoff_factor

    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 

rest_request(method: str, url: str, **kwargs) -> 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, default: {} ) –

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

Returns:

  • Response –

    Response object, or raises DogeAPIRequestError.

Source code in src/pydoge_api/client.py 
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
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:
            response = self.client.request(method, url, **kwargs)
            if response.status_code < 400 or response.status_code not in retriable:
                return response
        except httpx.RequestError as e:
            logger.warning(f"⚠️ Network error during {method} {url}: {e}")
            response = getattr(e, "response", None)
            if response is None:
                raise

        # 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

        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")