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

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, status, duration, and error type to the active MetricsSink.

Source code in asyncly/client/metrics/instrumentable_client.py

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

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

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
    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

def disable_metrics(self) -> None:
    """Stop emitting metrics (revert to the no-op sink)."""
    self._metrics_sink = NoopSink()
    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

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
            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

    return _Ctx()

Timeouts

asyncly.client.timeout.get_timeout

get_timeout(t: TimeoutType) -> ClientTimeout

Source code in asyncly/client/timeout.py

@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"
    ]
)