Как в асинхронном коде вызвать синхронную функцию?

Question

Для вызова синхронной функции в асинхронном коде используйте asyncio.to_thread() для CPU-bound задач или loop.run_in_executor() для выполнения в отдельном потоке. Для I/O-bound операций лучше переписать функцию в асинхронную версию.

YeaHub · Accepted Answer

Интеграция синхронного кода в асинхронное приложение требует осторожности, чтобы не блокировать event loop.Методы вызова синхронных функций:1. asyncio.to_thread() (Python 3.9+):import asyncio
import time

def sync_cpu_function():
    time.sleep(2)  # Имитация CPU-bound операции
    return "результат"

async def main():
    # Запуск в отдельном потоке
    result = await asyncio.to_thread(sync_cpu_function)
    print(f"Результат: {result}")

asyncio.run(main())2. loop.run_in_executor():import asyncio
import concurrent.futures

def sync_io_function():
    # Имитация I/O операции
    with open("file.txt", "r") as f:
        return f.read()

async def main():
    loop = asyncio.get_running_loop()
    # Запуск в ThreadPoolExecutor
    result = await loop.run_in_executor(
        None, sync_io_function  # None = default executor
    )
    print(f"Результат: {result}")

asyncio.run(main())Ограничения и рекомендации:CPU-bound задачи: используйте отдельные процессыI/O-bound задачи: используйте отдельные потокиБлокирующие вызовы: избегайте в основном потоке