Asyncly

Simple HTTP client and server for your integrations based on aiohttp.

Installation

Installation is possible in standard ways, such as PyPI or installation from a git repository directly.

Installing from PyPI:

pip install asyncly

Installing from github.com:

pip install git+https://github.com/andy-takker/asyncly

The package contains several extras and you can install additional dependencies if you specify them in this way.

For example, with msgspec:

pip install "asyncly[msgspec]"

Complete table of extras below:

example	description
pip install "asyncly[msgspec]"	For using msgspec structs
pip install "asyncly[orjson]"	For fast parsing json by orjson
pip install "asyncly[pydantic]"	For using pydantic models

Quick start guide

HttpClient

Simple HTTP Client for https://catfact.ninja. See full example in examples/catfact_client.py

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


class CatfactClient(BaseHttpClient):
    RANDOM_CATFACT_HANDLERS: ResponseHandlersType = MappingProxyType(
         {
              HTTPStatus.OK: parse_model(CatfactSchema),
         }
    )

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

Test Async Server for client

For the HTTP client, we create a server to which he will go and simulate real responses. You can dynamically change the responses from the server in a specific test.

Let's prepare the fixtures:

@pytest.fixture
async def catafact_service() -> AsyncIterator[MockService]:
    routes = [
        MockRoute("GET", "/fact", "random_catfact"),
    ]
    async with start_service(routes) as service:
        service.register(
            "random_catfact",
            JsonResponse({"fact": "test", "length": 4}),
        )
        yield service


@pytest.fixture
def catfact_url(catafact_service: MockService) -> URL:
    return catafact_service.url


@pytest.fixture
async def catfact_client(catfact_url: URL) -> AsyncIterator[CatfactClient]:
    async with ClientSession() as session:
        client = CatfactClient(
            client_name="catfact",
            session=session,
            url=catfact_url,
        )
        yield client

Now we can use them in tests. See full example in examples/test_catfact_client.py

async def test_fetch_random_catfact(catfact_client: CatfactClient) -> None:
    # use default registered handler
    fact = await catfact_client.fetch_random_cat_fact()
    assert fact == CatfactSchema(fact="test", length=4)


async def test_fetch_random_catfact_timeout(
    catfact_client: CatfactClient,
    catafact_service: MockService,
) -> None:
    # change default registered handler to time error handler
    catafact_service.register(
        "random_catfact",
        LatencyResponse(
            wrapped=JsonResponse({"fact": "test", "length": 4}),
            latency=1.5,
        ),
    )
    with pytest.raises(asyncio.TimeoutError):
        await catfact_client.fetch_random_cat_fact(timeout=1)