Могу ли я сделать асинхронную функцию неотличимой от синхронной? ⇐ Python

[Anonymous]

Если я могу написать функцию синхронно или асинхронно (например, для получения некоторых данных из API), в идеале мне хотелось бы реализовать ее только один раз как асинхронную и запускать ее как синхронную всякий раз, когда я реализую не-

Код: Выделить всё

async

.
Поэтому я ищу функцию запуска со следующей сигнатурой:

Код: Выделить всё

def run(coro: Coroutine[Any, Any, _ReturnT]) -> _ReturnT: ...

Простым решением было бы просто использовать функцию запуска, определенную в asyncio:

Код: Выделить всё

run1 = asyncio.run

Проблема в том, что если я оберну эту функцию в синхронную версию:

Код: Выделить всё

def sync_f():
return run(async_f())

Тогда я не могу использовать sync_f точно как синхронную функцию.
Чтобы увидеть это, представьте, что другой модуль (синхронизации) построен на sync_f, чтобы сделать другие синхронные вещи:

Код: Выделить всё

def sync_g():
print("Doing some synchronous things")
res = sync_f()
print("Doing some other synchronous things")
return res

И наконец, какая-то асинхронная функция async_h хочет использовать логику sync_g:

Код: Выделить всё

async def async_h():
print("Doing some asynchronous things")
res = sync_g()
print("Doing some other asynchronous things")
return res

Затем запуск этой последней функции, например, в блоке __main__ с asyncio.run(async_h()), приведет к следующей ошибке: RuntimeError: asyncio.run() нельзя вызвать из текущего цикла событий.
Я пытался быть немного умнее с моим определением запуска, пытаясь увидеть, в настоящее время выполняется цикл более высокого уровня, и выполняется мой сопрограмма в этой:

Код: Выделить всё

def run2(coro: Coroutine[Any, Any, _ReturnT]) -> _ReturnT:
try:
loop = asyncio.get_running_loop()
except RuntimeError:
return asyncio.run(coro)
else:
return loop.run_until_complete(coro)

Но безрезультатно: RuntimeError: этот цикл событий уже запущен.
Это имеет смысл, но я не смог найти что-то, что использовало бы текущий запуск (что-то вроде цикла.wait_until_complete(coro)).
Нет ли способа обернуть асинхронную функцию в обычную, которая будет работать точно так же, как синхронная, без подробностей реализации асинхронная версия просачивается в более высокие контексты?

Подробнее здесь: https://stackoverflow.com/questions/793 ... ronous-one