API Client

A reusable REST API client wrapper built on requests.Session that handles authentication, retries, rate limiting, and pagination — the plumbing behind OSINT and security-tooling integrations.

Project Overview

Security automation constantly talks to REST APIs — threat-intel feeds, asset inventories, CI systems, cloud providers. Doing it well means one persistent session, sane timeouts, automatic retries on transient errors, token auth from the environment (never hard-coded), and transparent pagination so callers get every result. This project builds a small, well-behaved ApiClient class plus a CLI that fetches all pages of a paginated endpoint.

[!warning] Use credentials responsibly Only call APIs you are authorized to use, within their rate limits and terms of service. Load tokens from environment variables or a secrets manager — never commit them to notes or code.

Objectives

Interact with a REST API using a persistent session.
Authenticate without hardcoding credentials.
Handle pagination, rate limits, and transient failures.
Retry with exponential backoff on the right status codes only.
Return typed results rather than raw JSON.

Requirements

Item	Detail
Python	3.9 or newer
Dependencies	`requests`; `dataclasses`, `os`, `time` from the standard library
Privileges	None
Credentials	Supplied via environment variable — never hardcoded
Target	A local mock API, or an API you are authorized to use

Architecture

Session — one requests.Session reuses TCP connections and default headers.
Auth — bearer token pulled from an environment variable and set as Authorization.
Retry policy — urllib3.util.Retry mounted on an HTTPAdapter retries 429/5xx with backoff.
Pagination — a generator follows next links (or ?page=) until exhausted, yielding items.
CLI — argparse picks the endpoint and page size; results are printed or counted.

env token ─▶ Session(auth + retries) ─▶ GET /items?page=1
                                             │ yield items
                                             ▼
                                   follow 'next' ─▶ ... ─▶ done

Directory Structure

api-client/
├── api_client.py       # ApiClient class + CLI
└── requirements.txt    # requests

Implementation Plan

One request. Fetch a JSON endpoint with a timeout and call raise_for_status().
Use a session. Move to requests.Session() so connections are pooled and headers are set once.
Authenticate. Read the token from os.environ, and fail clearly if it is absent.
Model the response. Parse into a @dataclass so callers get typed attributes, not dictionary lookups.
Handle pagination. Follow next links or page parameters until exhausted, yielding results lazily.
Retry correctly. Back off on 429, 502, 503, and 504; do not retry a 401 or a 400.
Respect rate limits. Honour Retry-After when present rather than guessing.
Wrap errors. Raise a domain-specific exception so callers do not depend on requests internals.

Complete Example Implementation

#!/usr/bin/env python3
"""api_client.py - Reusable REST client: auth, retries, pagination (educational).

Only call APIs you are authorized to use. Load tokens from the environment.
"""
import argparse
import os
import sys
from typing import Iterator

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


class ApiClient:
    """Thin wrapper around requests.Session with retries and pagination."""

    def __init__(self, base_url: str, token: str | None = None, timeout: float = 10.0):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({"Accept": "application/json",
                                     "User-Agent": "StudyApiClient/1.0"})
        if token:
            self.session.headers["Authorization"] = f"Bearer {token}"

        # Retry transient failures with exponential backoff.
        retry = Retry(total=4, backoff_factor=0.5,
                      status_forcelist=(429, 500, 502, 503, 504),
                      allowed_methods=("GET",))
        adapter = HTTPAdapter(max_retries=retry)
        self.session.mount("https://", adapter)
        self.session.mount("http://", adapter)

    def get(self, path: str, **params) -> dict:
        """GET a single JSON resource."""
        resp = self.session.get(f"{self.base_url}/{path.lstrip('/')}",
                                params=params, timeout=self.timeout)
        resp.raise_for_status()
        return resp.json()

    def paginate(self, path: str, per_page: int = 50, item_key: str | None = None
                 ) -> Iterator[dict]:
        """Yield every item across pages, following ?page= until empty."""
        page = 1
        while True:
            data = self.get(path, page=page, per_page=per_page)
            items = data[item_key] if item_key else data
            if not items:
                break
            yield from items
            if len(items) < per_page:   # short page == last page
                break
            page += 1


