"""Migrations-Engine für das Monorepo. Migriert bestehende Repositories in die Monorepo-Struktur unter Bewahrung der vollständigen Git-Historie (Commits, Branches, Tags, Autoreninformationen). Verantwortung: - Inkrementelle Migration einzelner Repos - Konflikterkennung (Pfadkollision, Branch-Namenskonflikt, Namenskonvention) - Backup-Erstellung für Rollback-Möglichkeit - Tracking des Migrationsstatus in einer Registry-Datei """ from __future__ import annotations import logging import re import subprocess from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from typing import Any, Literal import yaml from monorepo.models import MigrationPlan logger = logging.getLogger(__name__) # Namenskonvention: kebab-case, 2-50 Zeichen _NAME_PATTERN = re.compile(r"^[a-z0-9][a-z0-9\-]{0,48}[a-z0-9]$") # --------------------------------------------------------------------------- # Datenmodelle # --------------------------------------------------------------------------- @dataclass class MigrationConflict: """Ein erkannter Konflikt während der Migration.""" conflict_type: Literal["path_collision", "branch_name_conflict", "naming_convention"] description: str source: str = "" target: str = "" @dataclass class MigrationResult: """Ergebnis einer Migrations-Operation.""" success: bool repo_name: str target_path: str conflicts: list[MigrationConflict] = field(default_factory=list) error_message: str = "" commits_migrated: int = 0 branches_migrated: list[str] = field(default_factory=list) tags_migrated: list[str] = field(default_factory=list) @dataclass class ValidationResult: """Ergebnis einer Migrations-Validierung. Prüft Vollständigkeit einer Migration anhand von: - Commit-Anzahl - Vorhandensein aller Branches - Vorhandensein aller Tags - Dateibaum-Vollständigkeit - Test-Durchlauf (optional) """ success: bool repo_name: str checks: dict[str, bool] = field(default_factory=dict) details: list[str] = field(default_factory=list) error_message: str = "" @dataclass class MigrationRegistryEntry: """Eintrag in der Migrations-Registry.""" repo_name: str source_repo: str target_context: str target_name: str target_path: str mode: str status: Literal["completed", "in_progress", "failed", "rolled_back"] migrated_at: str backup_ref: str = "" commits_count: int = 0 branches: list[str] = field(default_factory=list) tags: list[str] = field(default_factory=list) # --------------------------------------------------------------------------- # MigrationEngine # --------------------------------------------------------------------------- class MigrationEngine: """Migriert bestehende Repos in die Monorepo-Struktur. Unterstützt inkrementelle Migration: Repositories werden einzeln und unabhängig voneinander migriert. Bereits migrierte und noch nicht migrierte Repos bleiben parallel funktionsfähig. Konflikterkennung: - Pfadkollision: Zielpfad existiert bereits - Branch-Namenskonflikt: Branch-Namen aus dem Quell-Repo kollidieren - Namenskonvention: target_name entspricht nicht kebab-case """ def __init__(self, monorepo_root: Path, registry_path: Path | None = None) -> None: """Initialisiert die MigrationEngine. Args: monorepo_root: Wurzelverzeichnis des Monorepos. registry_path: Pfad zur migrations.yaml Registry-Datei. Standard: shared/config/migrations.yaml """ self.monorepo_root = monorepo_root self.registry_path = registry_path or ( monorepo_root / "shared" / "config" / "migrations.yaml" ) self._registry: list[MigrationRegistryEntry] = self._load_registry() # ----------------------------------------------------------------------- # Öffentliche API # ----------------------------------------------------------------------- def migrate(self, plan: MigrationPlan) -> MigrationResult: """Führt die Migration eines einzelnen Repos durch. Ablauf: 1. Konflikterkennung (Pfadkollision, Naming, Branch-Konflikte) 2. Bei Konflikten: Migration pausieren, Konflikte zurückmelden 3. Backup-Referenz erstellen (Tag für Rollback) 4. Migration via git subtree add durchführen 5. Migrationsstatus in Registry speichern Args: plan: MigrationPlan mit Quell-Repo, Zielkontext, Name, Modus. Returns: MigrationResult mit Erfolgs/Fehler-Status und Details. """ target_path = f"{plan.target_context}/{plan.target_name}" # Schritt 1: Konflikterkennung conflicts = self._detect_conflicts(plan) if conflicts: logger.warning( "Migration von '%s' pausiert: %d Konflikte erkannt.", plan.source_repo, len(conflicts), ) # Status in Registry als fehlgeschlagen vermerken self._update_registry( plan=plan, target_path=target_path, status="failed", error="Konflikte erkannt – Migration pausiert.", ) return MigrationResult( success=False, repo_name=plan.target_name, target_path=target_path, conflicts=conflicts, error_message=( f"Migration pausiert: {len(conflicts)} Konflikt(e) erkannt. " f"Bitte Konflikte manuell auflösen." ), ) # Schritt 2: Backup-Referenz erstellen backup_ref = self._create_backup_ref(plan) # Schritt 3: Migration durchführen try: result = self._execute_migration(plan, target_path) except MigrationError as e: logger.error( "Migration von '%s' fehlgeschlagen: %s", plan.source_repo, str(e), ) self._update_registry( plan=plan, target_path=target_path, status="failed", backup_ref=backup_ref, error=str(e), ) return MigrationResult( success=False, repo_name=plan.target_name, target_path=target_path, error_message=str(e), ) # Schritt 4: Registry aktualisieren self._update_registry( plan=plan, target_path=target_path, status="completed", backup_ref=backup_ref, commits_count=result.commits_migrated, branches=result.branches_migrated, tags=result.tags_migrated, ) logger.info( "Migration von '%s' nach '%s' erfolgreich abgeschlossen " "(%d Commits, %d Branches, %d Tags).", plan.source_repo, target_path, result.commits_migrated, len(result.branches_migrated), len(result.tags_migrated), ) return result def get_registry(self) -> list[MigrationRegistryEntry]: """Gibt die aktuelle Migrations-Registry zurück. Returns: Liste aller MigrationRegistryEntry-Objekte. """ return list(self._registry) def is_migrated(self, repo_name: str) -> bool: """Prüft ob ein Repo bereits erfolgreich migriert wurde. Args: repo_name: Name des zu prüfenden Repos. Returns: True wenn das Repo den Status 'completed' hat. """ for entry in self._registry: if entry.repo_name == repo_name and entry.status == "completed": return True return False def validate(self, repo_name: str) -> ValidationResult: """Validiert eine abgeschlossene Migration. Prüft: - Commit-Anzahl im migrierten Subtree vs. Registry-Erwartung - Vorhandensein aller registrierten Branches - Vorhandensein aller registrierten Tags - Dateibaum-Vollständigkeit (Zielverzeichnis existiert und enthält Dateien) - Test-Durchlauf (pytest im Zielverzeichnis, optional) Args: repo_name: Name des zu validierenden Repos (wie in Registry). Returns: ValidationResult mit Einzelprüfungen und Gesamtergebnis. """ # Registry-Eintrag suchen entry = self._find_registry_entry(repo_name) if entry is None: return ValidationResult( success=False, repo_name=repo_name, checks={}, details=[], error_message=f"Repo '{repo_name}' nicht in der Migrations-Registry gefunden.", ) checks: dict[str, bool] = {} details: list[str] = [] # 1. Commit-Anzahl prüfen actual_commits = self._count_commits_in_subtree(entry.target_path) expected_commits = entry.commits_count if expected_commits > 0: commits_ok = actual_commits >= expected_commits else: # Kein Erwartungswert: mindestens 1 Commit reicht commits_ok = actual_commits >= 1 checks["commits"] = commits_ok details.append( f"Commits: {actual_commits} gefunden" f" (erwartet: {expected_commits or '≥1'}) → " f"{'OK' if commits_ok else 'FEHLER'}" ) # 2. Branches prüfen branches_ok = self._validate_branches(entry) checks["branches"] = branches_ok if entry.branches: details.append( f"Branches: {len(entry.branches)} registriert → " f"{'OK' if branches_ok else 'FEHLER'}" ) else: details.append("Branches: Keine registriert → OK (übersprungen)") # 3. Tags prüfen tags_ok = self._validate_tags(entry) checks["tags"] = tags_ok if entry.tags: details.append( f"Tags: {len(entry.tags)} registriert → " f"{'OK' if tags_ok else 'FEHLER'}" ) else: details.append("Tags: Keine registriert → OK (übersprungen)") # 4. Dateibaum-Vollständigkeit prüfen files_ok = self._validate_file_tree(entry.target_path) checks["files"] = files_ok details.append( f"Dateibaum: Verzeichnis '{entry.target_path}' → " f"{'OK' if files_ok else 'FEHLER (leer oder nicht vorhanden)'}" ) # 5. Test-Durchlauf (optional, schlägt nicht fehl wenn keine Tests vorhanden) tests_ok = self._run_validation_tests(entry.target_path) checks["tests"] = tests_ok details.append( f"Tests: → {'OK' if tests_ok else 'FEHLER (Tests fehlgeschlagen)'}" ) # Gesamtergebnis success = all(checks.values()) return ValidationResult( success=success, repo_name=repo_name, checks=checks, details=details, ) def rollback(self, repo_name: str) -> None: """Macht die Migration eines Repos rückgängig. Entfernt das migrierte Verzeichnis aus dem Monorepo via `git rm -rf` und erstellt einen Rollback-Commit. Der Rollback beeinflusst keine anderen migrierten Repos. Der Registry-Status wird auf 'rolled_back' gesetzt. Args: repo_name: Name des zurückzurollenden Repos. Raises: MigrationError: Wenn das Repo nicht in der Registry ist oder der Rollback fehlschlägt. """ entry = self._find_registry_entry(repo_name) if entry is None: raise MigrationError( f"Repo '{repo_name}' nicht in der Migrations-Registry gefunden. " f"Rollback nicht möglich." ) if entry.status == "rolled_back": raise MigrationError( f"Repo '{repo_name}' wurde bereits zurückgerollt." ) target_dir = self.monorepo_root / entry.target_path # Verzeichnis entfernen (git rm -rf) ohne andere Repos zu beeinflussen if target_dir.exists(): try: self._run_git(["rm", "-rf", entry.target_path]) except subprocess.CalledProcessError as e: combined = (e.stdout or "") + (e.stderr or "") raise MigrationError( f"Rollback fehlgeschlagen: git rm -rf {entry.target_path}: " f"{combined.strip()[:200]}" ) from e # Rollback-Commit erstellen try: self._run_git([ "commit", "-m", f"Rollback Migration von '{repo_name}' " f"(Pfad: {entry.target_path})", ]) except subprocess.CalledProcessError as e: combined = (e.stdout or "") + (e.stderr or "") raise MigrationError( f"Rollback-Commit fehlgeschlagen: {combined.strip()[:200]}" ) from e # Registry-Status aktualisieren entry.status = "rolled_back" self._save_registry() logger.info( "Rollback von '%s' (Pfad: %s) erfolgreich. " "Backup-Ref: %s", repo_name, entry.target_path, entry.backup_ref or "(keine)", ) # ----------------------------------------------------------------------- # Validierungs-Hilfsmethoden # ----------------------------------------------------------------------- def _find_registry_entry(self, repo_name: str) -> MigrationRegistryEntry | None: """Sucht einen Registry-Eintrag nach repo_name. Args: repo_name: Name des Repos. Returns: Den Eintrag oder None wenn nicht gefunden. """ for entry in self._registry: if entry.repo_name == repo_name: return entry return None def _validate_branches(self, entry: MigrationRegistryEntry) -> bool: """Prüft ob alle registrierten Branches noch vorhanden sind. Args: entry: Der Registry-Eintrag. Returns: True wenn alle Branches vorhanden sind (oder keine registriert). """ if not entry.branches: return True local_branches = self._get_local_branches() # Prüfe ob die migrierten Branches als lokale oder Remote-Branches existieren # Branches können nach der Migration als lokale Branches vorhanden sein for branch in entry.branches: if branch not in local_branches: # Prüfe auch Remote-Tracking-Branches try: result = self._run_git( ["branch", "-a", "--list", f"*{branch}*"] ) if branch not in result.stdout: logger.warning( "Branch '%s' für Repo '%s' nicht gefunden.", branch, entry.repo_name, ) return False except subprocess.CalledProcessError: return False return True def _validate_tags(self, entry: MigrationRegistryEntry) -> bool: """Prüft ob alle registrierten Tags noch vorhanden sind. Args: entry: Der Registry-Eintrag. Returns: True wenn alle Tags vorhanden sind (oder keine registriert). """ if not entry.tags: return True try: result = self._run_git(["tag", "--list"]) existing_tags = {t.strip() for t in result.stdout.splitlines() if t.strip()} except subprocess.CalledProcessError: return False for tag in entry.tags: if tag not in existing_tags: logger.warning( "Tag '%s' für Repo '%s' nicht gefunden.", tag, entry.repo_name, ) return False return True def _validate_file_tree(self, target_path: str) -> bool: """Prüft ob das Zielverzeichnis existiert und Dateien enthält. Args: target_path: Relativer Pfad zum migrierten Verzeichnis. Returns: True wenn das Verzeichnis existiert und nicht leer ist. """ target_dir = self.monorepo_root / target_path if not target_dir.exists(): return False if not target_dir.is_dir(): return False # Prüfe ob mindestens eine Datei vorhanden ist return any(target_dir.rglob("*")) def _run_validation_tests(self, target_path: str) -> bool: """Führt vorhandene Tests im Zielverzeichnis aus. Versucht pytest im Zielverzeichnis auszuführen. Wenn keine Tests vorhanden sind oder pytest nicht verfügbar ist, gilt die Prüfung als bestanden. Args: target_path: Relativer Pfad zum migrierten Verzeichnis. Returns: True wenn Tests bestehen oder keine Tests vorhanden sind. """ target_dir = self.monorepo_root / target_path if not target_dir.exists(): return True # Keine Tests möglich, zählt nicht als Fehler # Prüfe ob Test-Dateien vorhanden sind test_files = list(target_dir.rglob("test_*.py")) + list( target_dir.rglob("*_test.py") ) if not test_files: return True # Keine Tests vorhanden, gilt als bestanden try: subprocess.run( ["python", "-m", "pytest", str(target_dir), "--tb=short", "-q"], cwd=self.monorepo_root, capture_output=True, text=True, check=True, timeout=120, ) return True except subprocess.CalledProcessError: return False except (FileNotFoundError, subprocess.TimeoutExpired): # pytest nicht verfügbar oder Timeout → kein Fehlschlag return True # ----------------------------------------------------------------------- # Konflikterkennung # ----------------------------------------------------------------------- def _detect_conflicts(self, plan: MigrationPlan) -> list[MigrationConflict]: """Erkennt Konflikte vor der Migration. Prüft: 1. Namenskonvention (kebab-case) 2. Pfadkollision (Zielverzeichnis existiert bereits) 3. Branch-Namenskonflikt (Branch-Namen aus Quelle kollidieren) Args: plan: Der Migrationsplan. Returns: Liste erkannter Konflikte (leer = keine Konflikte). """ conflicts: list[MigrationConflict] = [] # 1. Namenskonvention prüfen if not self._validate_name(plan.target_name): conflicts.append( MigrationConflict( conflict_type="naming_convention", description=( f"Der Zielname '{plan.target_name}' entspricht nicht der " f"kebab-case-Namenskonvention (Kleinbuchstaben, Ziffern, " f"Bindestriche, 2-50 Zeichen, kein Bindestrich am Anfang/Ende)." ), source=plan.source_repo, target=plan.target_name, ) ) # 2. Pfadkollision prüfen target_dir = self.monorepo_root / plan.target_context / plan.target_name if target_dir.exists(): conflicts.append( MigrationConflict( conflict_type="path_collision", description=( f"Der Zielpfad '{plan.target_context}/{plan.target_name}' " f"existiert bereits im Monorepo." ), source=plan.source_repo, target=str(target_dir), ) ) # 3. Branch-Namenskonflikt prüfen branch_conflicts = self._detect_branch_conflicts(plan) conflicts.extend(branch_conflicts) return conflicts def _validate_name(self, name: str) -> bool: """Prüft ob ein Name der kebab-case-Konvention entspricht. Args: name: Der zu prüfende Name. Returns: True wenn der Name gültig ist. """ if not name: return False return _NAME_PATTERN.match(name) is not None def _detect_branch_conflicts(self, plan: MigrationPlan) -> list[MigrationConflict]: """Erkennt Branch-Namenskonflikte zwischen Quell-Repo und Monorepo. Prüft ob Branch-Namen aus dem Quell-Repo mit bestehenden Branches im Monorepo kollidieren würden. Args: plan: Der Migrationsplan. Returns: Liste der Branch-Namenskonflikte. """ conflicts: list[MigrationConflict] = [] # Branches im Quell-Repo ermitteln source_branches = self._get_remote_branches(plan.source_repo) if not source_branches: return conflicts # Bestehende Branches im Monorepo ermitteln local_branches = self._get_local_branches() # Prefixed Branch-Namen für die Migration (z.B. "repo-name/branch") # Ohne Prefix könnten Konflikte entstehen bei direkter Migration for branch in source_branches: # Typische Branches wie main/master/develop sind erwartbare Konflikte if branch in local_branches and branch not in ("main", "master"): conflicts.append( MigrationConflict( conflict_type="branch_name_conflict", description=( f"Branch '{branch}' aus '{plan.source_repo}' " f"existiert bereits im Monorepo." ), source=plan.source_repo, target=branch, ) ) return conflicts def _get_remote_branches(self, repo_url: str) -> list[str]: """Ermittelt die Branch-Namen eines Remote-Repos. Args: repo_url: URL oder Pfad zum Remote-Repository. Returns: Liste der Branch-Namen (ohne refs/heads/ Prefix). """ try: result = self._run_git(["ls-remote", "--heads", repo_url]) branches: list[str] = [] for line in result.stdout.splitlines(): # Format: "\trefs/heads/" parts = line.strip().split("\t") if len(parts) == 2 and parts[1].startswith("refs/heads/"): branch_name = parts[1].removeprefix("refs/heads/") branches.append(branch_name) return branches except subprocess.CalledProcessError: logger.warning( "Konnte Branches von '%s' nicht ermitteln. " "Branch-Konfliktprüfung wird übersprungen.", repo_url, ) return [] def _get_local_branches(self) -> list[str]: """Ermittelt die lokalen Branch-Namen im Monorepo. Returns: Liste der lokalen Branch-Namen. """ try: result = self._run_git(["branch", "--format=%(refname:short)"]) return [b.strip() for b in result.stdout.splitlines() if b.strip()] except subprocess.CalledProcessError: logger.warning("Konnte lokale Branches nicht ermitteln.") return [] # ----------------------------------------------------------------------- # Migration Execution # ----------------------------------------------------------------------- def _execute_migration(self, plan: MigrationPlan, target_path: str) -> MigrationResult: """Führt die eigentliche Migration via git subtree add durch. Ablauf: 1. Remote hinzufügen (temporär) 2. git subtree add für den main/master Branch 3. Branches und Tags vom Quell-Repo erfassen 4. Remote entfernen Args: plan: Der Migrationsplan. target_path: Relativer Zielpfad im Monorepo. Returns: MigrationResult bei Erfolg. Raises: MigrationError: Bei Fehler während der Migration. """ remote_name = f"migration-{plan.target_name}" try: # Remote hinzufügen self._add_remote(remote_name, plan.source_repo) # Fetch vom Remote self._fetch_remote(remote_name) # Hauptbranch ermitteln main_branch = self._detect_main_branch(remote_name) # git subtree add – bewahrt Historie self._run_git([ "subtree", "add", f"--prefix={target_path}", remote_name, main_branch, ]) # Migrationsergebnis zusammenstellen branches = self._get_fetched_branches(remote_name) tags = self._get_fetched_tags(remote_name) commits_count = self._count_commits_in_subtree(target_path) return MigrationResult( success=True, repo_name=plan.target_name, target_path=target_path, commits_migrated=commits_count, branches_migrated=branches, tags_migrated=tags, ) except subprocess.CalledProcessError as e: combined = (e.stdout or "") + (e.stderr or "") raise MigrationError( f"Git-Operation fehlgeschlagen: {combined.strip()[:300]}" ) from e finally: # Remote aufräumen (auch bei Fehler) self._remove_remote(remote_name) def _add_remote(self, name: str, url: str) -> None: """Fügt ein temporäres Remote hinzu. Args: name: Name des Remotes. url: URL oder Pfad zum Repository. """ try: self._run_git(["remote", "add", name, url]) except subprocess.CalledProcessError: # Remote existiert möglicherweise bereits – aktualisieren self._run_git(["remote", "set-url", name, url]) def _fetch_remote(self, remote_name: str) -> None: """Führt einen Fetch vom Remote durch inkl. Tags. Args: remote_name: Name des Remotes. """ self._run_git(["fetch", remote_name, "--tags"]) def _detect_main_branch(self, remote_name: str) -> str: """Erkennt den Hauptbranch eines Remotes (main oder master). Args: remote_name: Name des Remotes. Returns: Name des Hauptbranches. Raises: MigrationError: Wenn kein Hauptbranch gefunden wird. """ try: result = self._run_git([ "branch", "-r", "--list", f"{remote_name}/*" ]) remote_branches = [ b.strip().removeprefix(f"{remote_name}/") for b in result.stdout.splitlines() if b.strip() ] except subprocess.CalledProcessError: remote_branches = [] # Priorität: main > master > erster gefundener Branch for candidate in ("main", "master"): if candidate in remote_branches: return candidate if remote_branches: return remote_branches[0] raise MigrationError( f"Kein Branch im Remote '{remote_name}' gefunden. " f"Ist das Repository leer?" ) def _get_fetched_branches(self, remote_name: str) -> list[str]: """Ermittelt die Branches, die vom Remote gefetcht wurden. Args: remote_name: Name des Remotes. Returns: Liste der Branch-Namen. """ try: result = self._run_git([ "branch", "-r", "--list", f"{remote_name}/*" ]) branches = [] for line in result.stdout.splitlines(): branch = line.strip().removeprefix(f"{remote_name}/") if branch and "HEAD" not in branch: branches.append(branch) return branches except subprocess.CalledProcessError: return [] def _get_fetched_tags(self, remote_name: str) -> list[str]: """Ermittelt die Tags aus dem Quell-Repo. Listet Tags, die durch den Fetch des Remotes verfügbar sind. Args: remote_name: Name des Remotes (für Kontext). Returns: Liste der Tag-Namen. """ try: result = self._run_git(["tag", "--list"]) return [t.strip() for t in result.stdout.splitlines() if t.strip()] except subprocess.CalledProcessError: return [] def _count_commits_in_subtree(self, prefix: str) -> int: """Zählt die Commits im migrierten Subtree. Args: prefix: Subtree-Prefix (Zielpfad). Returns: Anzahl der Commits. """ try: result = self._run_git([ "log", "--oneline", "--", prefix ]) return len([l for l in result.stdout.splitlines() if l.strip()]) except subprocess.CalledProcessError: return 0 def _remove_remote(self, name: str) -> None: """Entfernt ein temporäres Remote (sicher, ignoriert Fehler). Args: name: Name des zu entfernenden Remotes. """ try: self._run_git(["remote", "remove", name]) except subprocess.CalledProcessError: logger.debug("Remote '%s' konnte nicht entfernt werden (evtl. nicht vorhanden).", name) # ----------------------------------------------------------------------- # Backup & Rollback Support # ----------------------------------------------------------------------- def _create_backup_ref(self, plan: MigrationPlan) -> str: """Erstellt eine Backup-Referenz (Tag) für Rollback-Möglichkeit. Erstellt einen Tag am aktuellen HEAD, der den Zustand vor der Migration markiert. Args: plan: Der Migrationsplan. Returns: Name des erstellten Backup-Tags. """ timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") backup_tag = f"pre-migration/{plan.target_name}/{timestamp}" try: self._run_git([ "tag", backup_tag, "-m", f"Backup vor Migration von '{plan.source_repo}' " f"nach '{plan.target_context}/{plan.target_name}'", ]) logger.info("Backup-Referenz erstellt: %s", backup_tag) except subprocess.CalledProcessError: logger.warning( "Backup-Tag '%s' konnte nicht erstellt werden.", backup_tag ) backup_tag = "" return backup_tag # ----------------------------------------------------------------------- # Registry-Verwaltung # ----------------------------------------------------------------------- def _load_registry(self) -> list[MigrationRegistryEntry]: """Lädt die Migrations-Registry aus der YAML-Datei. Returns: Liste aller Registry-Einträge. """ if not self.registry_path.exists(): return [] try: with open(self.registry_path, encoding="utf-8") as f: data: dict[str, Any] = yaml.safe_load(f) or {} except (yaml.YAMLError, OSError) as e: logger.warning("Registry konnte nicht geladen werden: %s", e) return [] entries: list[MigrationRegistryEntry] = [] for item in data.get("migrations", []): entries.append( MigrationRegistryEntry( repo_name=item.get("repo_name", ""), source_repo=item.get("source_repo", ""), target_context=item.get("target_context", ""), target_name=item.get("target_name", ""), target_path=item.get("target_path", ""), mode=item.get("mode", "direct"), status=item.get("status", "failed"), migrated_at=item.get("migrated_at", ""), backup_ref=item.get("backup_ref", ""), commits_count=item.get("commits_count", 0), branches=item.get("branches", []), tags=item.get("tags", []), ) ) return entries def _save_registry(self) -> None: """Speichert die Migrations-Registry in die YAML-Datei.""" data = { "version": "1.0", "migrations": [ { "repo_name": entry.repo_name, "source_repo": entry.source_repo, "target_context": entry.target_context, "target_name": entry.target_name, "target_path": entry.target_path, "mode": entry.mode, "status": entry.status, "migrated_at": entry.migrated_at, "backup_ref": entry.backup_ref, "commits_count": entry.commits_count, "branches": entry.branches, "tags": entry.tags, } for entry in self._registry ], } self.registry_path.parent.mkdir(parents=True, exist_ok=True) with open(self.registry_path, "w", encoding="utf-8") as f: yaml.dump(data, f, default_flow_style=False, allow_unicode=True) logger.debug("Migrations-Registry gespeichert: %s", self.registry_path) def _update_registry( self, plan: MigrationPlan, target_path: str, status: Literal["completed", "in_progress", "failed", "rolled_back"], backup_ref: str = "", commits_count: int = 0, branches: list[str] | None = None, tags: list[str] | None = None, error: str = "", ) -> None: """Aktualisiert oder erstellt einen Registry-Eintrag. Args: plan: Der Migrationsplan. target_path: Zielpfad im Monorepo. status: Aktueller Migrationsstatus. backup_ref: Backup-Tag-Name (falls erstellt). commits_count: Anzahl migrierter Commits. branches: Liste migrierter Branches. tags: Liste migrierter Tags. error: Fehlermeldung (falls fehlgeschlagen). """ # Bestehenden Eintrag suchen und aktualisieren for entry in self._registry: if entry.repo_name == plan.target_name: entry.status = status entry.backup_ref = backup_ref or entry.backup_ref entry.commits_count = commits_count or entry.commits_count entry.branches = branches if branches is not None else entry.branches entry.tags = tags if tags is not None else entry.tags entry.migrated_at = datetime.now().isoformat(timespec="seconds") self._save_registry() return # Neuen Eintrag erstellen new_entry = MigrationRegistryEntry( repo_name=plan.target_name, source_repo=plan.source_repo, target_context=plan.target_context, target_name=plan.target_name, target_path=target_path, mode=plan.mode, status=status, migrated_at=datetime.now().isoformat(timespec="seconds"), backup_ref=backup_ref, commits_count=commits_count, branches=branches or [], tags=tags or [], ) self._registry.append(new_entry) self._save_registry() # ----------------------------------------------------------------------- # Git-Hilfsmethoden # ----------------------------------------------------------------------- def _run_git( self, args: list[str], cwd: Path | None = None ) -> subprocess.CompletedProcess[str]: """Führt einen Git-Befehl aus. Args: args: Git-Argumente (ohne 'git' Präfix). cwd: Arbeitsverzeichnis. Standard: monorepo_root. Returns: CompletedProcess mit stdout/stderr. Raises: subprocess.CalledProcessError: Bei Fehler im Git-Befehl. """ cmd = ["git"] + args logger.debug("Ausführen: %s (cwd=%s)", " ".join(cmd), cwd or self.monorepo_root) return subprocess.run( cmd, cwd=cwd or self.monorepo_root, capture_output=True, text=True, check=True, ) # --------------------------------------------------------------------------- # Exceptions # --------------------------------------------------------------------------- class MigrationError(Exception): """Fehler während einer Migration."""