非同期パターン

bca.analyze is CPU-bound: the work is a tree-sitter parse plus the metric passes, both of which release the GIL on the Rust side via PyO3’s Python::detach. The canonical pattern for calling it from an asyncio application is therefore asyncio.to_thread:

async def analyze_async(path: Path) -> FuncSpaceDict | None:
    """Run ``bca.analyze(path)`` on the default thread executor."""
    return await asyncio.to_thread(bca.analyze, path)


async def analyze_all(
    paths: Iterable[Path],
) -> list[FuncSpaceDict | BaseException | None]:
    """Fan ``analyze_async`` out across ``paths`` with ``asyncio.gather``.

    ``return_exceptions=True`` matters here: ``bca.analyze`` runs
    inside ``asyncio.to_thread`` and Python threads cannot be
    cancelled. If one call raises and gather re-raises with
    ``return_exceptions=False``, the surviving threads keep running
    in the default executor, producing results that are silently
    discarded. With ``return_exceptions=True`` every thread's
    result (success OR exception) lands in the returned list so
    the caller can dispatch per-file.
    """
    return await asyncio.gather(
        *(analyze_async(p) for p in paths),
        return_exceptions=True,
    )

ネイティブ async ではなく to_thread を使う理由

bca.analyze is a synchronous Python function backed by synchronous Rust code; there is no await boundary inside it. Wrapping it in asyncio.to_thread:

1. 呼び出しをデフォルトのスレッドプールにスケジュールします。
2. Lets other coroutines progress while the parse and metric pass run.
3. Returns the result to the calling coroutine when done.

Because the Rust side releases the GIL across the heavy work, several to_thread(bca.analyze, ...) calls run in parallel. This is not co-operative I/O multiplexing; it is real multi-core use, capped by the thread pool’s size.

カスタムエグゼキューター

to_thread always uses the event loop’s default executor. For a tighter cap on the worker count, use loop.run_in_executor with a purpose-built pool instead:

import asyncio
from concurrent.futures import ThreadPoolExecutor

import big_code_analysis as bca

async def analyze_many(paths: list[str]) -> list[object]:
    loop = asyncio.get_running_loop()
    with ThreadPoolExecutor(max_workers=8) as pool:
        return await asyncio.gather(
            *(loop.run_in_executor(pool, bca.analyze, p) for p in paths)
        )

Start with a worker count near the CPUs available to the process. Benchmark your workload before increasing it; additional workers compete for CPU time.

結果のストリーミング

asyncio.as_completed lets you start consuming results as soon as the first analysis finishes, which helps when the per-file work varies widely in cost (a 5 KB file against a 500 KB generated bundle):

import asyncio
import big_code_analysis as bca

async def first_failure(paths: list[str]) -> str | None:
    """Return the path of the first file with cyclomatic > 50."""
    tasks = [asyncio.create_task(asyncio.to_thread(bca.analyze, p)) for p in paths]
    try:
        for coro in asyncio.as_completed(tasks):
            result = await coro
            if result is None:
                continue
            if result["metrics"]["cyclomatic"]["sum"] > 50:
                return result["name"]
    finally:
        for t in tasks:
            t.cancel()
    return None

The finally block cancels pending asyncio tasks; as_completed does not cancel them itself when the caller returns early. Analyses already running in worker threads continue until they finish; cancellation does not interrupt the Rust work.

アンチパターン：コルーチン内での bca.analyze の直接呼び出し

# これはやってはいけません。
async def bad(path: str) -> dict | None:
    return bca.analyze(path)  # 呼び出しのたびにイベントループをブロックします

async def does not make the body asynchronous. Without to_thread or an explicit executor, every coroutine that calls bca.analyze stalls the event loop for the full duration of the parse: other tasks waiting on I/O, timers, or queues all freeze until the parse returns. The to_thread wrapper is one line and is the difference between a responsive server and a single-threaded one.

analyze_batch が適している場面

静的で有限のパスのリストを処理していて結果のストリーミングが不要なら、bca.analyze_batch の方が gather(*to_thread(...)) よりシンプルです。呼び出し元スレッド上で逐次実行されますが、ファイルごとのエラーで例外を送出することはありません。イベントループの応答性を保つには、analyze_batch の呼び出し全体を asyncio.to_thread でラップしてください：

import asyncio
import big_code_analysis as bca

async def batch(paths: list[str]) -> list[object]:
    return await asyncio.to_thread(bca.analyze_batch, paths)

これは gather のファイル単位の並列性を、analyze_batch のよりシンプルなエラーモデルと引き換えにします。並列性と型付き OSError の振り分けの両方が欲しい場合は gather を、非同期呼び出しを 1 つにまとめて例外を送出しない契約が欲しい場合は to_thread(analyze_batch, paths) を選んでください。