feat(cli): emit the v2 Analysis envelope (schema_version/max_level/application)

Author: rahlk

codeanalyzer/__main__.py

@@ -353,7 +353,7 @@ def main(
 
             emit_neo4j(artifacts, options)
         elif options.output is None:
-            print(model_dump_json(artifacts, separators=(",", ":")))
+            print(model_dump_json(artifacts, exclude_none=True))
         else:
             options.output.mkdir(parents=True, exist_ok=True)
             _write_output(artifacts, options.output, options.format)
@@ -364,7 +364,7 @@ def _write_output(artifacts, output_dir: Path, format: OutputFormat):
     if format == OutputFormat.JSON:
         output_file = output_dir / "analysis.json"
         # Use Pydantic's model_dump_json() for compact output
-        json_str = model_dump_json(artifacts, indent=None)
+        json_str = model_dump_json(artifacts, indent=None, exclude_none=True)
         with output_file.open("w") as f:
             f.write(json_str)
         logger.info(f"Analysis saved to {output_file}")

codeanalyzer/core.py

@@ -11,6 +11,7 @@
 import ray
 from codeanalyzer.utils import logger
 from codeanalyzer.schema import (
+    Analysis,
     PyApplication,
     PyExternalSymbol,
     PyModule,
@@ -407,9 +408,9 @@ def walk_class(cl):
                 externals[sig] = PyExternalSymbol(name=name, module=module)
         return externals
 
-    def analyze(self) -> PyApplication:
-        """Analyze the project and return a PyApplication with symbol table.
-        
+    def analyze(self) -> Analysis:
+        """Analyze the project and return the v2 ``Analysis`` envelope.
+
         Uses caching to avoid re-analyzing unchanged files.
         """
         cache_file = self.cache_dir / "analysis_cache.json"
@@ -465,7 +466,11 @@ def analyze(self) -> PyApplication:
         # Save to cache
         self._save_analysis_cache(app, cache_file)
 
-        return app
+        return Analysis(
+            max_level=self.analysis_level,
+            k_limit=self.options.graph_field_depth,
+            application=app,
+        )
 
     def _load_pyapplication_from_cache(self, cache_file: Path) -> PyApplication:
         """Load cached analysis from file.

codeanalyzer/neo4j/emit.py

@@ -32,7 +32,7 @@
 from codeanalyzer.neo4j.cypher import render_cypher
 from codeanalyzer.neo4j.project import project
 from codeanalyzer.options import AnalysisOptions
-from codeanalyzer.schema import PyApplication
+from codeanalyzer.schema import Analysis
 from codeanalyzer.utils import logger
 
 
@@ -49,11 +49,11 @@ def emit_schema(output: Optional[Path]) -> None:
     logger.info(f"Neo4j schema written to {output / 'schema.json'}")
 
 
-def emit_neo4j(app: PyApplication, options: AnalysisOptions) -> None:
+def emit_neo4j(analysis: Analysis, options: AnalysisOptions) -> None:
     """Project the analysis to a graph and write it: a live Bolt push when
     ``--neo4j-uri`` is set, otherwise a self-contained ``graph.cypher`` snapshot."""
     app_name = options.app_name or Path(options.input).resolve().name
-    rows = project(app, app_name)
+    rows = project(analysis.application, app_name)
 
     if options.neo4j_uri:
         cfg = BoltConfig(

codeanalyzer/schema/__init__.py

@@ -72,6 +72,8 @@ def model_dump_json(model, **kwargs):
         v1_kwargs = {}
         if 'indent' in kwargs:
             v1_kwargs['indent'] = kwargs['indent']
+        if 'exclude_none' in kwargs:
+            v1_kwargs['exclude_none'] = kwargs['exclude_none']
         if 'separators' in kwargs:
             # In v1, separators is passed to dumps_kwargs
             v1_kwargs['separators'] = kwargs['separators']

test/test_v2_conformance.py

@@ -1,3 +1,8 @@
+import json
+import subprocess
+import sys
+from pathlib import Path
+
 from codeanalyzer.schema.assign_ids import assign_ids
 from codeanalyzer.schema.py_schema import PyApplication, PyModule, PyClass, PyCallable
 
@@ -13,3 +18,18 @@ def test_ids_assigned_down_the_tree():
     assert mod.id == "can://python/myapp/pkg/m.py"
     assert cl.id == "can://python/myapp/pkg/m.py/Hasher"
     assert fn.id == "can://python/myapp/pkg/m.py/Hasher/hash()"
+
+
+def test_cli_emits_v2_envelope(tmp_path: Path):
+    proj = tmp_path / "proj"; proj.mkdir()
+    (proj / "m.py").write_text("def f():\n    return 1\n", encoding="utf-8")
+    out = subprocess.run(
+        [sys.executable, "-m", "codeanalyzer", "-i", str(proj), "-a", "1", "--no-venv"],
+        capture_output=True, text=True, check=True,
+    ).stdout
+    payload = json.loads(out)
+    assert payload["schema_version"] == "2.0.0"
+    assert payload["language"] == "python"
+    assert payload["max_level"] == 1
+    assert payload["application"]["kind"] == "application"
+    assert "program_graphs" not in payload  # dissolved into the tree