feat: parallel file analysis with --parallel flag

Add concurrent file analysis support using asyncio.Semaphore.
The --parallel N option controls how many files are analyzed
simultaneously (default 1 preserves sequential behavior).

Configurable via CLI flag, config file, or programmatic API.
Includes tests for parallel scan paths.

Author: frankentini

src/ai_sec_scan/__init__.py

@@ -2,7 +2,7 @@
 
 from ai_sec_scan.models import Finding, ScanResult, Severity
 from ai_sec_scan.providers.base import BaseProvider
-from ai_sec_scan.scanner import collect_files, run_scan_sync
+from ai_sec_scan.scanner import collect_files, run_scan_sync, scan
 
 __version__ = "0.2.0"
 
@@ -14,4 +14,5 @@
     "__version__",
     "collect_files",
     "run_scan_sync",
+    "scan",
 ]

src/ai_sec_scan/cli.py

@@ -47,6 +47,7 @@
     "quiet": "quiet",
     "cache_dir": "cache_dir",
     "no_cache": "no_cache",
+    "parallel": "parallel",
 }
 
 
@@ -241,6 +242,13 @@ def version() -> None:
     default=False,
     help="Disable result caching.",
 )
+@click.option(
+    "--parallel",
+    default=1,
+    type=int,
+    show_default=True,
+    help="Number of files to analyze concurrently.",
+)
 def scan(
     path: str,
     provider: str,
@@ -256,6 +264,7 @@ def scan(
     quiet: bool,
     cache_dir: str | None,
     no_cache: bool,
+    parallel: int,
 ) -> None:
     """Scan a file or directory for security vulnerabilities."""
     from ai_sec_scan.scanner import collect_files, run_scan_sync
@@ -303,6 +312,7 @@ def scan(
         quiet=quiet,
         cache_dir=Path(cache_dir) if cache_dir else None,
         no_cache=no_cache,
+        parallel=parallel,
     )
 
     # Render output

src/ai_sec_scan/scanner.py

@@ -144,6 +144,7 @@ async def scan(
     quiet: bool = False,
     cache_dir: Path | None = None,
     no_cache: bool = False,
+    parallel: int = 1,
 ) -> ScanResult:
     """Run a security scan on a file or directory.
 
@@ -157,6 +158,9 @@ async def scan(
         quiet: Suppress progress output.
         cache_dir: Directory for the result cache. ``None`` uses the default.
         no_cache: Disable caching entirely when ``True``.
+        parallel: Maximum number of files to analyze concurrently. Defaults
+            to ``1`` (sequential). Values above 1 enable parallel analysis
+            using an ``asyncio.Semaphore``.
 
     Returns:
         ScanResult with all findings.
@@ -175,31 +179,67 @@ async def scan(
         )
 
     cache = None if no_cache else ResultCache(cache_dir=cache_dir)
+    concurrency = max(1, parallel)
 
     all_findings: list[Finding] = []
     start_time = time.monotonic()
 
-    if quiet:
-        for fp in files:
-            rel_path = str(fp.relative_to(path)) if path.is_dir() else fp.name
-            findings = await _analyze_file(fp, rel_path, provider, cache)
-            all_findings.extend(findings)
-    else:
-        with Progress(
-            SpinnerColumn(),
-            TextColumn("[progress.description]{task.description}"),
-            console=console,
-        ) as progress:
-            task = progress.add_task("Scanning...", total=len(files))
-
+    if concurrency == 1:
+        # Sequential path (original behaviour)
+        if quiet:
             for fp in files:
                 rel_path = str(fp.relative_to(path)) if path.is_dir() else fp.name
-                progress.update(task, description=f"Scanning {rel_path}")
+                findings = await _analyze_file(fp, rel_path, provider, cache)
+                all_findings.extend(findings)
+        else:
+            with Progress(
+                SpinnerColumn(),
+                TextColumn("[progress.description]{task.description}"),
+                console=console,
+            ) as progress:
+                task = progress.add_task("Scanning...", total=len(files))
+
+                for fp in files:
+                    rel_path = str(fp.relative_to(path)) if path.is_dir() else fp.name
+                    progress.update(task, description=f"Scanning {rel_path}")
+
+                    findings = await _analyze_file(fp, rel_path, provider, cache)
+                    all_findings.extend(findings)
+
+                    progress.advance(task)
+    else:
+        # Parallel path
+        semaphore = asyncio.Semaphore(concurrency)
+        results_lock = asyncio.Lock()
+
+        progress_ctx = (
+            None
+            if quiet
+            else Progress(
+                SpinnerColumn(),
+                TextColumn("[progress.description]{task.description}"),
+                console=console,
+            )
+        )
+
+        task_id = None
+        if progress_ctx is not None:
+            progress_ctx.start()
+            task_id = progress_ctx.add_task("Scanning...", total=len(files))
 
+        async def _process(fp: Path) -> None:
+            rel_path = str(fp.relative_to(path)) if path.is_dir() else fp.name
+            async with semaphore:
                 findings = await _analyze_file(fp, rel_path, provider, cache)
+            async with results_lock:
                 all_findings.extend(findings)
+            if progress_ctx is not None and task_id is not None:
+                progress_ctx.advance(task_id)
+
+        await asyncio.gather(*[_process(fp) for fp in files])
 
-                progress.advance(task)
+        if progress_ctx is not None:
+            progress_ctx.stop()
 
     duration = time.monotonic() - start_time
 
@@ -228,11 +268,12 @@ def run_scan_sync(
     quiet: bool = False,
     cache_dir: Path | None = None,
     no_cache: bool = False,
+    parallel: int = 1,
 ) -> ScanResult:
     """Synchronous wrapper for the async scan function."""
     return asyncio.run(
         scan(
             path, provider, include, exclude, max_file_size_kb,
-            min_severity, quiet, cache_dir, no_cache,
+            min_severity, quiet, cache_dir, no_cache, parallel,
         )
     )

tests/test_cli.py

@@ -46,6 +46,7 @@ def fake_run_scan_sync(  # type: ignore[no-untyped-def]
         quiet: bool = False,
         cache_dir: Path | None = None,
         no_cache: bool = False,
+        parallel: int = 1,
     ) -> ScanResult:
         captured["include"] = include
         captured["exclude"] = exclude
@@ -150,6 +151,7 @@ def fake_run_scan_sync(  # type: ignore[no-untyped-def]
         quiet: bool = False,
         cache_dir: Path | None = None,
         no_cache: bool = False,
+        parallel: int = 1,
     ) -> ScanResult:
         captured["quiet"] = quiet
         return ScanResult(
@@ -195,6 +197,7 @@ def fake_run_scan_sync(  # type: ignore[no-untyped-def]
         quiet: bool = False,
         cache_dir: Path | None = None,
         no_cache: bool = False,
+        parallel: int = 1,
     ) -> ScanResult:
         captured["include"] = include
         captured["max_file_size_kb"] = max_file_size_kb

tests/test_scanner.py

@@ -192,3 +192,46 @@ def test_scan_quiet_mode(self, tmp_path: Path) -> None:
         result = asyncio.run(scan(tmp_path, provider, quiet=True))
         assert result.files_scanned == 1
         assert result.findings == []
+
+    def test_parallel_scan_collects_all_findings(self, tmp_path: Path) -> None:
+        for i in range(5):
+            (tmp_path / f"mod{i}.py").write_text(f"x = {i}")
+        finding = Finding(
+            file_path="placeholder",
+            line_start=1,
+            severity=Severity.MEDIUM,
+            title="Test finding",
+            description="d",
+            recommendation="r",
+        )
+        provider = MockProvider(findings=[finding])
+        result = asyncio.run(scan(tmp_path, provider, quiet=True, parallel=3))
+        assert result.files_scanned == 5
+        assert len(result.findings) == 5
+
+    def test_parallel_scan_with_error_provider(self, tmp_path: Path) -> None:
+        (tmp_path / "a.py").write_text("x = 1")
+        (tmp_path / "b.py").write_text("y = 2")
+        provider = ErrorProvider()
+        result = asyncio.run(scan(tmp_path, provider, quiet=True, parallel=2))
+        assert result.files_scanned == 2
+        assert result.findings == []
+
+    def test_parallel_scan_severity_filter(self, tmp_path: Path) -> None:
+        for name in ("a.py", "b.py", "c.py"):
+            (tmp_path / name).write_text("code")
+        findings = [
+            Finding(
+                file_path="x",
+                line_start=1,
+                severity=Severity.LOW,
+                title="Low",
+                description="d",
+                recommendation="r",
+            ),
+        ]
+        provider = MockProvider(findings=findings)
+        result = asyncio.run(
+            scan(tmp_path, provider, quiet=True, parallel=2, min_severity="high")
+        )
+        assert result.findings == []