Add PyCG sharding for level 2 analysis on large projects

- Introduces --pycg-shard/--no-pycg-shard to run PyCG independently per
  Python package root instead of over the whole project, with cross-package
  imports treated as ghost nodes.
- Adds --pycg-shard-ceiling (default 100) to skip shards with too many
  files, and --pycg-shard-timeout (default 120s) as a final
  safety net for packages whose pointer fixpoint diverges indefinitely.
- Adds test fixtures (decorators_and_hof, class_hierarchy, async_patterns,
  Flask 3.0.3, requests 2.31.0) and corresponding CLI tests with PyCG-
  specific edge assertions.

Verified on a 6086-file project: 74,008 PyCG edges produced across 748/753
shards; 5 deep-OO framework shards timed out and were gracefully skipped.

Signed-off-by: Saurabh Sinha <sinha108@gmail.com>

Author: sinha108

codeanalyzer/__main__.py

@@ -80,6 +80,48 @@ def main(
     verbosity: Annotated[
         int, typer.Option("-v", count=True, help="Increase verbosity: -v, -vv, -vvv")
     ] = 0,
+    pycg_shard: Annotated[
+        bool,
+        typer.Option(
+            "--pycg-shard/--no-pycg-shard",
+            help=(
+                "Shard PyCG call-graph analysis by Python package (level 2 only). "
+                "When the project exceeds the 500-file ceiling, PyCG is run "
+                "independently per top-level package with cross-package imports "
+                "treated as ghost nodes. Without this flag, projects over the "
+                "ceiling fall back to Jedi-only edges."
+            ),
+        ),
+    ] = False,
+    pycg_shard_ceiling: Annotated[
+        int,
+        typer.Option(
+            "--pycg-shard-ceiling",
+            help=(
+                "Maximum files per shard when --pycg-shard is active (default 100). "
+                "Shards exceeding this limit are skipped; their call edges are "
+                "omitted from the call graph (Jedi edges for those packages are "
+                "still included). Lower values are safer for packages with deep "
+                "class hierarchies or heavy import graphs."
+            ),
+            min=1,
+        ),
+    ] = 100,
+    pycg_shard_timeout: Annotated[
+        int,
+        typer.Option(
+            "--pycg-shard-timeout",
+            help=(
+                "Per-shard wall-clock timeout in seconds when --pycg-shard is "
+                "active (default 120). A shard that exceeds this limit is skipped "
+                "gracefully. PyCG's fixpoint is bimodal: it either converges "
+                "quickly or diverges indefinitely, so the timeout acts as a final "
+                "safety net after the file-count ceiling. Set to 0 to disable. "
+                "POSIX only (macOS / Linux); ignored on Windows."
+            ),
+            min=0,
+        ),
+    ] = 120,
 ):
     options = AnalysisOptions(
         input=input,
@@ -93,6 +135,9 @@ def main(
         cache_dir=cache_dir,
         clear_cache=clear_cache,
         verbosity=verbosity,
+        pycg_shard=pycg_shard,
+        pycg_shard_ceiling=pycg_shard_ceiling,
+        pycg_shard_timeout=pycg_shard_timeout,
     )
 
     _set_log_level(options.verbosity)

codeanalyzer/core.py

@@ -247,26 +247,33 @@ def __enter__(self) -> "Codeanalyzer":
             for dep_file, pip_args in dependency_files:
                 if (self.project_dir / dep_file).exists():
                     logger.info(f"Installing dependencies from {dep_file}")
-                    self._cmd_exec_helper(
-                        [str(venv_python), "-m", "pip", "install", "-U"] + pip_args + [str(self.project_dir / dep_file)],
-                        cwd=self.project_dir,
-                        check=True,
-                    )
+                    try:
+                        self._cmd_exec_helper(
+                            [str(venv_python), "-m", "pip", "install", "-U"] + pip_args + [str(self.project_dir / dep_file)],
+                            cwd=self.project_dir,
+                            check=True,
+                        )
+                    except subprocess.CalledProcessError as e:
+                        logger.warning(
+                            f"Dependency installation from {dep_file} failed (analysis will continue): {e}"
+                        )
 
             # Handle Pipenv files
             if (self.project_dir / "Pipfile").exists():
                 logger.info("Installing dependencies from Pipfile")
-                # Note: This would require pipenv to be installed
-                self._cmd_exec_helper(
-                    [str(venv_python), "-m", "pip", "install", "pipenv"],
-                    cwd=self.project_dir,
-                    check=True,
-                )
-                self._cmd_exec_helper(
-                    ["pipenv", "install", "--dev"],
-                    cwd=self.project_dir,
-                    check=True,
-                )
+                try:
+                    self._cmd_exec_helper(
+                        [str(venv_python), "-m", "pip", "install", "pipenv"],
+                        cwd=self.project_dir,
+                        check=True,
+                    )
+                    self._cmd_exec_helper(
+                        ["pipenv", "install", "--dev"],
+                        cwd=self.project_dir,
+                        check=True,
+                    )
+                except subprocess.CalledProcessError as e:
+                    logger.warning(f"Pipenv dependency installation failed (analysis will continue): {e}")
 
             # Handle conda environment files
             conda_files = ["conda.yml", "environment.yml"]
@@ -284,11 +291,14 @@ def __enter__(self) -> "Codeanalyzer":
 
             if any((self.project_dir / file).exists() for file in package_definition_files):
                 logger.info("Installing project in editable mode")
-                self._cmd_exec_helper(
-                    [str(venv_python), "-m", "pip", "install", "-e", str(self.project_dir)],
-                    cwd=self.project_dir,
-                    check=True,
-                )
+                try:
+                    self._cmd_exec_helper(
+                        [str(venv_python), "-m", "pip", "install", "-e", str(self.project_dir)],
+                        cwd=self.project_dir,
+                        check=True,
+                    )
+                except subprocess.CalledProcessError as e:
+                    logger.warning(f"Editable install failed (analysis will continue): {e}")
             else:
                 logger.warning("No package definition files found, skipping editable installation")
 
@@ -552,11 +562,18 @@ def _get_pycg_call_graph(
         continue with Jedi-only edges.
         """
         try:
-            pycg = PyCG(self.project_dir, skip_tests=self.skip_tests)
+            pycg = PyCG(
+                self.project_dir,
+                skip_tests=self.skip_tests,
+                shard=self.options.pycg_shard,
+                shard_ceiling=self.options.pycg_shard_ceiling,
+                shard_timeout=self.options.pycg_shard_timeout,
+            )
             return pycg.build_call_graph_edges(symbol_table)
         except PyCGExceptions.PyCGImportError as exc:
             logger.warning(f"PyCG not installed — level 2 edges will be Jedi-only: {exc}")
             return []
         except PyCGExceptions.PyCGAnalysisError as exc:
             logger.warning(f"PyCG analysis failed — level 2 edges will be Jedi-only: {exc}")
+            logger.debug("PyCG full traceback:", exc_info=True)
             return []

codeanalyzer/options/options.py

@@ -22,3 +22,6 @@ class AnalysisOptions:
     cache_dir: Optional[Path] = None
     clear_cache: bool = False
     verbosity: int = 0
+    pycg_shard: bool = False
+    pycg_shard_ceiling: int = 100
+    pycg_shard_timeout: int = 120