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?

• Mast: What happened with passing the cond variable?
• Felix: @Mast When asking this question I thought without further context it would be weird to assume that someone would pass in an empty URL. But when I thought about it again, it is better to have it as it is in the code.

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

• Felix: Thanks for the suggestion, though I'm not a fan of running the loop every time a synchronous call should be made.