orchestrata_ultimate.refactor.codemod.persistence.repository

Reference › orchestrata_ultimate › refactor › codemod › persistence › repository

The graph persistence for codemod provenance. CodemodGraphRepository is a protocol with save_run, save_pass, save_findings, save_changes, query_runs, query_findings and query_affected_code. BoltSyncCodemodRepository implements it over a SyncGraphRepository (see _types) and writes inside a unit of work. It stores run, pass, finding and code-change nodes, linked by the relationships HAS_CODEMOD_RUN, HAS_PASS, HAS_FINDING, ABOUT, PRODUCED_BY, BREAKING_CHANGE and CHANGED_IN, using the labels and keys of the canonical schema in the knowledge_graph package. save_run, for example, merges the run node and attaches it to its project by slug. InMemoryCodemodRepository keeps the same data in dictionaries and is marked for testing, not production use. The GraphWriterObserver is its only importer.

At a glance

Class diagram

Dependency diagram

API

Classes

CodemodGraphRepository (protocol)

Bases: Protocol · Decorators: traces(spec='110', story='US-110-010', reqs=['REQ-110-028']), runtime_checkable · Line 24

CodemodGraphRepository — structural interface for codemod provenance.

BoltSyncCodemodRepository

Decorators: traces(spec='110', story='US-110-010', reqs=['REQ-110-029']) · Line 43

Bolt-based synchronous implementation of CodemodGraphRepository.

InMemoryCodemodRepository

Decorators: traces(spec='110', story='US-110-010', reqs=['REQ-110-029']) · Line 256

In-memory implementation of CodemodGraphRepository for testing.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/refactor/codemod/persistence/repository.py

"""SPEC-110 codemod engine — graph persistence repository.

Provides CodemodGraphRepository Protocol and BoltSyncCodemodRepository
implementation for writing codemod provenance to the operational graph DB.
"""

from __future__ import annotations

from typing import Protocol, runtime_checkable

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph import canonical_schema as cs
from orchestrata_ultimate.refactor.codemod._types import (
    CodeChange,
    CodemodPass,
    CodemodRun,
    SyncGraphRepository,
)
from orchestrata_ultimate.refactor.codemod.report import Finding


@traces(spec="110", story="US-110-010", reqs=["REQ-110-028"])
@runtime_checkable
class CodemodGraphRepository(Protocol):
    """CodemodGraphRepository — structural interface for codemod provenance."""

    def save_run(self, run: CodemodRun) -> None: ...

    def save_pass(self, run_id: str, pass_: CodemodPass) -> None: ...

    def save_findings(self, run_id: str, findings: list[Finding]) -> None: ...

    def save_changes(self, run_id: str, changes: list[CodeChange]) -> None: ...

    def query_runs(self, project: str, limit: int) -> list[CodemodRun]: ...

    def query_findings(self, run_id: str) -> list[Finding]: ...

    def query_affected_code(self, run_id: str) -> list[str]: ...


