How do I make async HTTP requests using httpx (VS requests) in Python?

Question

I'm new to httpx and async and I'm having trouble converting a function that calls to an external API using requests into an asynchronous function using httpx.

This is the original function:

import requests


def get_api_data(url: str) -> Optional[List[str]]:
    """Get the data from an API url.
    Args:
        url (str):  The API url.
    Returns:
        list: The data.
    """
    try:
        resp = requests.get(url)
        return resp.json()
    except requests.exceptions.HTTPError as errh:
        print("Http Error:", errh)
    except requests.exceptions.ConnectionError as errc:
        print("Error Connecting:", errc)
    except requests.exceptions.Timeout as errt:
        print("Timeout Error:", errt)
    except requests.exceptions.RequestException as err:
        print("OOps: Something Else", err)

and this is what I've tried:

import httpx


async def get_api_data(url: str) -> Optional[List[Any]]:
    """Get the data from an API url.
    Args:
        url (str):  The API url.
    Returns:
        list: The data.
    """
    try:
        async with httpx.AsyncClient() as client:
            resp = await client.get(url)
            return resp.json()
    except httpx.HTTPError as errh:
        print("Http Error:", errh)

But I'm not sure it's a valid code and how to test it...

An example of a test before changing the function (the requests version); It uses patch to mock:

import pytest
import requests

from unittest.mock import patch


def test_api_http_error():
    with patch('app.internal.world_clock.requests.get', side_effect=requests.exceptions.HTTPError):
        assert not get_api_data(TIMEZONES_BASE_URL)

I spent a lot of hours trying to figure this out, so if you have any suggestions for how to make this conversion properly and how to test it - I would greatly appreciate it.

Answer 1

Your function looks fine. Are you running your async function? I had that issue when I first tried using it to replace requests in my project. Instead of just calling it like: results = async_funtion(parameter) It needed to be: results = asyncio.run(async_funtion(parameter)) In the first example you are just creating a coroutine object, the second actually runs it.