asyncly 0.3.3

Simple HTTP client and server for your integrations based on aiohttp

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)