run_in_executor и anyio: sync-библиотеки в async коде

import asyncio, requests

async def bad_fetch(url):
    # ПЛОХО: requests.get блокирует event loop на всё время запроса
    # Все остальные coroutines замерзают и ждут
    return requests.get(url).text

import asyncio
from concurrent.futures import ThreadPoolExecutor
import requests

executor = ThreadPoolExecutor(max_workers=10)

async def fetch_sync(url: str) -> str:
    loop = asyncio.get_running_loop()
    # requests.get выполняется в отдельном потоке, event loop не блокируется
    return await loop.run_in_executor(executor, requests.get, url)

async def main():
    results = await asyncio.gather(
        fetch_sync("https://httpbin.org/get"),
        fetch_sync("https://httpbin.org/post"),
    )

from concurrent.futures import ProcessPoolExecutor

def heavy_computation(data: list) -> float:
    return sum(x ** 2 for x in data)  # CPU-intensive

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(
            pool, heavy_computation, list(range(10_000_000))
        )
    print(result)

import anyio

async def main():
    # to_thread.run_sync — синхронный код в потоке
    result = await anyio.to_thread.run_sync(blocking_function, arg1, arg2)

    # Параллельные задачи через TaskGroup
    async with anyio.create_task_group() as tg:
        tg.start_soon(fetch, url1)
        tg.start_soon(fetch, url2)

# Запуск под asyncio или trio
anyio.run(main)