Advanced Features

# Request interceptor -- adds auth token to every request
def auth_interceptor(config):
    config.headers['Authorization'] = 'Bearer my-secret-token'
    return config

# Response interceptor -- logs status of every response
def log_interceptor(response):
    print(f"{response.config.method} {response.config.url} -> {response.status}")
    return response

client.interceptors.request.use(auth_interceptor)
index = client.interceptors.response.use(log_interceptor)

# Remove an interceptor by index
client.interceptors.response.eject(index)

# Lightweight observability hooks, no interceptor needed:
client.get(
    '/report',
    onRequestStart=lambda config: print(f"starting {config.url}"),
    onRetry=lambda config, attempt, error: print(f"retry #{attempt}: {error}"),
    onRedirect=lambda config, response: print(f"redirected: {response.status}"),
)

from atomhttp import AtomHTTP, AbortController
from atomhttp.errors import AtomHTTPCancelError

client = AtomHTTP(base_url="https://api.example.com", timeout=30)
controller = AbortController()

# From another thread, UI callback, etc:
# controller.abort("user clicked cancel")

try:
    response = client.get("/slow-report", signal=controller.signal)
except AtomHTTPCancelError as e:
    print("request was cancelled:", e.message)

# One controller can cancel multiple in-flight requests at once.
# Works identically with AsyncAtomHTTP:
async def run():
    try:
        await async_client.get("/slow", signal=controller.signal)
    except AtomHTTPCancelError:
        print("cancelled")

client = AtomHTTP(base_url="https://api.example.com", max_workers=20)

# Run a batch concurrently:
responses = client.all([
    lambda: client.get("/a"),
    lambda: client.get("/b"),
    lambda: client.get("/c"),
])

# Fire-and-forget:
future = client.submit("GET", "/report")
# ... do other work ...
response = future.result()

# Same request, many URLs, results in input order:
responses = client.map("GET", ["/users/1", "/users/2", "/users/3"])

# Async equivalent:
responses = await async_client.all([
    async_client.get("/a"),
    async_client.get("/b"),
])

# Streaming a large response body
with client.stream("GET", "/export.csv") as response:
    for line in response.iter_lines():
        process(line)

# Downloading straight to disk, with progress
client.download(
    "/video.mp4", "video.mp4",
    onDownloadProgress=lambda loaded, total: print(f"{loaded}/{total}"),
)

# Pagination -- default ?page=N scheme
for items in client.paginate("/users"):
    for user in items:
        process(user)

# Pagination -- custom cursor-style API
for items in client.paginate(
    "/items",
    extract_items=lambda r: r.data["results"],
    has_next=lambda r: r.data.get("next") is not None,
):
    ...

# Async versions are identical, with async with / async for:
async with await async_client.stream("GET", "/export.csv") as response:
    async for line in response.iter_lines():
        process(line)

async for items in async_client.paginate("/users"):
    ...

from atomhttp import FormData
from pathlib import Path

form = FormData()
form.append('username', 'johndoe')
form.append('email', 'john@example.com')
form.append('avatar', open('profile.jpg', 'rb'), filename='profile.jpg')
form.append('document', Path('resume.pdf'), filename='resume.pdf')

# Supports multiple values for the same field
form.append('tags', 'python')
form.append('tags', 'http')
form.append('tags', 'async')

response = client.post('https://api.example.com/upload', data=form)

# Large file, with real upload progress (streams automatically):
form2 = FormData()
form2.append('video', Path('large-video.mp4'), filename='large-video.mp4')
client.post(
    '/upload', data=form2,
    onUploadProgress=lambda loaded, total: print(f"{loaded}/{total} bytes"),
)

from atomhttp import AtomHTTP
from atomhttp.cache import CacheInterceptor

client = AtomHTTP(base_url="https://api.example.com")
cache = CacheInterceptor()
client.interceptors.request.use(cache.on_request)
client.interceptors.response.use(cache.on_response)

client.get("/users/1")   # normal request, response cached (has ETag)
client.get("/users/1")   # sends If-None-Match; 304 -> served from cache

cache.clear()             # drop everything

client = AtomHTTP(base_url="https://api.example.com")   # cookies=True by default
client.get("/login")     # any Set-Cookie response headers are stored
client.get("/profile")   # matching cookies are sent automatically

no_cookies_client = AtomHTTP(base_url="...", cookies=False)

# Manual access
client.cookies.get("session")
client.cookies.set("session", "abc123", domain="api.example.com")
client.cookies.clear()

# XSRF -- automatic once the cookie is present in the jar
client.get("/state-changing-endpoint")  # X-XSRF-TOKEN header added automatically

# Preferred: built-in Basic Auth, applied automatically
response = client.get(
    'https://httpbin.org/basic-auth/username/password',
    auth={'username': 'username', 'password': 'password'},
)

# Helper classes for custom flows / interceptors
from atomhttp.auth import BasicAuth, BearerAuth

basic = BasicAuth('username', 'password')
response = client.get('https://httpbin.org/basic-auth/username/password',
                       headers=basic.get_header())

bearer = BearerAuth('your-jwt-token-here')
response = client.get('https://api.example.com/protected',
                       headers=bearer.get_header())

# Complex auth flows via interceptors
def oauth_interceptor(config):
    config.headers['Authorization'] = f'Bearer {get_fresh_token()}'
    return config

client.interceptors.request.use(oauth_interceptor)

# Proxy (explicit, or read from HTTP_PROXY/HTTPS_PROXY automatically)
client.get("/data", proxy={"host": "http://proxy.example.com:8080"})
client.get("/data", proxy={"host": "http://proxy:8080", "auth": {"username": "u", "password": "p"}})

# SOCKS proxy (pip install atomhttp[socks])
client.get("/data", proxy={"host": "socks5://127.0.0.1:1080"})

# TLS / mTLS
client.get("/secure", verify="/path/to/custom-ca.pem")
client.get("/secure", cert=("/path/to/cert.pem", "/path/to/key.pem"))

# Unix domain socket -- e.g. the Docker daemon
client.get("http://localhost/containers/json", socketPath="/var/run/docker.sock")

# Brotli (pip install atomhttp[brotli]) -- decoded transparently, no code changes

response = client.get(
    "/flaky-endpoint",
    retryConfig={
        "max_retries": 5,
        "backoff_factor": 0.3,
        "status_forcelist": [408, 429, 500, 502, 503, 504],
    },
    onRetry=lambda config, attempt, error: print(f"retry #{attempt}: {error}"),
)
# A server-sent "Retry-After" header is respected automatically,
# taking priority over the exponential backoff calculation.