app/async_executor.py

"""Dedicated async executor for CPU-bound or sync third-party work.

The project’s async request handlers must never block the event loop with
synchronous operations. For third-party libraries that have no async API,
wrap work via :func:`run_sync_in_executor`.
"""

from __future__ import annotations

import asyncio
import os
from concurrent.futures import ThreadPoolExecutor
from functools import partial
from typing import Any, Awaitable, Callable, TypeVar

T = TypeVar("T")

_max_workers = min(32, (os.cpu_count() or 1) + 4)
_executor: ThreadPoolExecutor = ThreadPoolExecutor(max_workers=_max_workers)


async def run_sync_in_executor(
    func: Callable[..., T],
    /,
    *args: Any,
    **kwargs: Any,
) -> T:
    """Run a synchronous callable in the process-wide thread pool.

    Notes:
        This is intended for async code paths only.
    """
    loop = asyncio.get_running_loop()
    bound = partial(func, *args, **kwargs)
    return await loop.run_in_executor(_executor, bound)