Simple HTTP client and server for your integrations based on aiohttp
Project description
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)
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.