orchestrata_ultimate.knowledge_graph._sync_queries
Reference › orchestrata_ultimate › knowledge_graph › _sync_queries
The read-side half of sync, extracted so the writer stays smaller: QueryMixin is mixed into GraphSyncer and contains only read-only queries against the operational graph (orchkg on orchgraph) through the host's _repo, a SyncGraphRepository (it was a raw Bolt driver before). Code-health queries are query_methods_without_tests (no TESTED_BY edge), query_dead_code (private methods with no incoming CALLS), query_deep_inheritance, query_untested_entrypoints (called but untested) and the two lookups query_class_methods and query_callers. get_stats counts nodes and relationships per label and type, and verify runs verification queries for a project and reports pass or fail. Limits default to 20. The node and edge builders stay in sync because they share an implicit id-construction contract.
At a glance
| Source | src/orchestrata_ultimate/knowledge_graph/_sync_queries.py |
| Lines of code | 283 |
| Classes | 1 |
| Functions | 0 |
| Imports (internal) | none |
| Imported by | orchestrata_ultimate.knowledge_graph.sync |
| Tests that import it | none |
Class diagram
Dependency diagram
API
Classes
QueryMixin
Line 19
Read-only Neo4j query/inspection methods for GraphSyncer.
| Attribute | Type | Default |
|---|---|---|
_repo |
Any |
| Method | Signature | Summary |
|---|---|---|
query_methods_without_tests |
query_methods_without_tests(limit: int = 20) -> list[dict[str, Any]] |
Return methods with no TESTED_BY relationship. |
query_dead_code |
query_dead_code(limit: int = 20) -> list[dict[str, Any]] |
Return private methods with no incoming CALLS edges. |
query_deep_inheritance |
query_deep_inheritance(limit: int = 20) -> list[dict[str, Any]] |
Return classes with the longest inheritance chains. |
query_untested_entrypoints |
query_untested_entrypoints(limit: int = 20) -> list[dict[str, Any]] |
Return methods that are called but not tested. |
query_class_methods |
query_class_methods(class_name: str) -> list[dict[str, Any]] |
Return all methods for a class by name. |
query_callers |
query_callers(method_name: str) -> list[dict[str, Any]] |
Return methods/functions that call a method by name. |
get_stats |
get_stats() -> dict[str, dict[str, int]] |
Node/relationship counts per label/type. |
verify |
verify(project: str) -> dict[str, Any] |
Run verification queries and report pass/fail. |
Source
# mypy: ignore-errors
"""Read-side query mixin for GraphSyncer.
Cycle-free read-only slice extracted from ``knowledge_graph.sync``. All
methods are read-only over ``self._repo`` (backlog #163 phase 5b — a
``SyncGraphRepository``, was a raw ``neo4j.Driver`` before) and are mixed
into ``GraphSyncer``. Node/edge sync builders stay in ``sync`` because they
share an implicit id-construction contract.
"""
from __future__ import annotations
import logging
from typing import Any
_logger = logging.getLogger(__name__)
class QueryMixin:
"""Read-only Neo4j query/inspection methods for GraphSyncer.
``self._repo`` is declared (not assigned) here as a structural contract
for the methods below — the concrete host class (``GraphSyncer``) owns
the real attribute. Declared as ``Any`` rather than
``SyncGraphRepository | None``: the host class's narrower
``BoltSyncRepository | None`` would otherwise be an incompatible
(invariant) mutable-attribute override.
"""
_repo: Any
# ── Query Methods ──────────────────────────────────────────────────
def query_methods_without_tests(self, limit: int = 20) -> list[dict[str, Any]]:
"""Return methods with no TESTED_BY relationship.
Args:
limit: Maximum number of results.
Returns:
List of method property dicts.
"""
if not self._repo:
raise RuntimeError("Not connected. Call connect() first.")
try:
result = self._repo.execute_read(
"MATCH (m:Method) "
"WHERE NOT (m)-[:TESTED_BY]->(:Test) "
"RETURN m {.name, .qualified_name, .module_path, .line} AS method "
"LIMIT $limit",
{"limit": limit},
)
return [record["method"] for record in result]
except Exception as exc:
_logger.error("query_methods_without_tests failed: %s", exc)
return []
def query_dead_code(self, limit: int = 20) -> list[dict[str, Any]]:
"""Return private methods with no incoming CALLS edges.
Args:
limit: Maximum number of results.
Returns:
List of method property dicts.
"""
if not self._repo:
raise RuntimeError("Not connected. Call connect() first.")
try:
result = self._repo.execute_read(
"MATCH (m:Method) "
"WHERE m.name STARTS WITH '_' "
"AND NOT (:Method)-[:CALLS]->(m) "
"AND NOT (:Function)-[:CALLS]->(m) "
"RETURN m {.name, .qualified_name, .module_path, .line} AS method "
"LIMIT $limit",
{"limit": limit},
)
return [record["method"] for record in result]
except Exception as exc:
_logger.error("query_dead_code failed: %s", exc)
return []
def query_deep_inheritance(self, limit: int = 20) -> list[dict[str, Any]]:
"""Return classes with the longest inheritance chains.
Args:
limit: Maximum number of results.
Returns:
List of dicts with class info and chain_length.
"""
if not self._repo:
raise RuntimeError("Not connected. Call connect() first.")
try:
result = self._repo.execute_read(
"MATCH path = (c:Class)-[:INHERITS*]->(base:Class) "
"WITH c, length(path) AS chain_length "
"RETURN c {.name, .qualified_name, .module_path} AS class, "
"chain_length ORDER BY chain_length DESC LIMIT $limit",
{"limit": limit},
)
return [
{
"class": record["class"],
"chain_length": record["chain_length"],
}
for record in result
]
except Exception as exc:
_logger.error("query_deep_inheritance failed: %s", exc)
return []
def query_untested_entrypoints(self, limit: int = 20) -> list[dict[str, Any]]:
"""Return methods that are called but not tested.
Args:
limit: Maximum number of results.
Returns:
List of method property dicts.
"""
if not self._repo:
raise RuntimeError("Not connected. Call connect() first.")
try:
result = self._repo.execute_read(
"MATCH (m:Method) "
"WHERE ((:Method)-[:CALLS]->(m) OR (:Function)-[:CALLS]->(m)) "
"AND NOT (m)-[:TESTED_BY]->(:Test) "
"RETURN m {.name, .qualified_name, .module_path, .line} AS method "
"LIMIT $limit",
{"limit": limit},
)
return [record["method"] for record in result]
except Exception as exc:
_logger.error("query_untested_entrypoints failed: %s", exc)
return []
def query_class_methods(self, class_name: str) -> list[dict[str, Any]]:
"""Return all methods for a class by name.
Args:
class_name: Name of the class (matches :Class name).
Returns:
List of method property dicts ordered by line number.
"""
if not self._repo:
raise RuntimeError("Not connected. Call connect() first.")
try:
result = self._repo.execute_read(
"MATCH (c:Class {name: $class_name})-[:DEFINES]->(m:Method) "
"RETURN m {.name, .qualified_name, .line, .is_async, "
".return_type} AS method ORDER BY m.line",
{"class_name": class_name},
)
return [record["method"] for record in result]
except Exception as exc:
_logger.error("query_class_methods failed: %s", exc)
return []
def query_callers(self, method_name: str) -> list[dict[str, Any]]:
"""Return methods/functions that call a method by name.
Args:
method_name: Name of the called method.
Returns:
List of caller property dicts.
"""
if not self._repo:
raise RuntimeError("Not connected. Call connect() first.")
try:
result = self._repo.execute_read(
"MATCH (caller)-[:CALLS]->(m:Method {name: $method_name}) "
"RETURN caller {.name, .qualified_name, .module_path} AS caller",
{"method_name": method_name},
)
return [record["caller"] for record in result]
except Exception as exc:
_logger.error("query_callers failed: %s", exc)
return []
# ── Stats ──────────────────────────────────────────────────────────
def get_stats(self) -> dict[str, dict[str, int]]:
"""Node/relationship counts per label/type.
Plain Cypher, not ``CALL db.labels()`` / ``db.relationshipTypes()``.
Those are Neo4j-only procedures: Memgraph answers "There is no
procedure named 'db.labels'", so this method was unusable against
orchgraph — the cluster Step 3.3 re-derives the AST subgraph onto and
that Phase 4 cuts over to. Aggregating with ``UNWIND labels(n)``
instead runs on both engines and, as a bonus, returns every count in
one pass rather than one query per label.
Returns:
Dict with "nodes" and "relationships" counts.
"""
if not self._repo:
raise RuntimeError("Not connected. Call connect() first.")
repo = self._repo
node_stats = {
row["label"]: row["cnt"]
for row in repo.execute_read(
"MATCH (n) UNWIND labels(n) AS label "
"RETURN label, count(*) AS cnt ORDER BY label"
)
}
rel_stats = {
row["rel_type"]: row["cnt"]
for row in repo.execute_read(
"MATCH ()-[r]->() "
"RETURN type(r) AS rel_type, count(*) AS cnt ORDER BY rel_type"
)
}
return {"nodes": node_stats, "relationships": rel_stats}
# ── Verify ─────────────────────────────────────────────────────────
def verify(self, project: str) -> dict[str, Any]:
"""Run verification queries and report pass/fail.
Args:
project: Project name to verify.
Returns:
Dict with verification results and "ok" flag.
"""
if not self._repo:
raise RuntimeError("Not connected. Call connect() first.")
repo = self._repo
results: dict[str, Any] = {}
# Check project exists
result = repo.execute_read(
"MATCH (p:Project {id: $project}) RETURN count(p) AS cnt",
{"project": project},
)
results["project_exists"] = result[0]["cnt"]
# Check modules linked to project
result = repo.execute_read(
"MATCH (p:Project {id: $project})-[:CONTAINS*0..]->(m:Module) "
"RETURN count(m) AS cnt",
{"project": project},
)
results["module_count"] = result[0]["cnt"]
# Check classes exist
result = repo.execute_read("MATCH (c:Class) RETURN count(c) AS cnt")
results["class_count"] = result[0]["cnt"]
# Check functions exist
result = repo.execute_read("MATCH (f:Function) RETURN count(f) AS cnt")
results["function_count"] = result[0]["cnt"]
# Check methods exist
result = repo.execute_read("MATCH (m:Method) RETURN count(m) AS cnt")
results["method_count"] = result[0]["cnt"]
# Check DEFINES edges
result = repo.execute_read(
"MATCH (:Class)-[:DEFINES]->(:Method) RETURN count(*) AS cnt"
)
results["has_method_edges"] = result[0]["cnt"]
# Check INHERITS edges
result = repo.execute_read(
"MATCH (:Class)-[:INHERITS]->(:Class) RETURN count(*) AS cnt"
)
results["inheritance_edges"] = result[0]["cnt"]
# Determine overall ok
results["ok"] = (
results.get("project_exists", 0) >= 1
and results.get("module_count", 0) >= 1
and results.get("class_count", 0) >= 1
)
return results