services.api_clients

 1from .base_client import BaseAPIClient
 2from .agify_client import AgifyClient
 3from .joke_client import JokeClient
 4from .postman_client import PostmanClient
 5
 6__all__ = [
 7    "BaseAPIClient",
 8    "AgifyClient",
 9    "JokeClient",
10    "PostmanClient",
11]
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.

TIMEOUT: int = 5
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.
class AgifyClient(services.api_clients.BaseAPIClient):
 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.

BASE_URL: str = 'https://api.agify.io'
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.
class JokeClient(services.api_clients.BaseAPIClient):
 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.

BASE_URL: str = 'https://official-joke-api.appspot.com/random_joke'
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.

class PostmanClient(services.api_clients.BaseAPIClient):
 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.

BASE_URL: str = 'https://postman-echo.com/post'
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.