@traces(spec="110", story="US-110-010", reqs=["REQ-110-029"])
class BoltSyncCodemodRepository:
    """Bolt-based synchronous implementation of CodemodGraphRepository.

    Uses SyncGraphRepository.unit_of_work() for transactional writes.
    Writes CodemodRun, CodemodPass, CodemodFinding, CodeChange nodes with
    edges: HAS_CODEMOD_RUN, HAS_PASS, HAS_FINDING, ABOUT, PRODUCED_BY,
    BREAKING_CHANGE, CHANGED_IN.
    """

    def __init__(self, repo: SyncGraphRepository) -> None:
        """Wire BoltSyncCodemodRepository's composed collaborators.

        Args:
            repo: Injected SyncGraphRepository for graph DB operations.
        """
        self._repo = repo

    @traces(spec="110", story="US-110-010", reqs=["REQ-110-029"])
    def save_run(self, run: CodemodRun) -> None:
        """MERGE CodemodRun node + HAS_CODEMOD_RUN edge to Project."""
        cypher = f"""
        MERGE (r:{cs.LABEL_CODEMOD_RUN} {{id: $id}})
        SET r.started_at = $started_at,
            r.completed_at = $completed_at,
            r.status = $status,
            r.dry_run = $dry_run,
            r.pass_count = $pass_count,
            r.finding_count = $finding_count,
            r.file_count = $file_count,
            r.duration_ms = $duration_ms
        WITH r
        MATCH (p:{cs.LABEL_PROJECT} {{slug: $project}})
        MERGE (p)-[:{cs.REL_HAS_CODEMOD_RUN}]->(r)
        """
        params = {
            "id": run.run_id,
            "started_at": run.started_at,
            "completed_at": run.completed_at,
            "status": run.status,
            "dry_run": run.dry_run,
            "pass_count": run.pass_count,
            "finding_count": run.finding_count,
            "file_count": run.file_count,
            "duration_ms": run.duration_ms,
            "project": run.project,
        }
        with self._repo.unit_of_work() as uow:
            uow.run(cypher, params)

    @traces(spec="110", story="US-110-010", reqs=["REQ-110-029"])
    def save_pass(self, run_id: str, pass_: CodemodPass) -> None:
        """MERGE CodemodPass node + HAS_PASS edge to CodemodRun."""
        cypher = f"""
        MERGE (cp:{cs.LABEL_CODEMOD_PASS} {{id: $id}})
        SET cp.pass_name = $pass_name,
            cp.pass_type = $pass_type,
            cp.file_count = $file_count,
            cp.duration_ms = $duration_ms
        WITH cp
        MATCH (r:{cs.LABEL_CODEMOD_RUN} {{id: $run_id}})
        MERGE (r)-[:{cs.REL_HAS_PASS}]->(cp)
        """
        params = {
            "id": pass_.pass_id,
            "run_id": run_id,
            "pass_name": pass_.pass_name,
            "pass_type": pass_.pass_type,
            "file_count": pass_.file_count,
            "duration_ms": pass_.duration_ms,
        }
        with self._repo.unit_of_work() as uow:
            uow.run(cypher, params)

    @traces(spec="110", story="US-110-010", reqs=["REQ-110-029"])
    def save_findings(self, run_id: str, findings: list[Finding]) -> None:
        """UNWIND batch MERGE CodemodFinding nodes + HAS_FINDING + ABOUT edges."""
        if not findings:
            return

        cypher = f"""
        UNWIND $findings AS f
        MERGE (cf:{cs.LABEL_CODEMOD_FINDING} {{id: f.id}})
        SET cf.severity = f.severity,
            cf.rule = f.rule,
            cf.path = f.path,
            cf.line = f.line,
            cf.message = f.message
        WITH cf, f
        MATCH (r:{cs.LABEL_CODEMOD_RUN} {{id: $run_id}})
        MERGE (r)-[:{cs.REL_HAS_FINDING}]->(cf)
        WITH cf, f
        OPTIONAL MATCH (m:{cs.LABEL_MODULE} {{path: f.path}})
        FOREACH (_ IN CASE WHEN m IS NOT NULL THEN [1] ELSE [] END |
            MERGE (cf)-[:{cs.REL_ABOUT}]->(m))
        """
        params = {
            "run_id": run_id,
            "findings": [
                {
                    "id": f"{run_id}:finding:{f.rule}:{f.path}:{f.line}",
                    "severity": f.severity,
                    "rule": f.rule,
                    "path": f.path,
                    "line": f.line,
                    "message": f.message,
                }
                for f in findings
            ],
        }
        with self._repo.unit_of_work() as uow:
            uow.run(cypher, params)

    @traces(spec="110", story="US-110-010", reqs=["REQ-110-029"])
    def save_changes(self, run_id: str, changes: list[CodeChange]) -> None:
        """UNWIND batch MERGE CodeChange nodes + PRODUCED_BY + CHANGED_IN edges."""
        if not changes:
            return

        cypher = f"""
        UNWIND $changes AS c
        MERGE (cc:{cs.LABEL_CHANGE}:CodeChange {{id: c.id}})
        SET cc.pass_name = c.pass_name,
            cc.description = c.description
        WITH cc, c
        MATCH (cp:{cs.LABEL_CODEMOD_PASS} {{id: c.pass_id}})
        MERGE (cc)-[:{cs.REL_PRODUCED_BY}]->(cp)
        WITH cc, c
        OPTIONAL MATCH (entity {{id: c.entity_id}})
        FOREACH (_ IN CASE WHEN entity IS NOT NULL THEN [1] ELSE [] END |
            MERGE (entity)-[:{cs.REL_CHANGED_IN}]->(cc))
        """
        params = {
            "changes": [
                {
                    "id": f"{run_id}:change:{c.pass_name}:{c.path}",
                    "pass_id": f"{run_id}:pass:{c.pass_name}",
                    "pass_name": c.pass_name,
                    "description": c.description,
                    "entity_id": str(c.entity_id) if c.entity_id is not None else None,
                }
                for c in changes
            ]
        }
        with self._repo.unit_of_work() as uow:
            uow.run(cypher, params)

    def query_runs(self, project: str, limit: int = 10) -> list[CodemodRun]:
        """Query CodemodRun nodes for a project."""
        cypher = f"""
        MATCH (p:{cs.LABEL_PROJECT} {{slug: $project}})-[:{cs.REL_HAS_CODEMOD_RUN}]->(r:{cs.LABEL_CODEMOD_RUN})
        RETURN r.id AS id, r.started_at AS started_at, r.completed_at AS completed_at,
               r.status AS status, r.dry_run AS dry_run, r.pass_count AS pass_count,
               r.finding_count AS finding_count, r.file_count AS file_count,
               r.duration_ms AS duration_ms, r.project AS project
        ORDER BY r.started_at DESC
        LIMIT $limit
        """
        params = {"project": project, "limit": limit}
        with self._repo.unit_of_work() as uow:
            result = uow.run(cypher, params)
            return [
                CodemodRun(
                    run_id=record["id"],
                    started_at=record["started_at"],
                    completed_at=record["completed_at"],
                    status=record["status"],
                    dry_run=record["dry_run"],
                    pass_count=record["pass_count"],
                    finding_count=record["finding_count"],
                    file_count=record["file_count"],
                    duration_ms=record["duration_ms"],
                    project=record["project"],
                )
                for record in result
            ]

    def query_findings(self, run_id: str) -> list[Finding]:
        """Query CodemodFinding nodes for a run."""
        cypher = f"""
        MATCH (r:{cs.LABEL_CODEMOD_RUN} {{id: $run_id}})-[:{cs.REL_HAS_FINDING}]->(cf:{cs.LABEL_CODEMOD_FINDING})
        RETURN cf.severity AS severity, cf.rule AS rule, cf.path AS path,
               cf.line AS line, cf.message AS message
        """
        params = {"run_id": run_id}
        with self._repo.unit_of_work() as uow:
            result = uow.run(cypher, params)
            return [
                Finding(
                    severity=record["severity"],
                    rule=record["rule"],
                    path=record["path"],
                    line=record["line"],
                    message=record["message"],
                    proof_tree=None,
                )
                for record in result
            ]

    def query_affected_code(self, run_id: str) -> list[str]:
        """Query entity IDs affected by a run's changes."""
        cypher = f"""
        MATCH (r:{cs.LABEL_CODEMOD_RUN} {{id: $run_id}})-[:{cs.REL_HAS_PASS}]->(cp:{cs.LABEL_CODEMOD_PASS})
        MATCH (cc:{cs.LABEL_CHANGE}:CodeChange)-[:{cs.REL_PRODUCED_BY}]->(cp)
        MATCH (entity)-[:{cs.REL_CHANGED_IN}]->(cc)
        RETURN DISTINCT entity.id AS entity_id
        """
        params = {"run_id": run_id}
        with self._repo.unit_of_work() as uow:
            result = uow.run(cypher, params)
            return [record["entity_id"] for record in result if record["entity_id"]]


