Как вызвать асинхронную функцию из дочернего потока, используя asyncio.run_coroutine_threadsafe() и цикл основного поток ⇐ Python

[Anonymous]

У меня есть приложение, которое везде работает с использованием async/await. Часто он вызывает синхронные функции. Этим функциям иногда приходится снова вызывать асинхронные функции. Я пытаюсь выполнить эту работу, используя asyncio.run_coroutine_threadsafe() из потока, используя цикл основного потока, но ожидая завершения выполнения Run_coroutine_threadsafe() Future, вызывая . result() — это то, что кажется зависающим.
Ниже приведен пример кода, который воспроизводит проблему в Python 3.12 и 3.13:

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

from asyncio import run_coroutine_threadsafe, get_running_loop, run, AbstractEventLoop
from collections.abc import Callable, Awaitable
from concurrent.futures.thread import ThreadPoolExecutor
from typing import TypeVar

_T = TypeVar("_T")

def _await_in_thread(loop: AbstractEventLoop, f: Callable[[], Awaitable[_T]]) -> _T:
"""
Await something inside a thread, using the main thread's loop.
"""

print("THIS IS PRINTED")
return run_coroutine_threadsafe(f(), loop).result()

def _await_to_thread(pool: ThreadPoolExecutor, f: Callable[[], Awaitable[_T]]) -> _T:
"""
Await something by moving it to a thread.
"""
return pool.submit(_await_in_thread, get_running_loop(), f).result()

async def _async_main(pool: ThreadPoolExecutor) -> None:
"""
Run the main application, which is asynchronous.
"""

# Eventually, the application calls a function that due to its nature (maybe a third-party API) is synchronous.
_some_sync_function(pool)

def _sync_main() -> None:
with ThreadPoolExecutor() as pool:
run(_async_main(pool))

def _some_sync_function(pool: ThreadPoolExecutor) -> None:
# This synchronous function then has to call a function that is asynchronous.
result = _await_to_thread(pool, _some_async_function)
assert result == 123

async def _some_async_function() -> int:
print("BUT THIS IS NOT PRINTED")
return 123

if __name__ == "__main__":
_sync_main()

Хотя пример мог бы быть немного проще, я хотел включить использование ThreadPoolExecutor, а также распространение возвращаемого значения, поскольку это два требования окончательного варианта. продукт.
Кроме того, совместимость с Python – 3.11+, но меня могут убедить сделать версию 3.12+, если нет другого решения.
Это является результатом единственного ответа в IRC и очень кратких примеров run_coroutine_threadsafe(), которые я нашел в документации Python, а также на различных сайтах вопросов и ответов и в блогах. Похоже, эта функция не широко известна и не используется людьми.
Мой вопрос: что я неправильно понял и как это должно работать?

Подробнее здесь: https://stackoverflow.com/questions/790 ... routine-th