Asyncly

A tiny async HTTP client and a real aiohttp mock server for testing your integrations — built on aiohttp.

asyncly gives you two pieces that fit together:

• BaseHttpClient — a thin, typed base class for HTTP clients with per-status response handlers, policy-driven retries, flexible timeouts, and first-class proxy support. InstrumentableHttpClient adds pluggable metrics (Prometheus / OpenTelemetry): requests, in-flight, network phases, and connection-pool stats.
• srvmocker — spin up a real aiohttp test server (not a transport patch) to simulate upstreams and socket faults, assert what your client sent, and even route through a mock proxy.

📖 Read the full documentation →

Installation

pip install asyncly

Optional extras — msgspec, pydantic, orjson, prometheus, opentelemetry:

pip install "asyncly[pydantic]"

Quickstart

Define a client by subclassing BaseHttpClient and mapping status codes to handlers:

from http import HTTPStatus
from types import MappingProxyType

from aiohttp import ClientSession, hdrs
from pydantic import BaseModel

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


class CatFact(BaseModel):
    fact: str
    length: int


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

Test it against a real mock server — no network, no monkeypatching:

from asyncly.srvmocker import JsonResponse, MockRoute, start_service


async def test_fetch_fact() -> None:
    routes = [MockRoute("GET", "/fact", "fact")]
    async with start_service(routes) as service:
        service.register("fact", JsonResponse({"fact": "Cats sleep a lot.", "length": 17}))

        async with ClientSession() as session:
            client = CatfactClient(url=service.url, session=session, client_name="catfact")
            fact = await client.fetch_fact()

        assert fact.fact == "Cats sleep a lot."
        service.assert_called("fact", times=1)

Prefer fixtures over boilerplate? asyncly ships a pytest plugin with mock_service and mock_proxy. See the Quickstart for more.

Why asyncly?

Unlike transport-patching mocks (aioresponses, respx), srvmocker runs a real aiohttp.TestServer inside your test loop — catching real sockets, timeouts, header auto-injection, and serialization quirks. See Testing strategies for the full comparison.

License

MIT