@traces(spec="110", story="US-110-010", reqs=["REQ-110-029"])
class InMemoryCodemodRepository:
    """In-memory implementation of CodemodGraphRepository for testing.

    Stores codemod provenance in memory dictionaries. Not for production use.
    """

    def __init__(self) -> None:
        """Initialize empty in-memory storage."""
        self._runs: dict[str, CodemodRun] = {}
        self._passes: dict[str, list[CodemodPass]] = {}
        self._findings: dict[str, list[Finding]] = {}
        self._changes: dict[str, list[CodeChange]] = {}

    def save_run(self, run: CodemodRun) -> None:
        """Store CodemodRun in memory."""
        self._runs[run.run_id] = run

    def save_pass(self, run_id: str, pass_: CodemodPass) -> None:
        """Store CodemodPass in memory."""
        if run_id not in self._passes:
            self._passes[run_id] = []
        self._passes[run_id].append(pass_)

    def save_findings(self, run_id: str, findings: list[Finding]) -> None:
        """Store findings in memory."""
        if run_id not in self._findings:
            self._findings[run_id] = []
        self._findings[run_id].extend(findings)

    def save_changes(self, run_id: str, changes: list[CodeChange]) -> None:
        """Store changes in memory."""
        if run_id not in self._changes:
            self._changes[run_id] = []
        self._changes[run_id].extend(changes)

    def query_runs(self, project: str, limit: int = 10) -> list[CodemodRun]:
        """Query runs by project."""
        runs = [r for r in self._runs.values() if r.project == project]
        runs.sort(key=lambda r: r.started_at, reverse=True)
        return runs[:limit]

    def query_findings(self, run_id: str) -> list[Finding]:
        """Query findings by run_id."""
        return self._findings.get(run_id, [])

    def query_affected_code(self, run_id: str) -> list[str]:
        """Query affected entity IDs by run_id."""
        changes = self._changes.get(run_id, [])
        return [str(c.entity_id) for c in changes if c.entity_id is not None]