# -*- coding: utf-8 -*- """ Abhängigkeitsgraph für Plugins. Verwaltet die Abhängigkeitsbeziehungen zwischen Plugins und ermöglicht topologische Sortierung. """ from __future__ import annotations from collections import defaultdict from dataclasses import dataclass, field from enum import IntEnum, auto from typing import Any, Iterator class CycleError(Exception): """Wird ausgelöst, wenn ein Zyklus erkannt wird.""" def __init__(self, cycle: list[str]) -> None: self.cycle = cycle super().__init__(f"Zyklische Abhängigkeit erkannt: {' -> '.join(cycle)}") class NodeState(IntEnum): """Zustand eines Knotens während der Traversierung.""" UNVISITED = auto() VISITING = auto() VISITED = auto() @dataclass class DependencyNode: """ Ein Knoten im Abhängigkeitsgraphen. Attributes: name: Plugin-Name version: Plugin-Version dependencies: Liste der Abhängigkeiten (Name -> Version-Requirement) optional_dependencies: Optionale Abhängigkeiten metadata: Zusätzliche Metadaten """ name: str version: str = "1.0.0" dependencies: dict[str, str] = field(default_factory=dict) optional_dependencies: dict[str, str] = field(default_factory=dict) metadata: dict[str, Any] = field(default_factory=dict) def has_dependency(self, name: str) -> bool: """Prüft, ob eine Abhängigkeit existiert.""" return name in self.dependencies or name in self.optional_dependencies def get_all_dependencies(self) -> dict[str, str]: """Gibt alle Abhängigkeiten zurück (required + optional).""" result = self.dependencies.copy() result.update(self.optional_dependencies) return result def to_dict(self) -> dict[str, Any]: """Konvertiert zu Dictionary.""" return { "name": self.name, "version": self.version, "dependencies": self.dependencies.copy(), "optional_dependencies": self.optional_dependencies.copy(), "metadata": self.metadata.copy(), } class DependencyGraph: """ Gerichteter Graph für Plugin-Abhängigkeiten. Ermöglicht: - Hinzufügen/Entfernen von Knoten - Zykluserkennung - Topologische Sortierung - Suche nach abhängigen Plugins Example: graph = DependencyGraph() # Knoten hinzufügen graph.add_node(DependencyNode( name="plugin_a", version="1.0.0", dependencies={"plugin_b": "^1.0.0"} )) graph.add_node(DependencyNode( name="plugin_b", version="1.2.0" )) # Ladereihenfolge berechnen order = graph.get_load_order() # ['plugin_b', 'plugin_a'] # Abhängige finden dependents = graph.get_dependents("plugin_b") # ['plugin_a'] # Als DOT exportieren dot = graph.to_dot() """ def __init__(self) -> None: """Initialisiert den Graphen.""" self._nodes: dict[str, DependencyNode] = {} self._edges: dict[str, set[str]] = defaultdict(set) # from -> set(to) self._reverse_edges: dict[str, set[str]] = defaultdict(set) # to -> set(from) @property def node_count(self) -> int: """Anzahl Knoten.""" return len(self._nodes) @property def edge_count(self) -> int: """Anzahl Kanten.""" return sum(len(targets) for targets in self._edges.values()) def add_node(self, node: DependencyNode) -> None: """ Fügt einen Knoten hinzu. Args: node: Der hinzuzufügende Knoten Raises: ValueError: Wenn der Knoten bereits existiert """ if node.name in self._nodes: raise ValueError(f"Knoten existiert bereits: {node.name}") self._nodes[node.name] = node # Kanten für Abhängigkeiten hinzufügen for dep_name in node.dependencies.keys(): self._edges[node.name].add(dep_name) self._reverse_edges[dep_name].add(node.name) def update_node(self, node: DependencyNode) -> None: """ Aktualisiert einen existierenden Knoten. Args: node: Der zu aktualisierende Knoten """ if node.name in self._nodes: self.remove_node(node.name) self.add_node(node) def remove_node(self, name: str) -> DependencyNode | None: """ Entfernt einen Knoten. Args: name: Knotenname Returns: Der entfernte Knoten oder None """ node = self._nodes.pop(name, None) if node is None: return None # Kanten entfernen if name in self._edges: for target in self._edges[name]: self._reverse_edges[target].discard(name) del self._edges[name] if name in self._reverse_edges: for source in self._reverse_edges[name]: self._edges[source].discard(name) del self._reverse_edges[name] return node def get_node(self, name: str) -> DependencyNode | None: """Gibt einen Knoten zurück.""" return self._nodes.get(name) def has_node(self, name: str) -> bool: """Prüft, ob ein Knoten existiert.""" return name in self._nodes def get_dependencies(self, name: str) -> set[str]: """ Gibt die direkten Abhängigkeiten eines Knotens zurück. Args: name: Knotenname Returns: Set der Abhängigkeiten """ return self._edges.get(name, set()).copy() def get_dependents(self, name: str) -> set[str]: """ Gibt alle Knoten zurück, die von diesem abhängen. Args: name: Knotenname Returns: Set der abhängigen Knoten """ return self._reverse_edges.get(name, set()).copy() def get_all_dependencies(self, name: str) -> set[str]: """ Gibt alle transitiven Abhängigkeiten zurück. Args: name: Knotenname Returns: Set aller direkten und indirekten Abhängigkeiten """ result: set[str] = set() stack = list(self._edges.get(name, set())) while stack: dep = stack.pop() if dep not in result: result.add(dep) stack.extend(self._edges.get(dep, set())) return result def get_all_dependents(self, name: str) -> set[str]: """ Gibt alle Knoten zurück, die direkt oder indirekt abhängig sind. Args: name: Knotenname Returns: Set aller abhängigen Knoten """ result: set[str] = set() stack = list(self._reverse_edges.get(name, set())) while stack: dep = stack.pop() if dep not in result: result.add(dep) stack.extend(self._reverse_edges.get(dep, set())) return result def has_cycle(self) -> bool: """ Prüft, ob der Graph einen Zyklus enthält. Returns: True wenn ein Zyklus existiert """ try: self._detect_cycle() return False except CycleError: return True def find_cycle(self) -> list[str] | None: """ Findet einen Zyklus im Graphen. Returns: Liste der Knoten im Zyklus oder None """ try: self._detect_cycle() return None except CycleError as e: return e.cycle def _detect_cycle(self) -> None: """Wirft CycleError wenn ein Zyklus gefunden wird.""" state: dict[str, NodeState] = { name: NodeState.UNVISITED for name in self._nodes } path: list[str] = [] def visit(name: str) -> None: if state.get(name, NodeState.UNVISITED) == NodeState.VISITED: return if state.get(name, NodeState.UNVISITED) == NodeState.VISITING: # Zyklus gefunden cycle_start = path.index(name) raise CycleError(path[cycle_start:] + [name]) state[name] = NodeState.VISITING path.append(name) for dep in self._edges.get(name, set()): visit(dep) path.pop() state[name] = NodeState.VISITED for name in self._nodes: if state[name] == NodeState.UNVISITED: visit(name) def get_load_order(self) -> list[str]: """ Berechnet die Ladereihenfolge (topologische Sortierung). Abhängigkeiten kommen vor den abhängigen Plugins. Returns: Liste der Knotennamen in Ladereihenfolge Raises: CycleError: Bei zyklischen Abhängigkeiten """ self._detect_cycle() # Kahn's Algorithmus in_degree = {name: 0 for name in self._nodes} for sources in self._edges.values(): for target in sources: if target in in_degree: in_degree[target] += 1 # Starte mit Knoten ohne eingehende Kanten queue = [name for name, degree in in_degree.items() if degree == 0] result: list[str] = [] while queue: # Sortiert für deterministische Reihenfolge queue.sort() name = queue.pop(0) result.append(name) for dep in self._reverse_edges.get(name, set()): if dep in in_degree: in_degree[dep] -= 1 if in_degree[dep] == 0: queue.append(dep) # Umkehren: Abhängigkeiten zuerst result.reverse() return result def get_unload_order(self) -> list[str]: """ Berechnet die Entlade-Reihenfolge. Umgekehrte Ladereihenfolge - abhängige Plugins zuerst. Returns: Liste der Knotennamen in Entlade-Reihenfolge """ return list(reversed(self.get_load_order())) def get_missing_dependencies(self) -> dict[str, list[str]]: """ Findet alle fehlenden Abhängigkeiten. Returns: Dictionary mit Knoten -> Liste fehlender Abhängigkeiten """ missing: dict[str, list[str]] = {} for name, node in self._nodes.items(): node_missing = [] for dep_name in node.dependencies.keys(): if dep_name not in self._nodes: node_missing.append(dep_name) if node_missing: missing[name] = node_missing return missing def get_root_nodes(self) -> list[str]: """ Gibt alle Knoten ohne Abhängigkeiten zurück. Returns: Liste der Root-Knoten """ return [ name for name, node in self._nodes.items() if not node.dependencies ] def get_leaf_nodes(self) -> list[str]: """ Gibt alle Knoten ohne abhängige zurück. Returns: Liste der Blatt-Knoten """ return [ name for name in self._nodes if not self._reverse_edges.get(name) ] def to_dot(self, highlight: set[str] | None = None) -> str: """ Exportiert den Graphen als DOT-Format. Args: highlight: Optional zu hervorhebende Knoten Returns: DOT-String für Graphviz """ highlight = highlight or set() lines = ["digraph dependencies {"] lines.append(' rankdir=LR;') lines.append(' node [shape=box];') # Knoten for name, node in self._nodes.items(): style = 'style=filled fillcolor=yellow' if name in highlight else '' label = f"{name}\\n{node.version}" lines.append(f' "{name}" [label="{label}" {style}];') # Kanten for source, targets in self._edges.items(): for target in sorted(targets): style = 'style=dashed' if target not in self._nodes else '' lines.append(f' "{source}" -> "{target}" [{style}];') lines.append("}") return "\n".join(lines) def to_mermaid(self) -> str: """ Exportiert den Graphen als Mermaid-Format. Returns: Mermaid-String """ lines = ["graph LR"] for name, node in self._nodes.items(): lines.append(f' {name}["{name} v{node.version}"]') for source, targets in self._edges.items(): for target in sorted(targets): lines.append(f' {source} --> {target}') return "\n".join(lines) def __contains__(self, name: str) -> bool: return name in self._nodes def __iter__(self) -> Iterator[DependencyNode]: return iter(self._nodes.values()) def __len__(self) -> int: return len(self._nodes)