Refactoring try blocks with async/await

Question

I have two functions which I would like to simplify. They have largely the same logic, but one is asynchronous and uses one await in a try block. I'd like to have the logic in one place, but could not figure out a clean way of doing that. Any suggestions?

from requests import get, HTTPError

# Dummy functions for external functionality
def sync_send(url):
    return get(url)

async def async_send(url):
    return get(url)


def next(url):
    if url is None:
        return

    try:
        response = sync_send(url)
        return response.json()
    except HTTPError as e:
        if e.response.status_code == 404:
            return
        else:
            raise

async def async_next(url):
    if url is None:
        return

    try:
        response = await async_send(url)
        return response.json()
    except HTTPError as e:
        if e.response.status_code == 404:
            return
        else:
            raise

So the point of this all is to provide a way of requesting for resources in an API two ways, sync and async. Despite the simple (synchronous) dummy functions, in async_next we use asynchronous IO to retrieve things. Other than that, the situation is exactly as it is here.

---

I have accepted that the if url is None cannot be refactored, and I currently think my only option is to make a more complex exception hierarchy, raising a dedicated 404 error to catch it without logic. Any other ideas?

Answer 1

Let's have async implementation:

async def async_next(url):
    if url is None:
        return

    try:
        response = await async_send(url)
        return response.json()
    except HTTPError as e:
        if e.response.status_code == 404:
            return
        else:
            raise

Then you can provide following sync bridge:

def next(url):
    return asyncio.run(async_next(url))

Or the following if the event loop is available and running:

def next(url):
    return loop.run_until_complete(async_next(url))

Note: check asyncio high level commands for more details