services.api_clients
class
BaseAPIClient:
7class BaseAPIClient: 8 """Base asynchronous HTTP client using httpx for GET and POST requests.""" 9 10 TIMEOUT: int = 5 # seconds 11 12 async def get( 13 self, url: str, params: Optional[dict | list[tuple[str, Any]]] = None 14 ) -> dict: 15 """ 16 Send an asynchronous GET request. 17 18 :param url: The target URL. 19 :param params: Optional query parameters as a dictionary or list of tuples. 20 :return: Parsed JSON response as a dictionary. 21 :raises httpx.HTTPStatusError: If the response contains an error status. 22 """ 23 async with httpx.AsyncClient(timeout=self.TIMEOUT) as client: 24 response = await client.get(url, params=params) 25 response.raise_for_status() 26 return response.json() 27 28 async def post(self, url: str, data: Optional[dict] = None) -> dict: 29 """ 30 Send an asynchronous POST request. 31 32 :param url: The target URL. 33 :param data: Optional payload as a dictionary. 34 :return: Parsed JSON response as a dictionary. 35 :raises httpx.HTTPStatusError: If the response contains an error status. 36 """ 37 async with httpx.AsyncClient(timeout=self.TIMEOUT) as client: 38 response = await client.post(url, json=data) 39 response.raise_for_status() 40 return response.json()
Base asynchronous HTTP client using httpx for GET and POST requests.
async def
get( self, url: str, params: Union[dict, list[tuple[str, Any]], NoneType] = None) -> dict:
12 async def get( 13 self, url: str, params: Optional[dict | list[tuple[str, Any]]] = None 14 ) -> dict: 15 """ 16 Send an asynchronous GET request. 17 18 :param url: The target URL. 19 :param params: Optional query parameters as a dictionary or list of tuples. 20 :return: Parsed JSON response as a dictionary. 21 :raises httpx.HTTPStatusError: If the response contains an error status. 22 """ 23 async with httpx.AsyncClient(timeout=self.TIMEOUT) as client: 24 response = await client.get(url, params=params) 25 response.raise_for_status() 26 return response.json()
Send an asynchronous GET request.
Parameters
- url: The target URL.
- params: Optional query parameters as a dictionary or list of tuples.
Returns
Parsed JSON response as a dictionary.
Raises
- httpx.HTTPStatusError: If the response contains an error status.
async def
post(self, url: str, data: Optional[dict] = None) -> dict:
28 async def post(self, url: str, data: Optional[dict] = None) -> dict: 29 """ 30 Send an asynchronous POST request. 31 32 :param url: The target URL. 33 :param data: Optional payload as a dictionary. 34 :return: Parsed JSON response as a dictionary. 35 :raises httpx.HTTPStatusError: If the response contains an error status. 36 """ 37 async with httpx.AsyncClient(timeout=self.TIMEOUT) as client: 38 response = await client.post(url, json=data) 39 response.raise_for_status() 40 return response.json()
Send an asynchronous POST request.
Parameters
- url: The target URL.
- data: Optional payload as a dictionary.
Returns
Parsed JSON response as a dictionary.
Raises
- httpx.HTTPStatusError: If the response contains an error status.
7class AgifyClient(BaseAPIClient): 8 """Client for interacting with the Agify API.""" 9 10 BASE_URL: str = "https://api.agify.io" 11 12 async def get_age(self, name: str, country: str) -> dict[str, Any]: 13 """ 14 Get age prediction for a single name. 15 16 :param name: Person's name. 17 :param country: Country code (ISO 3166-1 alpha-2). 18 :return: API response containing age prediction. 19 """ 20 params = {"name": name, "country_id": country} 21 return await self.get(self.BASE_URL, params=params) 22 23 async def get_batch_ages( 24 self, names: list[str], country: str 25 ) -> Union[list[dict[str, Any]], dict[str, Any]]: 26 """ 27 Get age predictions for a batch of names (max 10). 28 29 :param names: List of names (up to 10). 30 :param country: Country code. 31 :return: List of API responses for each name. 32 :raises ValueError: If more than 10 names are passed. 33 """ 34 if not names: 35 return [] 36 37 if len(names) > 10: 38 raise ValueError("Agify API supports up to 10 names per batch request.") 39 40 params = [("name[]", name) for name in names] 41 params.append(("country_id", country)) 42 43 return await self.get(self.BASE_URL, params=params)
Client for interacting with the Agify API.
async def
get_age(self, name: str, country: str) -> dict[str, typing.Any]:
12 async def get_age(self, name: str, country: str) -> dict[str, Any]: 13 """ 14 Get age prediction for a single name. 15 16 :param name: Person's name. 17 :param country: Country code (ISO 3166-1 alpha-2). 18 :return: API response containing age prediction. 19 """ 20 params = {"name": name, "country_id": country} 21 return await self.get(self.BASE_URL, params=params)
Get age prediction for a single name.
Parameters
- name: Person's name.
- country: Country code (ISO 3166-1 alpha-2).
Returns
API response containing age prediction.
async def
get_batch_ages( self, names: list[str], country: str) -> Union[list[dict[str, Any]], dict[str, Any]]:
23 async def get_batch_ages( 24 self, names: list[str], country: str 25 ) -> Union[list[dict[str, Any]], dict[str, Any]]: 26 """ 27 Get age predictions for a batch of names (max 10). 28 29 :param names: List of names (up to 10). 30 :param country: Country code. 31 :return: List of API responses for each name. 32 :raises ValueError: If more than 10 names are passed. 33 """ 34 if not names: 35 return [] 36 37 if len(names) > 10: 38 raise ValueError("Agify API supports up to 10 names per batch request.") 39 40 params = [("name[]", name) for name in names] 41 params.append(("country_id", country)) 42 43 return await self.get(self.BASE_URL, params=params)
Get age predictions for a batch of names (max 10).
Parameters
- names: List of names (up to 10).
- country: Country code.
Returns
List of API responses for each name.
Raises
- ValueError: If more than 10 names are passed.
7class JokeClient(BaseAPIClient): 8 """Client for retrieving random jokes.""" 9 10 BASE_URL: str = "https://official-joke-api.appspot.com/random_joke" 11 12 async def get_random_joke(self) -> dict[str, Any]: 13 """ 14 Retrieve a random joke. 15 16 :return: Dictionary containing joke data. 17 """ 18 return await self.get(self.BASE_URL)
Client for retrieving random jokes.
async def
get_random_joke(self) -> dict[str, typing.Any]:
12 async def get_random_joke(self) -> dict[str, Any]: 13 """ 14 Retrieve a random joke. 15 16 :return: Dictionary containing joke data. 17 """ 18 return await self.get(self.BASE_URL)
Retrieve a random joke.
Returns
Dictionary containing joke data.
7class PostmanClient(BaseAPIClient): 8 """Client for testing HTTP POST requests using Postman Echo.""" 9 10 BASE_URL: str = "https://postman-echo.com/post" 11 12 async def post_response(self, data: dict[str, Any]) -> dict[str, Any]: 13 """ 14 Send a POST request and get the echoed response. 15 16 :param data: Dictionary to send in the request body. 17 :return: Echoed response from Postman Echo. 18 """ 19 return await self.post(self.BASE_URL, data)
Client for testing HTTP POST requests using Postman Echo.
async def
post_response(self, data: dict[str, typing.Any]) -> dict[str, typing.Any]:
12 async def post_response(self, data: dict[str, Any]) -> dict[str, Any]: 13 """ 14 Send a POST request and get the echoed response. 15 16 :param data: Dictionary to send in the request body. 17 :return: Echoed response from Postman Echo. 18 """ 19 return await self.post(self.BASE_URL, data)
Send a POST request and get the echoed response.
Parameters
- data: Dictionary to send in the request body.
Returns
Echoed response from Postman Echo.