Kann ich eine asynchrone Funktion von einer synchronen Funktion ununterscheidbar machen? ⇐ Python

[Guest]

Wenn ich eine Funktion entweder auf synchrone oder asynchrone Weise schreiben kann (z. B. zum Abrufen einiger Daten von einer API), möchte ich sie idealerweise nur einmal als asynchron implementieren und sie immer dann als synchron ausführen, wenn ich eine nicht-

Code: Select all

async

-Funktion.
Ich suche daher eine run-Funktion mit der folgenden Signatur:

Code: Select all

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

Eine einfache Lösung wäre, einfach die in asyncio definierte Ausführungsfunktion zu verwenden:

Code: Select all

run1 = asyncio.run

Das Problem besteht darin, dass, wenn ich diese Funktion in eine synchrone Version einbinde:

Code: Select all

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

Dann kann ich sync_f nicht genau als synchrone Funktion verwenden.
Um dies zu sehen, stellen Sie sich ein anderes (Sync-)Modul vor, das auf sync_f aufbaut, um dies zu tun andere synchrone Dinge:

Code: Select all

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

Und dass endlich eine asynchrone Funktion async_h die sync_g-Logik verwenden möchte:

Code: Select all

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

Das Ausführen dieser letzten Funktion, beispielsweise in einem __main__-Block mit asyncio.run(async_h()), führt zu folgendem Fehler: RuntimeError: asyncio.run() kann nicht aus einer laufenden Ereignisschleife aufgerufen werden.
Ich habe versucht, mit meiner Definition von run etwas intelligenter zu sein, um herauszufinden, ob Derzeit läuft eine übergeordnete Schleife, und zwar meine Coroutine in das:

Code: Select all

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)

Aber ohne Erfolg: RuntimeError: Diese Ereignisschleife läuft bereits.
Was Sinn macht, aber ich konnte nichts finden, das die aktuelle Ausführung verwenden würde (so etwas wie loop.wait_until_complete(coro)).
Gibt es keine Möglichkeit, eine asynchrone Funktion in eine normale Funktion zu packen, die genau wie eine synchrone Funktion funktioniert? ohne die Implementierung zu haben Detail der asynchronen Version, die in die höheren Kontexte gelangt?