バッチ処理
bca.analyze_batch(paths) は、イテラブル内のすべてのパスに対して bca.analyze と同じ分析を実行し、ファイル単位のエラーで例外を送出することはありません。各結果要素は、分析結果の dict か、失敗内容を記述する bca.AnalysisFailure のいずれかです。結果は入力順を保持するため、どのパスもスキップされない限り zip(inputs, results) はインデックスどおりに対応します。analyze_batch は analyze と同じキーワード専用オプション — exclude_tests、allow_lossy_path、skip_generated(デフォルト True)、metrics — を共有しており、2 つのエントリポイントは同じ挙動を保ちます。
def run(paths: Iterable[Path]) -> dict[str, int]:
"""Analyse ``paths`` as a batch and bucket successes vs failures.
Returns a small summary dict (`ok`, `errors`, `total`) so the
accompanying test can assert on it without re-parsing.
"""
materialised = list(paths)
# `skip_generated=False` は入力ごとに 1 つの結果要素を保証します
# (生成ファイルは除外されず分析されます)。そのため `materialised`
# に対する `strict=True` の zip が `ValueError` を送出することは
# ありません。2.0 のデフォルト(`skip_generated=True`)では生成
# ファイルの入力はスロットを生まず、長さがずれて strict な zip が
# 失敗します — `pipeline_db.py` で修正されたバグ #660 と同じです。
results = bca.analyze_batch(materialised, skip_generated=False)
ok = 0
errors = 0
for path, result in zip(materialised, results, strict=True):
if isinstance(result, bca.AnalysisFailure):
errors += 1
print(f" skip {path}: ({result.error_kind}) {result.error}")
else:
ok += 1
sloc = result["metrics"]["loc"]["sloc"]
print(f" ok {path}: sloc = {sloc:.0f}")
return {"ok": ok, "errors": errors, "total": len(materialised)}
重要な契約をいくつか挙げます。
AnalysisFailureは例外として送出されるのではなく、「返されます」。Exceptionのサブクラスではないため、isinstance(slot, bca.AnalysisFailure)が判別手段になります。pathsは遅延評価で消費されるため、ジェネレーターも使えます — ただしzipのために入力を保持したい場合は、先にリストへ実体化してください。- デフォルトの
skip_generated=Trueでは、生成ファイルは スキップ され、要素を 一切 生成しません。そのため結果リストは入力より短くなることがあります — これは、生成ファイルに対してNoneを返す単一ファイル版analyzeと正確に一致する挙動です。入力ごとに 1 要素を保証したい場合はskip_generated=Falseを渡してください(2.0 より前のデフォルト)。このデフォルトが 2.0 で反転したのは、analyzeとanalyze_batchを切り替えても生成ファイルの扱いが暗黙に変わらないようにするためです。
ディレクトリの走査:analyze_paths
analyze_batch は、明示的なリスト で渡されたパスをそのまま分析します。まずソースファイルを 見つける ところから始めたい場合 —「リポジトリを丸ごと分析したい」場合 — は、CLI の gitignore 対応ウォーカーを再利用する analyze_paths(#658)を使ってください。
import big_code_analysis as bca
results = bca.analyze_paths("path/to/repo", include="*.py")
各位置引数のシードにはファイルまたはディレクトリを指定できます。ディレクトリは .gitignore、include / exclude グロブ(単一のグロブ文字列またはシーケンス。先頭の ./ は省略可能で、dir/** ≡ ./dir/**)、および生成ファイルフィルターを尊重して走査されます。ファイルを直接指すシードは exclude に関係なく常に分析されます — 明示的な指定は ignore 系のルールに優先します — が、include はベース名による絞り込みとして引き続き適用されます。respect_gitignore=False を指定すると、ignore されたファイルも走査対象になります。結果は analyze_batch と同じ list[FuncSpaceDict | AnalysisFailure] の形と、例外を送出しない契約に従い、同じ exclude_tests / allow_lossy_path / skip_generated / metrics / vcs / vcs_per_function キーワード引数を転送します。
変更履歴メトリクスの付与
analyze_batch と analyze_paths は、単一ファイル版 analyze と同じ vcs=True / vcs_per_function=True キーワード引数を受け付けます(#670)。バッチは、対象ファイルを含むリポジトリごとに履歴インデックス / blame エンジンを 1 つだけ 構築し、そのリポジトリのファイル間で再利用します — analyze(p, vcs=True) を内包表記で回した場合にファイルごとに繰り返される走査を償却する形です。あるファイルで VCS が失敗しても、その AST メトリクスは無傷のまま残ります(AnalysisFailure になることはありません)。どのリポジトリにも属さないファイルには、単に vcs ブロックが付かないだけです。(ファイル単位の付与ではなく)リポジトリ全体をランキングしたい場合は、専用の big_code_analysis.vcs サーフェスを使ってください。
並列実行
analyze_batch に組み込みの並行処理はありません — 逐次的なスイープです。並列化するには、ファイルごとの analyze 呼び出しをスレッドプールへファンアウトしてください。
def run_parallel(paths: Iterable[Path], *, workers: int = 4) -> list[FuncSpaceDict | None]:
"""Fan ``analyze`` out across a thread pool.
PyO3 releases the GIL across each file's read + parse, so a
thread pool actually parallelises the heavy work. Use this when
you need per-file exceptions instead of ``AnalysisFailure`` slots.
"""
def _analyze(p: Path) -> FuncSpaceDict | None:
return bca.analyze(p)
with ThreadPoolExecutor(max_workers=workers) as pool:
return list(pool.map(_analyze, paths))
PyO3 の Python::detach は、各ファイルの読み込みと tree-sitter によるパースの間 GIL を解放するため、スレッドがインタープリターのロックで直列化されることはありません — ロックを取り合う協調動作ではなく、本当の並列処理です。
AnalysisFailure の分類
error_kind は閉じた Literal です。
error_kind | 発生条件 |
|---|---|
"UnsupportedLanguage" | 未知の拡張子で、シバン / emacs モードにも一致しない |
"ParseError" | tree-sitter がソースを拒否したか、まれな内部シリアライズ失敗(internal: serialization error: …) |
"IoError" | std::fs::read が失敗した、「または」 パスが有効な UTF-8 でなかった |
AnalysisFailure は凍結(frozen)されており、3 つのフィールドすべてに対する __eq__ / __hash__ / __repr__ を実装しているため、呼び出し側はエラーを set に入れて実行間で失敗を重複排除できます。リトライの分類のために、errno は Rust のデフォルト書式によって error 文字列内に保持されます。
import re
match = re.search(r"\(os error (\d+)\)$", slot.error)
errno = int(match.group(1)) if match else None
型付きのディスパッチ(FileNotFoundError、PermissionError など)が必要な場合は、analyze_batch の代わりにファイルごとに bca.analyze(path) を呼んでください — 単一ファイル版 analyze は正規の OSError サブクラスを送出します。エラー処理を参照してください。