Reference

Complete API reference, comparisons, migration notes, and practical examples

Comparison with Other Libraries

AtomHTTP combines what requests and httpx each do well -- sync-first ergonomics, optional async, and modern features like cancellation and streaming -- in one client built on urllib3.

FeatureAtomHTTPrequestshttpx
Sync API (no event loop required)✓✓
Async API, same transport as sync✓ (separate client)
Minimal runtime dependencies✓ (urllib3 only)✓✓ (httpcore)
Cancellation (AbortController)Manual task.cancel()
Persistent thread pool for concurrency✓ (.all/.submit/.map)Manual ThreadPoolExecutorn/a (asyncio.gather)
Streaming multipart uploads (constant memory)✓ (automatic)Manual generatorManual generator
Upload/download progress callbacks
Pagination helper✓ (.paginate())
ETag/Cache-Control caching✓ (CacheInterceptor)Needs requests-cache
Request/response interceptors✓ (event hooks)
Retry with backoff + Retry-After✓ (built-in)Needs urllib3 Retry manuallyNeeds external lib
Unix domain socket supportNeeds requests-unixsocket✓
SOCKS proxy support✓ (extra)✓ (extra)✓ (extra)
Mock adapter for testing✓ (MockAdapter)Needs responses/requests-mock✓ (MockTransport)

Side-by-side code

Basic Request

A single request, the most common case for every library.

1
2
3
4
5
from atomhttp import AtomHTTP

client = AtomHTTP(base_url='https://jsonplaceholder.typicode.com')
response = client.get('/posts/1')   # no await needed
print(response.status, response.data['title'])

Concurrent Requests

Running several requests at once.

1
2
3
4
5
6
7
8
9
10
11
12
from atomhttp import AtomHTTP

client = AtomHTTP(base_url='https://jsonplaceholder.typicode.com', max_workers=10)

responses = client.all([
    lambda: client.get('/posts/1'),
    lambda: client.get('/posts/2'),
    lambda: client.get('/posts/3'),
])

for resp in responses:
    print(f"Post {resp.data['id']}: {resp.data['title'][:30]}...")

Upload Progress Tracking

Tracking upload progress with a callback.

1
2
3
4
5
6
7
8
9
from atomhttp import AtomHTTP

def on_upload(loaded, total):
    print(f"Upload: {loaded}/{total} bytes")

client = AtomHTTP(base_url='https://httpbin.org')
with open('test.txt', 'rb') as f:
    resp = client.post('/post', data=f.read(), onUploadProgress=on_upload)
print("Status:", resp.status)

Request Cancellation

Aborting an in-flight request from elsewhere in the program.

1
2
3
4
5
6
7
8
9
10
11
from atomhttp import AtomHTTP, AbortController
from atomhttp.errors import AtomHTTPCancelError

client = AtomHTTP(base_url='https://httpbin.org', timeout=30)
controller = AbortController()

# controller.abort() from another thread cancels this immediately:
try:
    client.get('/delay/10', signal=controller.signal)
except AtomHTTPCancelError:
    print("cancelled")

Status Validation

Rejecting non-2xx responses automatically.

1
2
3
4
5
6
7
8
9
from atomhttp import AtomHTTP
from atomhttp.errors import AtomHTTPRequestError

client = AtomHTTP(base_url='https://httpbin.org')

try:
    resp = client.get('/status/404', validateStatus=lambda status: status < 400)
except AtomHTTPRequestError as e:
    print(f"Request failed with status {e.response.status}")

Complete Examples

Two full, realistic examples combining several features.

Authenticated API client with retries and caching:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from atomhttp import AtomHTTP
from atomhttp.cache import CacheInterceptor

client = AtomHTTP(
    base_url="https://api.example.com",
    timeout=10,
    headers={"Authorization": "Bearer YOUR_TOKEN"},
    retryConfig={"max_retries": 3, "status_forcelist": [500, 502, 503, 504]},
)

cache = CacheInterceptor()
client.interceptors.request.use(cache.on_request)
client.interceptors.response.use(cache.on_response)

for items in client.paginate("/users"):
    for user in items:
        print(user["name"])

client.close()

Concurrent file downloads with a shared thread pool:

1
2
3
4
5
6
7
8
9
10
11
12
from atomhttp import AtomHTTP

client = AtomHTTP(base_url="https://files.example.com", max_workers=8)

urls = ["/a.zip", "/b.zip", "/c.zip"]
futures = [client.submit("GET", "/download" + u) for u in urls]

for url, future in zip(urls, futures):
    response = future.result()
    with open(url.lstrip("/"), "wb") as f:
        f.write(response.data)
    print(f"saved {url}")

API Methods Reference

Every method exists identically on AsyncAtomHTTP, just awaited.

