Как правильно вызывать синхронный код из асинхронной функции в Python

«Как правильно вызывать синхронный код из асинхронной функции в Python» — вопрос из категории Асинхронность, который задают на 10% собеседований Python Разработчик. Ниже — развёрнутый ответ с разбором ключевых моментов.

Ответ

Да, это возможно, но требует правильного подхода, чтобы не блокировать цикл событий (event loop), что является главной проблемой при смешивании синхронного и асинхронного кода.

Существует два основных сценария:

1. Для быстрых, неблокирующих операций (CPU-bound)

Если синхронная функция выполняется быстро и не ожидает операций ввода-вывода (I/O), ее можно вызывать напрямую. Это не нанесет вреда производительности.

import asyncio

def fast_sync_operation(a, b):
    # Быстрые вычисления без ожидания
    return a + b

async def main():
    result = fast_sync_operation(5, 10)
    print(f"Результат: {result}")
    await asyncio.sleep(1) # Другая асинхронная работа

asyncio.run(main())

2. Для долгих, блокирующих операций (I/O-bound)

Если функция выполняет долгие операции (например, работа с файлами, запросы к БД через синхронную библиотеку, тяжелые вычисления), ее нужно запускать в отдельном потоке, чтобы не замораживать event loop. Для этого используется loop.run_in_executor().

import asyncio
import time

def slow_blocking_io():
    print("Начало долгой операции...")
    time.sleep(2) # Имитация блокирующей операции (например, чтение файла)
    print("...долгая операция завершена")
    return "Данные получены"

async def main():
    loop = asyncio.get_running_loop()
    print("Запускаем блокирующую задачу без остановки event loop")

    # Запуск синхронной функции в пуле потоков по умолчанию
    result = await loop.run_in_executor(None, slow_blocking_io)

    print(f"Результат из executor: {result}")

asyncio.run(main())

Ключевое правило: Никогда не вызывайте блокирующие I/O операции напрямую в асинхронном коде. Всегда используйте run_in_executor или ищите асинхронные аналоги библиотек.