Scenario: I often need to write Python functions like:

1. take some parameters and format them
2. call an API with the formatted parameters
3. parse the result and return chosen values

There's a huge problem in step #2.

In today's Python world, troubles arise because async/await are "infectious", In practice this function is splitted - like in Python stdlib, where a vanilla method and its async counterpart amethod often come in pairs. Package authors scramble to provide sync transport and another async transport. I discovered this ugly fact while reading the source code ofredis-py, httpx and elasticsearch-py. Duplicate and lookalike code was always written twice. All it takes is some random async IOs in one place and your code would be forced to change forever.

Is there a way to write the function in one place, but callable both with async and without?

I pondered this question for ages, and today I stumbled upon something interesting:

  def s1():
    return asyncio.sleep(1)

  async def s2():
    return await async.sleep(1)

There's virtually no difference when calling await s1() and await s2()

I vaguely remembered how Python’s coroutines were designed, and after some tinkering, I came up with this snippet:

import asyncio, types, functools

def aa(f):
    """
    make a function both awaitable and sync
    idk how to property name this. anti-asyncio (aa) maybe?
    """
    @functools.wraps(f)
    def wrapper(func, *args, **kwargs):
        if asyncio.iscoroutinefunction(func):
            return types.coroutine(f)(func, *args, **kwargs)
        else:
            async def afunc(*a, **kw):
                return func(*a, **kw)
            g = types.coroutine(f)(afunc, *args, **kwargs)
            try:
                while True: next(g)
            except StopIteration as ex:
                return ex.value
    return wrapper


@aa
def my_func(func, *args, **kwargs):
    # prepare args, kwargs here
    # add a prefix `yield from` everywhere, for either sync/async
    result = yield from func(*args, **kwargs)
    # handle the result here
    return result


import httpx

# async
async def main():
    # the same as `await httpx.AsyncClient(timeout=3).get('https://est.im')`
    print(await my_func(httpx.AsyncClient(timeout=3).get, 'https://est.im/'))
asyncio.run(main())


# sync
print(my_func(httpx.get, 'https://est.im'))
# works the same as httpx.get('https://est.im')

The above shows a single function called my_func, dependency injection of an HTTP get call of either sync/async, allows for customizable pre- and post-processing logic, and returns the result with clean syntax.

The only mental tax: inside my_func, you have to replace all await keyword with `yield from.

Update 2025-05-16: The only mental tax: add a yield from prefix for every funccalls for IO or API, either sync or async.

It solves all problems for my scenario and I’ve yet to find a simpler solution. If you have a good name for the @aa decorator please comment!

A sidenote, I am not sure if this method affects async schedulers and blocks something maybe? Like the while True might be a new kind of GIL. Also i haven't looked at problems with contextvars.