Skip to content

Client

asyncly.client.base.BaseHttpClient 

BaseHttpClient(
    url: URL | str,
    session: ClientSession,
    client_name: str,
    *,
    proxy: URL | str | None = None,
    proxy_auth: BasicAuth | None = None,
)

Typed base class for building async HTTP API clients.

Subclass it and add one method per endpoint, delegating to _make_req with a mapping of status codes to response handlers. The aiohttp.ClientSession is injected, so connection pooling and lifecycle stay under your control.

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

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

Initialize the client.

Parameters:

Name Type Description Default
url URL | str

Base URL the client's endpoints are resolved against.

 required
session ClientSession

The aiohttp.ClientSession to issue requests with. The caller owns its lifecycle.

 required
client_name str

Identifier used in metrics labels and error messages.

 required
proxy URL | str | None

Default proxy URL for every request. Can be overridden per request by passing proxy= to _make_req.

 None
proxy_auth BasicAuth | None

Default BasicAuth credentials for the proxy.

 None
Source code in asyncly/client/base.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def __init__(
    self,
    url: URL | str,
    session: ClientSession,
    client_name: str,
    *,
    proxy: URL | str | None = None,
    proxy_auth: BasicAuth | None = None,
) -> None:
    """Initialize the client.

    Args:
        url: Base URL the client's endpoints are resolved against.
        session: The `aiohttp.ClientSession` to issue requests with. The
            caller owns its lifecycle.
        client_name: Identifier used in metrics labels and error messages.
        proxy: Default proxy URL for every request. Can be overridden
            per request by passing `proxy=` to `_make_req`.
        proxy_auth: Default `BasicAuth` credentials for the proxy.
    """
    self._url = url if isinstance(url, URL) else URL(url)
    self._session = session
    self._client_name = client_name
    self._proxy = URL(proxy) if isinstance(proxy, str) else proxy
    self._proxy_auth = proxy_auth

url property 

url: URL

The base URL the client was configured with.

asyncly.client.metrics.instrumentable_client.InstrumentableHttpClient 

InstrumentableHttpClient(
    url: URL | str,
    session: ClientSession,
    client_name: str,
    *,
    proxy: URL | str | None = None,
    proxy_auth: BasicAuth | None = None,
)

Bases: BaseHttpClient

BaseHttpClient that records request metrics through a pluggable sink.

Behaves exactly like BaseHttpClient until a sink is enabled. Each completed request reports its client name, method, resolved route, logical operation, status, outcome, duration, and error type to the active MetricsSink.

Source code in asyncly/client/metrics/instrumentable_client.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init__(
    self,
    url: URL | str,
    session: ClientSession,
    client_name: str,
    *,
    proxy: URL | str | None = None,
    proxy_auth: BasicAuth | None = None,
) -> None:
    super().__init__(
        url=url,
        session=session,
        client_name=client_name,
        proxy=proxy,
        proxy_auth=proxy_auth,
    )
    self._metrics_sink: MetricsSink = NoopSink()
    self._resolve_route: RouteResolver = default_route_resolver
    self._sink_has_lifecycle: bool = False
    self._sink_has_phases: bool = False

enable_metrics 

enable_metrics(
    sink: MetricsSink,
    *,
    route_resolver: RouteResolver | None = None,
) -> None

Start emitting metrics to sink.

Parameters:

Name Type Description Default
sink MetricsSink

The metrics sink to report each request to.

 required
route_resolver RouteResolver | None

Optional override for how request URLs are normalized into low-cardinality route labels.

 None
Source code in asyncly/client/metrics/instrumentable_client.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def enable_metrics(
    self, sink: MetricsSink, *, route_resolver: RouteResolver | None = None
) -> None:
    """Start emitting metrics to ``sink``.

    Args:
        sink: The metrics sink to report each request to.
        route_resolver: Optional override for how request URLs are
            normalized into low-cardinality route labels.
    """
    self._metrics_sink = sink
    self._sink_has_lifecycle = _detect_lifecycle(sink)
    self._sink_has_phases = hasattr(sink, "observe_phase")
    if route_resolver is not None:
        self._resolve_route = route_resolver

disable_metrics 

disable_metrics() -> None

Stop emitting metrics (revert to the no-op sink).

Source code in asyncly/client/metrics/instrumentable_client.py 
86
87
88
89
90
91
def disable_metrics(self) -> None:
    """Stop emitting metrics (revert to the no-op sink)."""
    self._metrics_sink = NoopSink()
    self._sink_has_lifecycle = False
    self._sink_has_phases = False
    self._resolve_route = default_route_resolver

instrument 

instrument(
    sink: MetricsSink,
    *,
    route_resolver: RouteResolver | None = None,
)

Context manager that enables sink for the duration of a block.

Restores the previous sink and route resolver on exit.

