Implementing of own asynchronous clientΒΆ
Requires httpx package to be installed.
1"""
2Here we gonna create our own client
3with one of web lib
4that we, may be, want to use
5instead of ready to use clients.
6
7Used `httpx` package for this goal.
8"""
9
10import asyncio
11import logging
12from typing import (
13 Any,
14 Optional
15)
16
17import httpx
18
19
20from freedictionaryapi.clients.base_async_client import BaseAsyncDictionaryApiClient
21# or `from freedictionaryapi.clients import BaseAsyncDictionaryApiClient`
22from freedictionaryapi.errors import DictionaryApiError
23# or `from freedictionaryapi impot DictionaryApiError`
24from freedictionaryapi.languages import (
25 LanguageCodes,
26 DEFAULT_LANGUAGE_CODE
27)
28
29logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
30logger = logging.getLogger(__name__)
31
32
33# So, lets implement:
34# (simple class with requests but without __init__ changing check example for ``sync`` client)
35# 1. class with httpx.AsyncClient
36
37
38class NotSoSimpleOwnDictionaryApiClient(BaseAsyncDictionaryApiClient):
39
40 def __init__(self, default_language_code: LanguageCodes = DEFAULT_LANGUAGE_CODE, *,
41 client: Optional[httpx.AsyncClient] = None) -> None:
42 super().__init__(default_language_code)
43
44 if client is None:
45 self._client = httpx.AsyncClient()
46 else:
47 self._session = client
48
49 if not isinstance(self._client, httpx.AsyncClient):
50 raise TypeError('I`m expecting to get client with `httpx.AsyncClient` type...')
51
52 async def __aenter__(self) -> 'NotSoSimpleOwnDictionaryApiClient':
53 return self
54
55 async def __aexit__(self, exc_type, exc_val, exc_tb):
56 await self.close()
57
58 async def fetch_api_response(self, url: str) -> tuple[int, Any]:
59 # implement abstract method
60 # from docs we see
61 # '''
62 # ...
63 # The most important part is returning - method must return tuple of:
64 # 1. integer code of the API response;
65 # 2. python object loaded from API response with JSON decoding.
66 # ...
67 # '''
68
69 response = await self._client.get(url)
70 response_status_code = response.status_code
71 json_response = response.json()
72
73 return (response_status_code, json_response)
74
75 async def close(self) -> None:
76 await self._client.aclose()
77
78 logger.info('Client has been closed')
79
80
81async def main():
82 # lets test our client
83
84 # # not so simple one
85 async with NotSoSimpleOwnDictionaryApiClient() as not_so_simple_client:
86 word = ' program '
87 print('{:*^20}'.format(word))
88 try:
89 parser = await not_so_simple_client.fetch_parser(word)
90 except DictionaryApiError:
91 logger.error('API error')
92 else:
93 print(f'Definitions: {parser.get_all_definitions()!r}')
94
95
96if __name__ == '__main__':
97 loop = asyncio.get_event_loop()
98 loop.run_until_complete(main())