run_in_executor and anyio: Sync Libraries in Async Code

import asyncio, requests

async def bad_fetch(url):
    # BAD: requests.get blocks the event loop for the entire request duration
    # All other coroutines freeze and wait
    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 runs in a separate thread, event loop is not blocked
    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 — sync code in a thread
    result = await anyio.to_thread.run_sync(blocking_function, arg1, arg2)

    # Parallel tasks via TaskGroup
    async with anyio.create_task_group() as tg:
        tg.start_soon(fetch, url1)
        tg.start_soon(fetch, url2)

# Run under asyncio or trio
anyio.run(main)