utils/async_helpers.py

"""Async utility helpers for ProjectCraft."""

import asyncio
from typing import Any, Callable


async def run_in_executor(fn: Callable, *args) -> Any:
    """Run a synchronous function in a thread pool executor without blocking the event loop."""
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(None, fn, *args)


async def gather_with_errors(*coros) -> list:
    """
    Run coroutines in parallel. Returns list of results.
    Failed coroutines return the exception object instead of raising.
    """
    return await asyncio.gather(*coros, return_exceptions=True)