# -*- coding: utf-8 -*- """ Abhängigkeitsauflösung für Plugins. Löst Abhängigkeiten auf und berechnet die optimale Plugin-Ladereihenfolge. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any from trixy_core.plugins.dependencies.graph import ( DependencyGraph, DependencyNode, CycleError, ) from trixy_core.plugins.versioning.semver import parse_version from trixy_core.plugins.versioning.compatibility import ( VersionChecker, CompatibilityResult, ) class ResolutionError(Exception): """Fehler bei der Abhängigkeitsauflösung.""" def __init__( self, message: str, missing: list[str] | None = None, incompatible: dict[str, str] | None = None, cycle: list[str] | None = None, ) -> None: super().__init__(message) self.missing = missing or [] self.incompatible = incompatible or {} self.cycle = cycle @dataclass class Dependency: """ Eine Plugin-Abhängigkeit. Attributes: name: Name des Plugins version_requirement: Versionsanforderung optional: Ist optional reason: Grund für die Abhängigkeit """ name: str version_requirement: str = "*" optional: bool = False reason: str = "" def to_dict(self) -> dict[str, Any]: """Konvertiert zu Dictionary.""" return { "name": self.name, "version_requirement": self.version_requirement, "optional": self.optional, "reason": self.reason, } @dataclass class ResolutionResult: """ Ergebnis einer Abhängigkeitsauflösung. Attributes: success: Auflösung erfolgreich load_order: Ladereihenfolge resolved: Aufgelöste Abhängigkeiten missing: Fehlende Abhängigkeiten incompatible: Inkompatible Versionen warnings: Warnungen """ success: bool load_order: list[str] = field(default_factory=list) resolved: dict[str, str] = field(default_factory=dict) # name -> version missing: list[str] = field(default_factory=list) incompatible: dict[str, CompatibilityResult] = field(default_factory=dict) warnings: list[str] = field(default_factory=list) def to_dict(self) -> dict[str, Any]: """Konvertiert zu Dictionary.""" return { "success": self.success, "load_order": self.load_order, "resolved": self.resolved, "missing": self.missing, "incompatible": { k: v.to_dict() for k, v in self.incompatible.items() }, "warnings": self.warnings, } class DependencyResolver: """ Löst Plugin-Abhängigkeiten auf. Erstellt einen Abhängigkeitsgraphen, prüft Versionskompatibilität und berechnet die optimale Ladereihenfolge. Example: resolver = DependencyResolver() # Plugin-Informationen registrieren resolver.register_plugin( "my_plugin", version="1.0.0", dependencies={ "core_plugin": "^2.0.0", "util_plugin": "~1.5.0", } ) resolver.register_plugin("core_plugin", "2.1.0") resolver.register_plugin("util_plugin", "1.5.3") # Auflösen result = resolver.resolve() if result.success: for plugin in result.load_order: load_plugin(plugin) else: print(f"Fehler: {result.missing}, {result.incompatible}") # Visualisierung print(resolver.graph.to_dot()) """ def __init__(self, strict: bool = False) -> None: """ Initialisiert den Resolver. Args: strict: Strikte Versionsprüfung """ self._graph = DependencyGraph() self._checker = VersionChecker(strict=strict) self._available: dict[str, str] = {} # name -> version @property def graph(self) -> DependencyGraph: """Der Abhängigkeitsgraph.""" return self._graph def register_plugin( self, name: str, version: str = "1.0.0", dependencies: dict[str, str] | None = None, optional_dependencies: dict[str, str] | None = None, metadata: dict[str, Any] | None = None, ) -> DependencyNode: """ Registriert ein Plugin. Args: name: Plugin-Name version: Plugin-Version dependencies: Erforderliche Abhängigkeiten optional_dependencies: Optionale Abhängigkeiten metadata: Zusätzliche Metadaten Returns: Der erstellte Knoten """ node = DependencyNode( name=name, version=version, dependencies=dependencies or {}, optional_dependencies=optional_dependencies or {}, metadata=metadata or {}, ) if self._graph.has_node(name): self._graph.update_node(node) else: self._graph.add_node(node) self._available[name] = version return node def unregister_plugin(self, name: str) -> bool: """ Entfernt ein Plugin. Args: name: Plugin-Name Returns: True wenn entfernt """ if self._graph.remove_node(name): self._available.pop(name, None) return True return False def resolve( self, targets: list[str] | None = None, include_optional: bool = False, ) -> ResolutionResult: """ Löst Abhängigkeiten auf. Args: targets: Spezifische Plugins zum Auflösen (None = alle) include_optional: Optionale Abhängigkeiten einbeziehen Returns: ResolutionResult """ warnings: list[str] = [] missing: list[str] = [] incompatible: dict[str, CompatibilityResult] = {} # Fehlende Abhängigkeiten finden missing_deps = self._graph.get_missing_dependencies() for plugin, deps in missing_deps.items(): for dep in deps: node = self._graph.get_node(plugin) if node and dep in node.optional_dependencies: if include_optional: warnings.append( f"Optionale Abhängigkeit '{dep}' für '{plugin}' nicht gefunden" ) else: missing.append(f"{dep} (benötigt von {plugin})") # Versionskompatibilität prüfen for node in self._graph: for dep_name, requirement in node.dependencies.items(): if dep_name not in self._available: continue result = self._checker.check(requirement, self._available[dep_name]) if not result.compatible: incompatible[f"{node.name}->{dep_name}"] = result elif result.warnings: for warning in result.warnings: warnings.append(f"{node.name}->{dep_name}: {warning}") # Zyklus prüfen try: load_order = self._graph.get_load_order() # Nur angeforderte Plugins filtern if targets: needed = set(targets) for target in targets: needed.update(self._graph.get_all_dependencies(target)) load_order = [p for p in load_order if p in needed] except CycleError as e: return ResolutionResult( success=False, missing=missing, incompatible=incompatible, warnings=[f"Zyklische Abhängigkeit: {' -> '.join(e.cycle)}"], ) # Erfolg bestimmen success = not missing and not incompatible return ResolutionResult( success=success, load_order=load_order, resolved={name: self._available[name] for name in load_order if name in self._available}, missing=missing, incompatible=incompatible, warnings=warnings, ) def check_can_install( self, name: str, version: str, dependencies: dict[str, str] | None = None, ) -> ResolutionResult: """ Prüft, ob ein Plugin installiert werden kann. Args: name: Plugin-Name version: Plugin-Version dependencies: Plugin-Abhängigkeiten Returns: ResolutionResult """ dependencies = dependencies or {} missing = [] incompatible: dict[str, CompatibilityResult] = {} for dep_name, requirement in dependencies.items(): if dep_name not in self._available: missing.append(dep_name) else: result = self._checker.check(requirement, self._available[dep_name]) if not result.compatible: incompatible[dep_name] = result success = not missing and not incompatible return ResolutionResult( success=success, missing=missing, incompatible=incompatible, ) def check_can_uninstall(self, name: str) -> tuple[bool, list[str]]: """ Prüft, ob ein Plugin deinstalliert werden kann. Args: name: Plugin-Name Returns: (Kann deinstalliert werden, Liste abhängiger Plugins) """ dependents = list(self._graph.get_dependents(name)) return (not dependents, dependents) def get_install_order(self, plugins: list[str]) -> list[str]: """ Berechnet die Installationsreihenfolge für mehrere Plugins. Args: plugins: Liste der zu installierenden Plugins Returns: Sortierte Installationsreihenfolge """ needed: set[str] = set(plugins) for plugin in plugins: needed.update(self._graph.get_all_dependencies(plugin)) order = self._graph.get_load_order() return [p for p in order if p in needed] def get_uninstall_order(self, plugins: list[str]) -> list[str]: """ Berechnet die Deinstallationsreihenfolge. Args: plugins: Liste der zu entfernenden Plugins Returns: Sortierte Deinstallationsreihenfolge """ to_remove: set[str] = set(plugins) # Alle abhängigen auch entfernen for plugin in plugins: to_remove.update(self._graph.get_all_dependents(plugin)) order = self._graph.get_unload_order() return [p for p in order if p in to_remove] def get_affected_by_update( self, name: str, new_version: str, ) -> list[str]: """ Findet alle Plugins, die von einem Update betroffen sind. Args: name: Plugin-Name new_version: Neue Version Returns: Liste betroffener Plugins """ affected = [] dependents = self._graph.get_all_dependents(name) for dependent in dependents: node = self._graph.get_node(dependent) if node and name in node.dependencies: requirement = node.dependencies[name] result = self._checker.check(requirement, new_version) if not result.compatible: affected.append(dependent) return affected def clear(self) -> None: """Löscht alle registrierten Plugins.""" self._graph = DependencyGraph() self._available.clear() def get_statistics(self) -> dict[str, Any]: """Gibt Statistiken zurück.""" return { "total_plugins": self._graph.node_count, "total_dependencies": self._graph.edge_count, "root_plugins": len(self._graph.get_root_nodes()), "leaf_plugins": len(self._graph.get_leaf_nodes()), "has_cycles": self._graph.has_cycle(), "missing_dependencies": sum( len(deps) for deps in self._graph.get_missing_dependencies().values() ), }