def main() -> int:
    parser = argparse.ArgumentParser(description="Paginated REST API client.")
    parser.add_argument("base_url", help="API base URL, e.g. https://api.example.com")
    parser.add_argument("path", help="endpoint path, e.g. /v1/items")
    parser.add_argument("--per-page", type=int, default=50)
    parser.add_argument("--item-key", default=None,
                        help="JSON key holding the list (if not a top-level array)")
    parser.add_argument("--token-env", default="API_TOKEN",
                        help="env var holding the bearer token (default: API_TOKEN)")
    args = parser.parse_args()

    client = ApiClient(args.base_url, token=os.environ.get(args.token_env))
    try:
        count = 0
        for item in client.paginate(args.path, args.per_page, args.item_key):
            count += 1
            print(item.get("id", item))
        print(f"\n[*] Retrieved {count} item(s).")
    except requests.HTTPError as exc:
        print(f"[!] HTTP error: {exc}", file=sys.stderr)
        return 1
    except requests.RequestException as exc:
        print(f"[!] Request failed: {exc}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

pip install requests
export API_TOKEN="your-token-here"
python api_client.py https://api.github.com /users/anthropics/repos --per-page 30 --token-env API_TOKEN

Output

64778136
132935648
...
[*] Retrieved 42 item(s).

Usage

# Credentials come from the environment, never the command line
export LAB_API_TOKEN='your-lab-token'
export LAB_API_BASE='http://127.0.0.1:8080'

# Fetch a resource
python3 api_client.py get /v1/hosts

# Paginated listing
python3 api_client.py list /v1/findings --page-size 50

# With retry tuning
python3 api_client.py list /v1/findings --max-retries 5 --timeout 15

Example Output

$ python3 api_client.py list /v1/findings --page-size 2
[*] GET http://127.0.0.1:8080/v1/findings?page=1&size=2  -> 200 (0.04s)
[*] GET http://127.0.0.1:8080/v1/findings?page=2&size=2  -> 200 (0.03s)
[*] GET http://127.0.0.1:8080/v1/findings?page=3&size=2  -> 200 (0.03s, empty)

Finding(id=1, severity='high',   title='Exposed .git directory')
Finding(id=2, severity='medium', title='Missing HSTS header')
Finding(id=3, severity='low',    title='Server version disclosure')

[*] 3 findings across 3 requests

$ unset LAB_API_TOKEN && python3 api_client.py get /v1/hosts
[!] LAB_API_TOKEN is not set - refusing to continue

Error Handling

Condition	Status / Exception	Response
Missing credentials	—	Fail before the first request with a clear message; exit 2
Bad credentials	401	Raise `AuthError` — do not retry
Malformed request	400	Raise `ClientError` with the response body — do not retry
Not found	404	Return `None` or raise `NotFound`, depending on the call
Rate limited	429	Sleep for `Retry-After`, then retry up to the limit
Server error	502/503/504	Retry with exponential backoff
Network failure	`requests.ConnectionError`	Retry, then raise `ApiUnavailable`
Response is not JSON	`json.JSONDecodeError`	Raise with the first 200 bytes of the body for diagnosis

Retrying a 401 or a 400 is always wrong — the request will never succeed and you may trigger an account lockout.

Security Considerations

[!warning] Authorized use only Interact only with APIs you own or are explicitly authorized to use, and stay within their terms of service and rate limits.

Never hardcode credentials. Read tokens from environment variables or a secrets manager. A token committed to a repository must be treated as compromised and rotated.
Never pass tokens as command-line arguments — the full command line is visible via ps and lands in shell history.
Never log the Authorization header. Redact it explicitly in any request logging, and give any client class a __repr__ that masks the token.
Always verify TLS. verify=False exposes the token to anyone who can intercept the connection.
Respect Retry-After and rate limits. Aggressive retrying is indistinguishable from an attack and may get your access revoked.
Validate everything you receive. API responses are untrusted input — never eval() them, and check types before use.
Scope the token minimally — read-only where reading is all you need.

Testing

from unittest.mock import MagicMock, patch

import pytest

import api_client


def test_missing_token_fails_fast(monkeypatch):
    monkeypatch.delenv("LAB_API_TOKEN", raising=False)
    with pytest.raises(api_client.ConfigError):
        api_client.Client.from_env()


def test_401_is_not_retried():
    with patch("requests.Session.get") as get:
        get.return_value = MagicMock(status_code=401)
        client = api_client.Client(base="http://x", token="t")
        with pytest.raises(api_client.AuthError):
            client.get("/v1/hosts")
        assert get.call_count == 1          # exactly one attempt


def test_429_is_retried_with_backoff():
    responses = [MagicMock(status_code=429, headers={"Retry-After": "0"}),
                 MagicMock(status_code=200, json=lambda: {"items": []})]
    with patch("requests.Session.get", side_effect=responses) as get:
        client = api_client.Client(base="http://x", token="t")
        client.get("/v1/findings")
        assert get.call_count == 2

Mock every HTTP call — tests must never touch a real API. Cover: success, 401, 429 with retry, 500 with backoff, pagination, and malformed JSON.

Extension Ideas

Rate-limit awareness — read X-RateLimit-Remaining/Retry-After and sleep proactively.
Cursor pagination — support Link: rel="next" headers and opaque cursors, not just ?page=.
Async — swap to httpx.AsyncClient for high-throughput fan-out.
Response caching — add requests-cache for idempotent GETs during development.
Typed models — parse responses into dataclass/pydantic models for safer downstream code.

Troubleshooting

Symptom	Likely cause	Fix
401 on every request	Token missing, expired, or wrong header format	Check the scheme — `Bearer <token>` versus a raw token
Client hangs	No timeout set on the session	Pass `timeout=` to every request
Only the first page returned	Pagination not followed	Follow the `next` link or increment the page parameter until empty
Repeated 429s	Retrying too aggressively	Honour `Retry-After` and add jitter to the backoff
`SSLError`	Corporate proxy or self-signed certificate	Point `REQUESTS_CA_BUNDLE` at the right CA — do not disable verification
`JSONDecodeError`	HTML error page returned instead of JSON	Log the status and the first bytes of the body
Token appears in logs	Header logged verbatim	Redact `Authorization` before logging

References

[[Web-Crawler]]
[[Whois-Lookup]]
[[DNS-Enumeration]]
[[Mini-Projects/Readme|Mini-Projects]] — module index
[[Readme|Python for Security Professionals]] — course home

Navigation Menu

Uh oh!

API Client

Project Overview

Objectives

Requirements

Architecture

Directory Structure

Implementation Plan

Complete Example Implementation

Output

Usage

Example Output

Error Handling

Security Considerations

Testing

Extension Ideas

Troubleshooting

References

Related

Uh oh!

FilesExpand file tree

API-Client.md

Latest commit

History

API-Client.md

File metadata and controls

API Client

Project Overview

Objectives

Requirements

Architecture

Directory Structure

Implementation Plan

Complete Example Implementation

Output

Usage

Example Output

Error Handling

Security Considerations

Testing

Extension Ideas

Troubleshooting

References

Related