aiohttp Client¶
Now we have a REST server, let’s write REST client to it.
The idea is: create a Client
class with .list()
, .get()
, .create()
etc. methods to operate on blog posts collection.
Data structures¶
We need a Post
dataclass to provide post related fields (and avoid dictionaries in
our API):
from dataclasses import dataclass
@dataclass(frozen=True)
class Post:
id: int
owner: str
editor: str
title: str
text: Optional[str]
def pprint(self) -> None:
print(f"Post {self.id}")
...
Client class¶
Client
is a class with embedded aiohttp.ClientSession
. Also, we need the REST
server URL to connect and user name to provide creator / last-editor information:
class Client:
def __init__(self, base_url: URL, user: str) -> None:
self._base_url = base_url
self._user = user
self._client = aiohttp.ClientSession(raise_for_status=True)
To properly close Client
instance add .close()
method:
async def close(self) -> None:
return await self._client.close()
Now is a time for implementing a method to access the REST server, e.g. .list()
:
async def list(self) -> List[Post]:
async with self._client.get(self._make_url("api")) as resp:
ret = await resp.json()
return [Post(text=None, **item) for item in ret["data"]]
It makes GET {base_url}/api
request, read JSON response and returns a list of
Post
objects.
_make_url
is a helper for prepending base url to API endpoints. For the tutorial
it is simple but in real life you often need to do more work, e.g. provide Authorization
HTTP header, sign your request etc.:
def _make_url(self, path: str) -> URL:
return self._base_url / path
Client Usage¶
The usage is simple:
async def fetch():
client = Client("http://localhost:8080", "John")
try:
posts = await client.list()
for post in posts:
post.pprint()
finally:
await client.close()
try
/finally
is not very convinient, many people prefer async with
statement. It saves 3 lines and (more important) avoids silly errors like instantiating
a client variable inside try
/finally
block.
async def fetch():
async with Client("http://localhost:8080", "John") as client:
posts = await client.list()
for post in posts:
post.pprint()
To support this form we need to implement __aenter__
/ __aexit__
async
Client
methods:
async def __aenter__(self) -> "Client":
return self
async def __aexit__(
self,
exc_type: Optional[Type[BaseException]],
exc_val: Optional[BaseException],
exc_tb: Optional[TracebackType],
) -> Optional[bool]:
await self.close()
Full Client Example¶
In full example we provide a Command Line Tool to work with REST API server by using famous Click library.
The Click usage is out of scope of the tutorial itself, but you can learn the full example on your own: Full REST client example .