feat(ontology): add --skip-property-graph for user-owned graph DDL (#104)

Lets users with their own CREATE PROPERTY GRAPH DDL — managed by
Terraform, dbt, or hand-authored — populate base tables from BQ AA
traces without overwriting the graph object on every run.

Changes
- ontology_orchestrator.build_ontology_graph gains
  skip_property_graph: bool = False. When True, phase 5 is not
  invoked: no OntologyPropertyGraphCompiler is constructed, no
  CREATE OR REPLACE PROPERTY GRAPH runs.
- Result dict gains property_graph_status with values "created" /
  "failed" / "skipped:user_requested", plus skipped_reason
  ("user_requested") when phase 5 was skipped.
- ontology-build CLI gains --skip-property-graph and threads
  property_graph_status through to the curated output dict so JSON
  consumers can distinguish "skipped" from "failed" without parsing
  stderr.
- Exit handling: skipped_reason == "user_requested" exits 0 silently;
  the existing exit-1-with-error behavior is preserved for actual
  graph-creation failures.

Tests
- test_skip_property_graph_does_not_construct_compiler asserts the
  compiler class is never called (mock.assert_not_called) when the
  flag is set.
- test_property_graph_status_created_on_success and
  test_property_graph_status_failed_on_compiler_false cover the two
  default-mode status values.
- CLI tests cover exit 0 with status="skipped:user_requested",
  default skip_property_graph=False threading, and exit 1 with
  status="failed" on actual creation failure.

135/135 tests in test_ontology_orchestrator.py + test_cli.py pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: caohy1988

src/bigquery_agent_analytics/cli.py

@@ -1238,6 +1238,16 @@ def ontology_build(
     no_ai_generate: bool = typer.Option(
         False, help="Skip AI.GENERATE; fetch raw payloads instead."
     ),
+    skip_property_graph: bool = typer.Option(
+        False,
+        "--skip-property-graph",
+        help=(
+            "Skip CREATE OR REPLACE PROPERTY GRAPH. Use when the caller "
+            "owns their own property-graph DDL and only wants the SDK to "
+            "populate base tables. CLI exits 0 with "
+            "property_graph_status='skipped:user_requested'."
+        ),
+    ),
     fmt: str = typer.Option(
         "json",
         "--format",
@@ -1261,6 +1271,7 @@ def ontology_build(
         table_id=table_id,
         endpoint=endpoint,
         use_ai_generate=not no_ai_generate,
+        skip_property_graph=skip_property_graph,
     )
 
     output = {
@@ -1271,9 +1282,19 @@ def ontology_build(
         "tables_created": result["tables_created"],
         "rows_materialized": result["rows_materialized"],
         "property_graph_created": result["property_graph_created"],
+        "property_graph_status": result.get(
+            "property_graph_status",
+            "created" if result["property_graph_created"] else "failed",
+        ),
     }
     typer.echo(format_output(output, fmt))
 
+    # Distinguish "user-requested skip" (exit 0) from "creation failed"
+    # (exit 1). Same property_graph_created=False, different operator
+    # intent — JSON consumers read property_graph_status to tell them
+    # apart without parsing stderr.
+    if result.get("skipped_reason") == "user_requested":
+      return
     if not result["property_graph_created"]:
       typer.echo(
           "Error: Property Graph creation failed. "

src/bigquery_agent_analytics/ontology_orchestrator.py

@@ -300,14 +300,16 @@ def build_ontology_graph(
     endpoint: str = "gemini-2.5-flash",
     use_ai_generate: bool = True,
     location: Optional[str] = None,
+    skip_property_graph: bool = False,
 ) -> dict[str, Any]:
   """Run the full ontology graph pipeline end-to-end.
 
   1. Load the YAML spec (or use pre-loaded ``spec``).
   2. Extract an ``ExtractedGraph`` from agent telemetry.
   3. Create physical tables (if not exists).
   4. Materialize extracted nodes/edges into tables.
-  5. Create the BigQuery Property Graph.
+  5. Create the BigQuery Property Graph (skipped when
+     ``skip_property_graph=True``).
 
   Args:
       session_ids: Sessions to extract from.
@@ -323,10 +325,22 @@ def build_ontology_graph(
       endpoint: AI.GENERATE model endpoint.
       use_ai_generate: If True, uses server-side AI extraction.
       location: BigQuery location.
+      skip_property_graph: When True, skip phase 5 (do not run
+          ``CREATE OR REPLACE PROPERTY GRAPH``). Use this when the
+          caller owns their own property-graph DDL and only wants
+          the SDK to populate base tables. The result dict reports
+          ``property_graph_created=False`` with
+          ``skipped_reason="user_requested"`` and
+          ``property_graph_status="skipped:user_requested"``, which
+          callers (and the CLI) use to distinguish a deliberate
+          skip from a creation failure.
 
   Returns:
       A dict with keys: ``spec``, ``graph``, ``tables_created``,
       ``rows_materialized``, ``property_graph_created``,
+      ``property_graph_status`` (one of ``"created"``, ``"failed"``,
+      ``"skipped:user_requested"``), ``skipped_reason`` (only set
+      when phase 5 was skipped, e.g. ``"user_requested"``),
       ``graph_name``, ``graph_ref``.
   """
   from .ontology_graph import OntologyGraphManager
@@ -391,24 +405,36 @@ def build_ontology_graph(
   rows_materialized = materializer.materialize(graph, session_ids)
   logger.info("Rows materialized: %s", rows_materialized)
 
-  # 5. Create property graph.
-  compiler = OntologyPropertyGraphCompiler(
-      project_id=project_id,
-      dataset_id=dataset_id,
-      spec=spec,
-      location=location,
-  )
-  pg_created = compiler.create_property_graph(graph_name=name)
-
   graph_ref = f"{project_id}.{dataset_id}.{name}"
-  logger.info("Property Graph %r created=%s.", graph_ref, pg_created)
 
-  return {
+  # 5. Create property graph (or skip when caller owns the DDL).
+  result: dict[str, Any] = {
       "spec": spec,
       "graph": graph,
       "tables_created": tables_created,
       "rows_materialized": rows_materialized,
-      "property_graph_created": pg_created,
       "graph_name": name,
       "graph_ref": graph_ref,
   }
+  if skip_property_graph:
+    logger.info(
+        "Property Graph creation skipped (skip_property_graph=True); "
+        "caller owns the DDL for graph %r.",
+        graph_ref,
+    )
+    result["property_graph_created"] = False
+    result["skipped_reason"] = "user_requested"
+    result["property_graph_status"] = "skipped:user_requested"
+  else:
+    compiler = OntologyPropertyGraphCompiler(
+        project_id=project_id,
+        dataset_id=dataset_id,
+        spec=spec,
+        location=location,
+    )
+    pg_created = compiler.create_property_graph(graph_name=name)
+    logger.info("Property Graph %r created=%s.", graph_ref, pg_created)
+    result["property_graph_created"] = pg_created
+    result["property_graph_status"] = "created" if pg_created else "failed"
+
+  return result

tests/test_cli.py

@@ -2472,3 +2472,111 @@ def test_bad_spec_path_exit_2(self):
         ],
     )
     assert result.exit_code == 2
+
+  @patch("bigquery_agent_analytics.ontology_orchestrator.build_ontology_graph")
+  def test_skip_property_graph_exits_zero_with_status(self, mock_build):
+    """--skip-property-graph: exit 0, status='skipped:user_requested'."""
+    from bigquery_agent_analytics.ontology_models import ExtractedGraph
+
+    mock_build.return_value = {
+        "graph_name": "g",
+        "graph_ref": "proj.ds.g",
+        "graph": ExtractedGraph(name="test"),
+        "tables_created": {"mako_DecisionPoint": "p.d.decision_points"},
+        "rows_materialized": {"mako_DecisionPoint": 2},
+        "property_graph_created": False,
+        "skipped_reason": "user_requested",
+        "property_graph_status": "skipped:user_requested",
+        "spec": MagicMock(),
+    }
+
+    result = runner.invoke(
+        app,
+        [
+            "ontology-build",
+            "--project-id=proj",
+            "--dataset-id=ds",
+            f"--spec-path={self._SPEC_PATH}",
+            "--session-ids=sess1",
+            "--env=p.d",
+            "--skip-property-graph",
+        ],
+    )
+    assert result.exit_code == 0
+    # Skip path must NOT print the "Property Graph creation failed" stderr.
+    assert "Property Graph creation failed" not in result.output
+    parsed = json.loads(result.output)
+    assert parsed["property_graph_created"] is False
+    assert parsed["property_graph_status"] == "skipped:user_requested"
+
+    # Flag is threaded through to the orchestrator.
+    _, kwargs = mock_build.call_args
+    assert kwargs["skip_property_graph"] is True
+
+  @patch("bigquery_agent_analytics.ontology_orchestrator.build_ontology_graph")
+  def test_default_invocation_omits_skip_flag(self, mock_build):
+    """Default invocation passes skip_property_graph=False."""
+    from bigquery_agent_analytics.ontology_models import ExtractedGraph
+
+    mock_build.return_value = {
+        "graph_name": "g",
+        "graph_ref": "proj.ds.g",
+        "graph": ExtractedGraph(name="test"),
+        "tables_created": {},
+        "rows_materialized": {},
+        "property_graph_created": True,
+        "property_graph_status": "created",
+        "spec": MagicMock(),
+    }
+
+    result = runner.invoke(
+        app,
+        [
+            "ontology-build",
+            "--project-id=proj",
+            "--dataset-id=ds",
+            f"--spec-path={self._SPEC_PATH}",
+            "--session-ids=sess1",
+            "--env=p.d",
+        ],
+    )
+    assert result.exit_code == 0
+    parsed = json.loads(result.output)
+    assert parsed["property_graph_status"] == "created"
+
+    _, kwargs = mock_build.call_args
+    assert kwargs["skip_property_graph"] is False
+
+  @patch("bigquery_agent_analytics.ontology_orchestrator.build_ontology_graph")
+  def test_property_graph_failure_status_failed(self, mock_build):
+    """When the orchestrator reports failure, exit 1 with status='failed'.
+
+    Distinguishes the failure path from the user-requested-skip path by
+    asserting the status field, not just the exit code.
+    """
+    from bigquery_agent_analytics.ontology_models import ExtractedGraph
+
+    mock_build.return_value = {
+        "graph_name": "g",
+        "graph_ref": "proj.ds.g",
+        "graph": ExtractedGraph(name="test"),
+        "tables_created": {},
+        "rows_materialized": {},
+        "property_graph_created": False,
+        "property_graph_status": "failed",
+        "spec": MagicMock(),
+    }
+
+    result = runner.invoke(
+        app,
+        [
+            "ontology-build",
+            "--project-id=proj",
+            "--dataset-id=ds",
+            f"--spec-path={self._SPEC_PATH}",
+            "--session-ids=sess1",
+            "--env=p.d",
+        ],
+    )
+    assert result.exit_code == 1
+    assert "Property Graph creation failed" in result.output

tests/test_ontology_orchestrator.py

@@ -469,3 +469,107 @@ def test_partial_table_creation_raises(
     # Materialize and property graph should NOT have been called.
     mock_mat_cls.return_value.materialize.assert_not_called()
     mock_pg_cls.return_value.create_property_graph.assert_not_called()
+
+  @patch(
+      "bigquery_agent_analytics.ontology_property_graph"
+      ".OntologyPropertyGraphCompiler"
+  )
+  @patch("bigquery_agent_analytics.ontology_materializer.OntologyMaterializer")
+  @patch("bigquery_agent_analytics.ontology_graph.OntologyGraphManager")
+  def test_skip_property_graph_does_not_construct_compiler(
+      self, mock_mgr_cls, mock_mat_cls, mock_pg_cls
+  ):
+    """When skip_property_graph=True, the compiler is never constructed."""
+    mock_mgr_cls.return_value.extract_graph.return_value = ExtractedGraph(
+        name="test"
+    )
+    mock_mat_cls.return_value.create_tables.return_value = dict(
+        _ALL_YMGO_TABLES
+    )
+    mock_mat_cls.return_value.materialize.return_value = {}
+
+    result = build_ontology_graph(
+        session_ids=["sess1"],
+        spec_path=_DEMO_SPEC_PATH,
+        project_id="proj",
+        dataset_id="ds",
+        env="p.d",
+        skip_property_graph=True,
+    )
+
+    # Compiler must not be constructed and create_property_graph must
+    # not be called when skip_property_graph=True.
+    mock_pg_cls.assert_not_called()
+    mock_pg_cls.return_value.create_property_graph.assert_not_called()
+
+    # Tables and rows still produced.
+    mock_mat_cls.return_value.create_tables.assert_called_once()
+    mock_mat_cls.return_value.materialize.assert_called_once()
+
+    # Result reports the skip distinctly from a creation failure.
+    assert result["property_graph_created"] is False
+    assert result["skipped_reason"] == "user_requested"
+    assert result["property_graph_status"] == "skipped:user_requested"
+
+  @patch(
+      "bigquery_agent_analytics.ontology_property_graph"
+      ".OntologyPropertyGraphCompiler"
+  )
+  @patch("bigquery_agent_analytics.ontology_materializer.OntologyMaterializer")
+  @patch("bigquery_agent_analytics.ontology_graph.OntologyGraphManager")
+  def test_property_graph_status_created_on_success(
+      self, mock_mgr_cls, mock_mat_cls, mock_pg_cls
+  ):
+    """Default flow with successful graph creation reports 'created'."""
+    mock_mgr_cls.return_value.extract_graph.return_value = ExtractedGraph(
+        name="test"
+    )
+    mock_mat_cls.return_value.create_tables.return_value = dict(
+        _ALL_YMGO_TABLES
+    )
+    mock_mat_cls.return_value.materialize.return_value = {}
+    mock_pg_cls.return_value.create_property_graph.return_value = True
+
+    result = build_ontology_graph(
+        session_ids=["sess1"],
+        spec_path=_DEMO_SPEC_PATH,
+        project_id="proj",
+        dataset_id="ds",
+        env="p.d",
+    )
+
+    assert result["property_graph_created"] is True
+    assert result["property_graph_status"] == "created"
+    assert "skipped_reason" not in result
+
+  @patch(
+      "bigquery_agent_analytics.ontology_property_graph"
+      ".OntologyPropertyGraphCompiler"
+  )
+  @patch("bigquery_agent_analytics.ontology_materializer.OntologyMaterializer")
+  @patch("bigquery_agent_analytics.ontology_graph.OntologyGraphManager")
+  def test_property_graph_status_failed_on_compiler_false(
+      self, mock_mgr_cls, mock_mat_cls, mock_pg_cls
+  ):
+    """Default flow where create_property_graph returns False reports
+    'failed' (distinct from 'skipped:user_requested')."""
+    mock_mgr_cls.return_value.extract_graph.return_value = ExtractedGraph(
+        name="test"
+    )
+    mock_mat_cls.return_value.create_tables.return_value = dict(
+        _ALL_YMGO_TABLES
+    )
+    mock_mat_cls.return_value.materialize.return_value = {}
+    mock_pg_cls.return_value.create_property_graph.return_value = False
+
+    result = build_ontology_graph(
+        session_ids=["sess1"],
+        spec_path=_DEMO_SPEC_PATH,
+        project_id="proj",
+        dataset_id="ds",
+        env="p.d",
+    )
+
+    assert result["property_graph_created"] is False
+    assert result["property_graph_status"] == "failed"
+    assert "skipped_reason" not in result