Asyncronous API calls on Python

In

Libraries required

Asyncio: CLI interpreter

  • Install

    1
    python -m pip install asyncio

  • Use

    1
    python -m asyncio

HTTPX: Async GET

  • Install

    1
    pyhton -m pip install httpx

  • Use (similar to requests)

    1
    2
    3
    4
    5
    6
    7
    8
    9
    """
    Make the HTTP call and return the response in a typed and verified mode
    """
    import httpx

    async with httpx.AsyncClient() as client:
      response = await client.get('https://ifconfig.co/json')
    print(response)
    # <Response [200 OK]>

Pydantic: response data validation

Validate so we can access the JSON body of this response, to build typing.
If the API returns something unexpected, we want to return a friendly exception, instead of an AttributeError.

  • Install

    1
    python -m pip install pydantic

  • Use

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    """
    Integrate typing: (r.json)
    - Check the url result exists, check the typing
    """
    from pydantic import BaseModel
    from ipaddress import IPv4Address

    class IfConfig(BaseModel):
      ip: IPv4Address
      ip_decimal: int

    ifconfig = IfConfig.parse_obj(r.json())
    print(ifconfig)
    # IfConfig(ip=IPv4Address('78.153.21.75'), ip_decimal=1807729179)

Creation of an API client

  • API client code

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    from ipaddress import IPv4Address

    from pydantic import BaseModel
    from httpx import AsyncClient

    IFCONFIG_URL = 'https://ifconfig.co/'

    class IfConfig(BaseModel):
      ip: IPv4Address
      ip_decimal: int

    # client should inherit from AsyncClient
    class IfConfigClient(AsyncClient):
      def __init__(self):
        super().__init__(base_url=IFCONFIG_URL)

      # async call
      async def get_ifconfig(self):
        request = await self.get('json')

        # validation
        try:
          ifconfig = IfConfig.parse_obj(request.json())
        except ValidationError:
          print('Something went wrong!')

        return ifconfig

  • API call

    1
    2
    3
    4
    5
    6
    7
    from ifconfig import IfConfigClient

    async with IfConfigClient() as client:
      ifconfig = await client.get_ifconfig()
      
    print(ifconfig)
    # IfConfig(ip=IPv4Address('78.153.21.75'), ip_decimal=1807729179)

API test with mocks

  • Use
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    from unittest.mock import patch
    import ifconfig

    mock_ip = IPv4Address('78.153.21.75')
    mock_ip_decimal = 1807729179
    mock_ifconfig = ifconfig.IfConfig(ip=mock_ip, ip_decimal=mock_ip_decimal)

    @patch.object(ifconfig.IfConfigClient, 'get_ifconfig', return_value=mock_ifconfig)
    def test_ifconfig_processing(get_ifconfig):
      assert get_ifconfig.assert_awaited_once()