HTTP client

BaseHttpClient is a thin, typed base class for building API clients on top of an aiohttp ClientSession. You subclass it and add one method per endpoint.

Anatomy

from http import HTTPStatus
from types import MappingProxyType

from aiohttp import hdrs
from asyncly import BaseHttpClient, DEFAULT_TIMEOUT, ResponseHandlersType
from asyncly.client.handlers.pydantic import parse_model
from asyncly.client.timeout import TimeoutType


class CatfactClient(BaseHttpClient):
    FACT_HANDLERS: ResponseHandlersType = MappingProxyType(
        {HTTPStatus.OK: parse_model(CatFact)}
    )

    async def fetch_fact(self, timeout: TimeoutType = DEFAULT_TIMEOUT) -> CatFact:
        return await self._make_req(
            method=hdrs.METH_GET,
            url=self._url / "fact",
            handlers=self.FACT_HANDLERS,
            timeout=timeout,
        )

Key ideas:

• The session is injected — you create and own the ClientSession, the client just uses it. This keeps connection pooling, auth, and lifecycle in your control.
• self._url is the base URL as a yarl.URL; build endpoint URLs with the / operator.
• self._make_req(...) issues the request and dispatches the response to a handler chosen by status code.

Construction

from aiohttp import ClientSession

async with ClientSession() as session:
    client = CatfactClient(
        url="https://catfact.ninja",   # str or yarl.URL
        session=session,
        client_name="catfact",         # used in metrics and error messages
    )

The url property exposes the configured base URL:

client.url  # URL('https://catfact.ninja')

Timeouts

Every request method takes a timeout of type TimeoutType — an aiohttp.ClientTimeout, a datetime.timedelta, or a plain number of seconds. They are all normalized internally:

from datetime import timedelta

await client.fetch_fact(timeout=2)                    # 2 seconds
await client.fetch_fact(timeout=2.5)                  # 2.5 seconds
await client.fetch_fact(timeout=timedelta(seconds=2)) # timedelta

The default is aiohttp's DEFAULT_TIMEOUT, re-exported from the top-level asyncly package.

Retries

Retries are opt-in per endpoint. Pass an immutable RetryPolicy to _make_req:

from asyncly import RetryPolicy

RETRY_TRANSIENT = RetryPolicy(max_attempts=3)


async def fetch_fact(self) -> CatFact:
    return await self._make_req(
        method=hdrs.METH_GET,
        url=self._url / "fact",
        handlers=self.FACT_HANDLERS,
        retry=RETRY_TRANSIENT,
    )

The defaults retry transient statuses (408, 425, 429, 500, 502, 503, 504), connection/payload failures, and timeouts. Only idempotent methods are eligible. Retry-After supports both delta-seconds and HTTP dates; otherwise a capped exponential strategy with full jitter is used.

Each attempt gets the full per-request timeout. Cancellation is never retried, and a streaming/file-like data= body is treated as non-replayable. JSON, bytes/strings, and ordinary form mappings can be sent again safely. The last HTTP response still goes through the configured status handler, while the last transport exception is raised unchanged.

Use a callback when the application needs logical retry telemetry:

def observe_retry(event):
    logger.info(
        "retry %s attempt=%s delay=%s reason=%s",
        event.kind,
        event.context.attempt,
        event.delay,
        event.reason,
    )

await self._make_req(
    ...,
    retry=RetryPolicy(backoff=lambda context: 0.0),
    retry_observer=observe_retry,
)

The observer receives scheduled, suppressed, and exhausted events. A custom backoff callable makes tests deterministic and avoids real delays.

Proxy support

BaseHttpClient accepts proxy and proxy_auth, set once on the client or overridden per request. They are forwarded to aiohttp:

from aiohttp import BasicAuth

client = CatfactClient(
    url="https://catfact.ninja",
    session=session,
    client_name="catfact",
    proxy="http://127.0.0.1:8080",
    proxy_auth=BasicAuth("user", "secret"),
)

See Proxy testing for spinning up a mock proxy to verify a client really routes through one.

Extra request arguments

_make_req forwards any extra keyword arguments to ClientSession.request, so you can pass headers, params, json, data, etc.:

return await self._make_req(
    method=hdrs.METH_POST,
    url=self._url / "items",
    handlers=self.CREATE_HANDLERS,
    json={"name": name},
    headers={"X-Tenant": tenant},
)

Error handling

If a response status has no matching handler, _make_req raises UnhandledStatusException, which carries the status, url, and client_name. Map the statuses you expect — including error codes — to handlers. See Response handlers.

Metrics

To record request metrics (latency, status, errors), use InstrumentableHttpClient instead, which adds enable_metrics() / instrument() on top of the same API.