Source code for curate_ipsum.storage.incremental

"""
Incremental update engine for graph persistence.

Detects which files changed since the last extraction and updates
only the affected graph nodes/edges, avoiding full re-extraction.

Uses SHA-256 file hashing to detect changes. The hash map is persisted
via the GraphStore's file_hashes table.

Decision: D-015
"""

from __future__ import annotations

import hashlib
import logging
import time
from dataclasses import dataclass, field
from pathlib import Path

from curate_ipsum.storage.graph_store import GraphStore

LOG = logging.getLogger("storage.incremental")


[docs] @dataclass class ChangeSet: """Files that changed since last extraction.""" added: list[str] = field(default_factory=list) modified: list[str] = field(default_factory=list) removed: list[str] = field(default_factory=list) @property def has_changes(self) -> bool: return bool(self.added or self.modified or self.removed) @property def total_changed(self) -> int: return len(self.added) + len(self.modified) + len(self.removed)
[docs] def to_dict(self) -> dict: return { "added": self.added, "modified": self.modified, "removed": self.removed, "total_changed": self.total_changed, }
[docs] @dataclass class UpdateResult: """Result of an incremental graph update.""" added_nodes: int = 0 removed_nodes: int = 0 modified_files: int = 0 total_files_scanned: int = 0 duration_ms: int = 0 change_set: ChangeSet | None = None full_rebuild: bool = False
[docs] def to_dict(self) -> dict: return { "added_nodes": self.added_nodes, "removed_nodes": self.removed_nodes, "modified_files": self.modified_files, "total_files_scanned": self.total_files_scanned, "duration_ms": self.duration_ms, "change_set": self.change_set.to_dict() if self.change_set else None, "full_rebuild": self.full_rebuild, }
[docs] class IncrementalEngine: """ Detects file changes and performs incremental graph updates. Workflow: 1. Compute current file hashes for all matching files 2. Compare with stored hashes → ChangeSet 3. For removed files: delete nodes/edges 4. For added/modified files: re-extract and merge 5. Update stored file hashes """ def __init__(self, store: GraphStore) -> None: self._store = store
[docs] @staticmethod def compute_file_hashes( directory: Path, pattern: str = "**/*.py", ) -> dict[str, str]: """ Compute SHA-256 hashes for all files matching pattern. Args: directory: Root directory to scan pattern: Glob pattern for files Returns: Dict mapping relative file paths to their SHA-256 hex digests """ hashes: dict[str, str] = {} dir_path = Path(directory) for file_path in dir_path.glob(pattern): if file_path.is_file(): try: content = file_path.read_bytes() digest = hashlib.sha256(content).hexdigest() rel_path = str(file_path.relative_to(dir_path)) hashes[rel_path] = digest except (OSError, PermissionError) as exc: LOG.warning("Cannot hash %s: %s", file_path, exc) return hashes
[docs] def detect_changes( self, project_id: str, current_hashes: dict[str, str], ) -> ChangeSet: """ Compare current file hashes with stored hashes to find changes. Args: project_id: Project identifier current_hashes: Current file → hash mapping Returns: ChangeSet with added, modified, and removed files """ stored_hashes = self._store.get_file_hashes(project_id) current_files = set(current_hashes.keys()) stored_files = set(stored_hashes.keys()) added = sorted(current_files - stored_files) removed = sorted(stored_files - current_files) modified = sorted(f for f in current_files & stored_files if current_hashes[f] != stored_hashes[f]) change_set = ChangeSet(added=added, modified=modified, removed=removed) LOG.info( "Change detection for %s: %d added, %d modified, %d removed", project_id, len(added), len(modified), len(removed), ) return change_set
[docs] def update_graph( self, project_id: str, directory: Path, pattern: str = "**/*.py", extractor_func=None, ) -> UpdateResult: """ Perform an incremental graph update. Args: project_id: Project identifier directory: Root directory of the project pattern: File glob pattern extractor_func: Optional callable(file_path) → (nodes, edges) for extraction. If None, only file hash tracking and node deletion are performed. Returns: UpdateResult with counts of changes made """ start = time.monotonic() result = UpdateResult() # Step 1: Compute current hashes current_hashes = self.compute_file_hashes(directory, pattern) result.total_files_scanned = len(current_hashes) # Step 2: Detect changes change_set = self.detect_changes(project_id, current_hashes) result.change_set = change_set if not change_set.has_changes: result.duration_ms = int((time.monotonic() - start) * 1000) LOG.info("No changes detected for project %s", project_id) return result # Step 3: Remove nodes for deleted files for file_path in change_set.removed: removed = self._store.delete_nodes_by_file(file_path, project_id) result.removed_nodes += removed # Step 4: Remove nodes for modified files (will be re-extracted) for file_path in change_set.modified: removed = self._store.delete_nodes_by_file(file_path, project_id) result.removed_nodes += removed # Step 5: Re-extract added and modified files files_to_extract = change_set.added + change_set.modified result.modified_files = len(files_to_extract) if extractor_func and files_to_extract: for file_path in files_to_extract: try: full_path = Path(directory) / file_path nodes, edges = extractor_func(str(full_path)) for node_data in nodes: self._store.store_node(node_data, project_id) result.added_nodes += 1 for edge_data in edges: self._store.store_edge(edge_data, project_id) except Exception as exc: LOG.warning("Failed to extract %s: %s", file_path, exc) # Step 6: Update stored file hashes # Remove hashes for deleted files updated_hashes = dict(current_hashes.items()) self._store.set_file_hashes(project_id, updated_hashes) result.duration_ms = int((time.monotonic() - start) * 1000) LOG.info( "Incremental update for %s: +%d nodes, -%d nodes, %d files, %dms", project_id, result.added_nodes, result.removed_nodes, result.modified_files, result.duration_ms, ) return result
[docs] def force_full_rebuild( self, project_id: str, graph, directory: Path, pattern: str = "**/*.py", ) -> UpdateResult: """ Force a complete graph rebuild (drop all + store full graph). Args: project_id: Project identifier graph: The complete CallGraph to persist directory: Root directory for file hash computation pattern: File glob pattern Returns: UpdateResult marked as full_rebuild """ start = time.monotonic() # Store the entire graph (replaces existing) self._store.store_graph(graph, project_id) # Compute and store file hashes hashes = self.compute_file_hashes(directory, pattern) self._store.set_file_hashes(project_id, hashes) result = UpdateResult( added_nodes=len(graph.nodes), removed_nodes=0, modified_files=len(hashes), total_files_scanned=len(hashes), duration_ms=int((time.monotonic() - start) * 1000), full_rebuild=True, ) LOG.info( "Full rebuild for %s: %d nodes, %d files, %dms", project_id, result.added_nodes, result.modified_files, result.duration_ms, ) return result