DataOps-platform/app/core/knowledge/impact.py

from __future__ import annotations

from collections import defaultdict, deque
from collections.abc import Iterable, Set

from app.core.knowledge.contracts import ImpactResult, KnowledgeDependencyDraft


def _object_key(point_key: str) -> str:
    parts = point_key.split("/", 2)
    return "/".join(parts[:2]) if len(parts) >= 2 else point_key


def propagate_impact(
    changed_point_keys: Set[str],
    dependencies: Iterable[KnowledgeDependencyDraft],
    *,
    allowed_relations: Set[str],
    max_hops: int,
    max_points: int,
    max_objects: int,
) -> ImpactResult:
    if max_hops < 0 or max_points < 0 or max_objects < 1:
        raise ValueError("impact propagation limits are invalid")

    reverse: dict[str, list[str]] = defaultdict(list)
    for dependency in dependencies:
        if dependency.source not in {"governance", "deterministic_rule"}:
            continue
        if dependency.relation_type not in allowed_relations:
            continue
        reverse[dependency.to_point_key].append(dependency.from_point_key)
    for values in reverse.values():
        values.sort()

    roots = sorted(set(changed_point_keys))
    visited = set(roots)
    queue = deque((root, 0, (root,)) for root in roots)
    impacted: set[str] = set()
    paths: dict[str, tuple[str, ...]] = {}
    objects: set[str] = set()
    truncated = False

    while queue:
        current, hop, path = queue.popleft()
        if hop >= max_hops:
            if reverse.get(current):
                truncated = True
            continue
        for dependent in reverse.get(current, ()):
            if dependent in visited:
                continue
            candidate_object = _object_key(dependent)
            if len(impacted) >= max_points or (
                candidate_object not in objects and len(objects) >= max_objects
            ):
                truncated = True
                continue
            visited.add(dependent)
            impacted.add(dependent)
            objects.add(candidate_object)
            dependent_path = (*path, dependent)
            paths[dependent] = dependent_path
            queue.append((dependent, hop + 1, dependent_path))

    return ImpactResult(
        impacted_point_keys=frozenset(impacted),
        paths=paths,
        truncated=truncated,
    )