A single Python function for both async/sync
Scenario: I often need to write Python functions like:
- take some parameters and format them
- call an API with the formatted parameters
- 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.
Posted
stdout