Source code in asyncly/client/metrics/instrumentable_client.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def instrument(  # type: ignore[no-untyped-def]
    self, sink: MetricsSink, *, route_resolver: RouteResolver | None = None
):
    """Context manager that enables ``sink`` for the duration of a block.

    Restores the previous sink and route resolver on exit.
    """
    client = self

    class _Ctx:
        def __enter__(self) -> "InstrumentableHttpClient":
            self._prev_sink = client._metrics_sink
            self._prev_resolver = client._resolve_route
            self._prev_has_lifecycle = client._sink_has_lifecycle
            self._prev_has_phases = client._sink_has_phases
            client.enable_metrics(sink, route_resolver=route_resolver)
            return client

        def __exit__(
            self,
            exc_type: type[BaseException] | None,
            exc: BaseException | None,
            tb: TracebackType | None,
        ) -> None:
            client._metrics_sink = self._prev_sink
            client._resolve_route = self._prev_resolver
            client._sink_has_lifecycle = self._prev_has_lifecycle
            client._sink_has_phases = self._prev_has_phases

    return _Ctx()

Retries

asyncly.client.retry.RetryPolicy dataclass 

RetryPolicy(
    max_attempts: int = 3,
    statuses: Collection[int] = (
        lambda: DEFAULT_RETRY_STATUSES
    )(),
    exceptions: tuple[
        type[BaseException], ...
    ] = DEFAULT_RETRY_EXCEPTIONS,
    methods: Collection[str] = (
        lambda: IDEMPOTENT_METHODS
    )(),
    backoff: BackoffStrategy = full_jitter_backoff,
    respect_retry_after: bool = True,
)

Policy controlling whether and when an HTTP request is retried.

should_retry 

should_retry(context: RetryContext) -> bool

Return whether context is eligible for another attempt.

Source code in asyncly/client/retry.py 
126
127
128
129
130
131
132
133
134
def should_retry(self, context: RetryContext) -> bool:
    """Return whether ``context`` is eligible for another attempt."""

    return (
        self._matches_outcome(context)
        and context.method in self.methods
        and context.replayable
        and context.attempt < self.max_attempts
    )

get_delay 

get_delay(
    context: RetryContext,
    *,
    retry_after: str | None = None,
    now: datetime | None = None,
) -> float

Return the delay before the next attempt.

A valid HTTP Retry-After value takes precedence when enabled; otherwise the configured backoff strategy is used.

Source code in asyncly/client/retry.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def get_delay(
    self,
    context: RetryContext,
    *,
    retry_after: str | None = None,
    now: datetime | None = None,
) -> float:
    """Return the delay before the next attempt.

    A valid HTTP ``Retry-After`` value takes precedence when enabled;
    otherwise the configured backoff strategy is used.
    """

    delay = None
    if self.respect_retry_after and retry_after is not None:
        delay = _parse_retry_after(retry_after, now=now)
    if delay is None:
        delay = float(self.backoff(context))
    if not isfinite(delay) or delay < 0:
        raise ValueError("retry delay must be a finite non-negative number")
    return delay

asyncly.client.retry.RetryContext dataclass 

RetryContext(
    method: str,
    url: URL,
    attempt: int,
    max_attempts: int,
    replayable: bool = True,
    response_status: int | None = None,
    exception: BaseException | None = None,
)

Immutable description of one physical request attempt.

attempt is one-based and describes the attempt that just produced response_status or exception.

asyncly.client.retry.RetryEvent dataclass 

RetryEvent(
    kind: RetryEventKind,
    context: RetryContext,
    delay: float = 0.0,
    reason: RetryReason = "status",
)

Notification emitted for a retry decision.

asyncly.client.retry.full_jitter_backoff 

full_jitter_backoff(context: RetryContext) -> float

Return capped exponential backoff with full jitter.

The cap starts at 0.5 seconds after the first failed attempt and doubles to a maximum of 30 seconds. Pass a deterministic callable to RetryPolicy in tests when real jitter is undesirable.

Source code in asyncly/client/retry.py 
78
79
80
81
82
83
84
85
86
87
def full_jitter_backoff(context: RetryContext) -> float:
    """Return capped exponential backoff with full jitter.

    The cap starts at 0.5 seconds after the first failed attempt and doubles to
    a maximum of 30 seconds. Pass a deterministic callable to ``RetryPolicy``
    in tests when real jitter is undesirable.
    """

    cap = min(0.5 * (2 ** (context.attempt - 1)), 30.0)
    return random.uniform(0.0, cap)

Timeouts

asyncly.client.timeout.get_timeout 

get_timeout(t: TimeoutType) -> ClientTimeout
Source code in asyncly/client/timeout.py 
 9
10
11
@singledispatch
def get_timeout(t: TimeoutType) -> ClientTimeout:
    raise TypeError(f"Unknown type {type(t)}")

asyncly.client.timeout.TimeoutType module-attribute 

TimeoutType = ClientTimeout | timedelta | int | float

Typing

asyncly.client.typing.ResponseHandlersType module-attribute 

ResponseHandlersType = Mapping[
    HTTPStatus | int | str, ResponseHandler
]

asyncly.client.typing.MethodType module-attribute 

MethodType = (
    HTTPMethod
    | Literal[
        "GET",
        "POST",
        "PUT",
        "DELETE",
        "PATCH",
        "HEAD",
        "OPTIONS",
        "TRACE",
        "CONNECT",
    ]
)