MethodDescriptionExample
client.get(url, **kwargs)HTTP GET requestclient.get('/users')
client.post(url, data, **kwargs)HTTP POST requestclient.post('/users', data={...})
client.put(url, data, **kwargs)HTTP PUT requestclient.put('/users/1', data={...})
client.patch(url, data, **kwargs)HTTP PATCH requestclient.patch('/users/1', data={...})
client.delete(url, **kwargs)HTTP DELETE requestclient.delete('/users/1')
client.request(method, url, **kwargs)Generic request methodclient.request('GET', '/users')
client.stream(method, url, **kwargs)Streamed response, read incrementallywith client.stream('GET', '/f') as r: ...
client.download(url, path, **kwargs)Download straight to diskclient.download('/f.zip', 'f.zip')
client.paginate(url, **kwargs)Walk a paginated endpointfor items in client.paginate('/users'): ...
client.all(calls, max_workers=None)Run request thunks concurrentlyclient.all([lambda: client.get('/a')])
client.submit(method, url, **kwargs)Fire-and-forget on the thread poolfuture = client.submit('GET', '/a')
client.map(method, urls, **kwargs)Same request, many URLs, concurrentlyclient.map('GET', ['/a', '/b'])
client.close()Release pooled connections and the thread poolclient.close()
client.as_async() / async_client.as_sync()Convert between sync/async, sharing stateclient.as_async()

RequestConfig Fields

Every field below can be set on the client (as a default) or per-request (overriding the default).

FieldType / DefaultDescription
base_urlstr = ""Prefix for relative URLs
timeoutint/float/timedelta = 30Request timeout in seconds
headersdict = {}Default/per-request headers
paramsdict = {}Query string parameters
dataAny = NoneRequest body: dict/list, FormData, str, or bytes
cookiesbool = TrueEnable the client's persistent cookie jar
max_workersint = 10Thread pool size for .all()/.submit()/.map()
maxRedirectsint = 5Max redirects to follow (0 disables)
maxContentLength / maxBodyLengthint = -1Response/request size caps in bytes (-1 = unlimited)
responseTypestr = "json"json | text | blob | arraybuffer | stream
validateStatusCallable | Nonefn(status) -> bool; raises AtomHTTPRequestError on False
authdict | None{"username": ..., "password": ...} for Basic Auth
proxydict | None{'host': 'http://...'} or socks5://... ; falls back to env vars
verifybool | str = TrueTLS verification on/off, or a custom CA bundle path
certstr | tuple | NonemTLS client certificate
retryConfigdict | Nonemax_retries, backoff_factor, status_forcelist
signalAbortSignal | NoneAbortController().signal for cancellation
socketPathstr | NoneUnix domain socket path
onUploadProgress / onDownloadProgressCallable | Nonefn(loaded, total)
onRequestStart / onRetry / onRedirectCallable | NoneLightweight observability hooks
adapterBaseAdapter | NonePer-request adapter override, e.g. MockAdapter

Error Codes Reference

AtomHTTP provides standardized error codes for programmatic error handling. By default no exception is raised for 4xx/5xx -- opt in with validateStatus or raise_for_status().

Error CodeDescriptionException Type
ERR_BAD_{status}Bad request (4xx) or rejected by validateStatusAtomHTTPRequestError
ERR_NETWORKDNS failure, connection refused, or other transport errorAtomHTTPNetworkError
ECONNABORTEDRequest exceeded its timeoutAtomHTTPTimeoutError
ERR_CANCELEDRequest was aborted via AbortControllerAtomHTTPCancelError

Response Types

AtomHTTP supports multiple response types for different use cases.

json

Parses as dict/list. Default option -- falls back to raw text if the body isn't valid JSON.

text

Returns the body as a decoded str. Good for HTML, CSV, plain text.

blob / arraybuffer

Returns the raw body as bytes. For images, PDFs, ZIP files.

stream

Raw urllib3.HTTPResponse for manual reading -- prefer client.stream() instead for the friendlier API.

Migrating from v1

v2 is a full rewrite: sync-first by default, minimal runtime dependencies, built on urllib3. The biggest change is that AtomHTTP is no longer async -- if you want async, use the new AsyncAtomHTTP class instead.

Making a request

Before (v1)

1
2
3
4
# v1
client = AtomHTTP({'baseURL': 'https://api.example.com'})
response = await client.get('/users/1')
await client.close()

After (v2.1)

1
2
3
4
# v2.1
client = AtomHTTP(base_url='https://api.example.com')
response = client.get('/users/1')   # no await needed
client.close()

Concurrent requests

Before (v1)

1
2
3
4
# v1
responses = await AtomHTTP.all([
    client.get('/a'), client.get('/b'),
])

After (v2.1)

1
2
3
4
5
6
# v2.1
responses = client.all([
    lambda: client.get('/a'),
    lambda: client.get('/b'),
])
# or, if you still want async: await async_client.all([async_client.get('/a'), ...])

Async, if you still want it

Before (v1)

1
2
3
# v1 -- async was the only option
client = AtomHTTP({'baseURL': '...'})
response = await client.get('/users/1')

After (v2.1)

1
2
3
# v2.1 -- async is optional, via a separate class
async with AsyncAtomHTTP(base_url='...') as client:
    response = await client.get('/users/1')

Progress callback naming

Before (v1)

1
2
# v1
await client.post('/upload', data=f, on_upload_progress=cb)

After (v2.1)

1
2
# v2.1 -- camelCase, matching axios-style config
client.post('/upload', data=f, onUploadProgress=cb)

Other breaking changes: the constructor now takes plain keyword arguments (AtomHTTP(base_url=..., timeout=...)) instead of a single config dict; several internal v1 modules that were dead code (unused duplicate adapters, an unused cookie manager, an unused redirect handler) were removed entirely rather than ported forward.