Working with APIs & Webhooks

import httpx
 
with httpx.Client(timeout=30.0) as client:
    resp = client.get(
        "https://api.example.com/v1/items",
        headers={"Authorization": "Bearer TOKEN"},
        params={"page": 1, "limit": 100},
    )
    resp.raise_for_status()
    data = resp.json()

import hashlib
import hmac
import time
from typing import Any, Iterator
 
import httpx
 
class ApiClient:
    def __init__(self, base_url: str, token: str) -> None:
        self._client = httpx.Client(
            base_url=base_url,
            headers={"Authorization": f"Bearer {token}"},
            timeout=httpx.Timeout(30.0),
        )
 
    def list_items(self, page_size: int = 100) -> Iterator[dict[str, Any]]:
        cursor: str | None = None
        while True:
            params = {"limit": page_size}
            if cursor:
                params["cursor"] = cursor
            resp = self._client.get("/v1/items", params=params)
            resp.raise_for_status()
            payload = resp.json()
            for item in payload.get("items", []):
                yield item
            cursor = payload.get("next_cursor")
            if not cursor:
                break
 
    def close(self) -> None:
        self._client.close()
 
def verify_webhook(body: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)
 
if __name__ == "__main__":
    client = ApiClient("https://api.example.com", "dev-token")
    for i, item in enumerate(client.list_items()):
        if i >= 3:
            break
        print(item.get("id"))
    client.close()

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential
 
@retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=0.5))
def fetch(client: httpx.Client, url: str) -> httpx.Response:
    resp = client.get(url)
    resp.raise_for_status()
    return resp