# #region GitServiceBranchMixin [C:4] [TYPE Module] [SEMANTICS git, branch, checkout, list, namespace, lock] # @LAYER Infrastructure # @BRIEF Branch and commit operations for GitService — gitflow branches, list/create/checkout branches, commit changes (all concurrent-safe via per-dashboard locks). # @RELATION CALLED_BY -> [GitService] from datetime import UTC, datetime import os from fastapi import HTTPException from git import Repo from git.exc import GitCommandError from src.core.logger import belief_scope, logger # #region GitServiceBranchMixin [C:3] [TYPE Class] # @BRIEF Mixin providing branch and commit operations for GitService. class GitServiceBranchMixin: # region _ensure_gitflow_branches [C:4] [TYPE Function] [SEMANTICS git,gitflow,branch,lock] # @PURPOSE: Ensure standard GitFlow branches (main/dev/preprod) exist locally and on origin. # @PRE repo is a valid GitPython Repo instance. # @POST main, dev, preprod are available in local repository and pushed to origin when available. # Active branch unchanged (no spurious checkout). def _ensure_gitflow_branches(self, repo: Repo, dashboard_id: int) -> None: with belief_scope("GitService._ensure_gitflow_branches"): required_branches = ["main", "dev", "preprod"] local_heads = {head.name: head for head in getattr(repo, "heads", [])} base_commit = None try: base_commit = repo.head.commit except Exception: base_commit = None if "main" in local_heads: base_commit = local_heads["main"].commit if base_commit is None: logger.reason( f"Skipping branch bootstrap for dashboard {dashboard_id}: repository has no commits", extra={"src": "_ensure_gitflow_branches"}, ) return if "main" not in local_heads: local_heads["main"] = repo.create_head("main", base_commit) logger.reason(f"Created local branch main for dashboard {dashboard_id}", extra={"src": "_ensure_gitflow_branches"}) for branch_name in ("dev", "preprod"): if branch_name in local_heads: continue local_heads[branch_name] = repo.create_head(branch_name, local_heads["main"].commit) logger.reason( f"Created local branch {branch_name} for dashboard {dashboard_id}", extra={"src": "_ensure_gitflow_branches"}, ) try: origin = repo.remote(name="origin") except ValueError: logger.reason( f"Remote origin is not configured for dashboard {dashboard_id}; skipping remote branch creation", extra={"src": "_ensure_gitflow_branches"}, ) return remote_branch_names = set() try: origin.fetch() for ref in origin.refs: remote_head = getattr(ref, "remote_head", None) if remote_head: remote_branch_names.add(str(remote_head)) except Exception as e: logger.reason(f"Failed to fetch origin refs: {e}", extra={"src": "_ensure_gitflow_branches"}) for branch_name in required_branches: if branch_name in remote_branch_names: continue try: origin.push(refspec=f"{branch_name}:{branch_name}") logger.reason( f"Pushed branch {branch_name} to origin for dashboard {dashboard_id}", extra={"src": "_ensure_gitflow_branches"}, ) except Exception as e: logger.error(f"[_ensure_gitflow_branches][Coherence:Failed] Failed to push branch {branch_name} to origin: {e}") raise HTTPException( status_code=500, detail=f"Failed to create default branch '{branch_name}' on remote: {e!s}", ) # Fix 6: Only checkout if not already on dev try: current_branch = repo.active_branch.name except Exception: current_branch = None if current_branch != "dev": try: repo.git.checkout("dev") logger.reason(f"Checked out default branch dev for dashboard {dashboard_id}", extra={"src": "_ensure_gitflow_branches"}) except Exception as e: logger.reason(f"Could not checkout dev branch for dashboard {dashboard_id}: {e}", extra={"src": "_ensure_gitflow_branches"}) # endregion _ensure_gitflow_branches # region list_branches [C:4] [TYPE Function] [SEMANTICS git,branch,list,lock] # @PURPOSE: List all branches (excluding tags) for a dashboard's repository, concurrent-safe. # @PRE Repository for dashboard_id exists. # @POST Returns a list of branch metadata dictionaries (no tag refs). # @RETURN List[dict] def list_branches(self, dashboard_id: int) -> list[dict]: with self._locked(dashboard_id), belief_scope("GitService.list_branches"): repo = self.get_repo(dashboard_id) logger.reason(f"Listing branches for {dashboard_id}. Refs: {repo.refs}", extra={"src": "list_branches"}) branches = [] for ref in repo.refs: try: # Fix 8: Skip tag refs ref_name = str(ref.name) if ref_name.startswith('refs/tags/'): continue name = ref_name.replace('refs/heads/', '').replace('refs/remotes/origin/', '') if any(b['name'] == name for b in branches): continue branches.append({ "name": name, "commit_hash": ref.commit.hexsha if hasattr(ref, 'commit') else "0000000", "is_remote": ref.is_remote() if hasattr(ref, 'is_remote') else False, "last_updated": datetime.fromtimestamp(ref.commit.committed_date) if hasattr(ref, 'commit') else datetime.now(UTC) }) except Exception as e: logger.reason(f"Skipping ref {ref}: {e}", extra={"src": "list_branches"}) try: active_name = repo.active_branch.name if not any(b['name'] == active_name for b in branches): branches.append({ "name": active_name, "commit_hash": "0000000", "is_remote": False, "last_updated": datetime.now(UTC) }) except Exception as e: logger.reason(f"Could not determine active branch: {e}", extra={"src": "list_branches"}) if not branches: branches.append({ "name": "dev", "commit_hash": "0000000", "is_remote": False, "last_updated": datetime.now(UTC) }) return branches # endregion list_branches # region create_branch [C:4] [TYPE Function] [SEMANTICS git,branch,create,lock] # @PURPOSE: Create a new branch from an existing one (concurrent-safe). # @PARAM name (str) - New branch name. # @PARAM from_branch (str) - Source branch. # @PRE Repository exists; name is valid; from_branch exists or repo is empty. # @POST A new branch is created in the repository. def create_branch(self, dashboard_id: int, name: str, from_branch: str = "main"): with self._locked(dashboard_id): with belief_scope("GitService.create_branch"): repo = self.get_repo(dashboard_id) logger.reason(f"Creating branch {name} from {from_branch}", extra={"src": "create_branch"}) if not repo.heads and not repo.remotes: logger.reason("Repository is empty. Creating initial commit to enable branching.", extra={"src": "create_branch"}) readme_path = os.path.join(repo.working_dir, "README.md") if not os.path.exists(readme_path): with open(readme_path, "w") as f: f.write(f"# Dashboard {dashboard_id}\nGit repository for Superset dashboard integration.") repo.index.add(["README.md"]) repo.index.commit("Initial commit") try: repo.commit(from_branch) except Exception: logger.reason(f"Source branch {from_branch} not found, using HEAD", extra={"src": "create_branch"}) from_branch = repo.head try: new_branch = repo.create_head(name, from_branch) return new_branch except Exception as e: logger.error(f"[create_branch][Coherence:Failed] {e}") raise # endregion create_branch # region checkout_branch [C:4] [TYPE Function] [SEMANTICS git,branch,checkout,lock] # @PURPOSE: Switch to a specific branch (concurrent-safe). # @PRE Repository exists and the specified branch name exists. # @POST The repository working directory is updated to the specified branch. # @SIDE_EFFECT May raise HTTPException(409) if local changes conflict with checkout. # @SIDE_EFFECT May raise HTTPException(500) if Git operation fails for other reasons. def checkout_branch(self, dashboard_id: int, name: str): with self._locked(dashboard_id): with belief_scope("GitService.checkout_branch"): repo = self.get_repo(dashboard_id) logger.reason(f"Checking out branch {name}", extra={"src": "checkout_branch"}) try: repo.git.checkout(name) except GitCommandError as e: stderr = str(e.stderr or "") details = str(e) lowered = stderr.lower() if "local changes" in lowered or "would be overwritten" in lowered: # Parse the list of files from stderr files = [] for line in stderr.split("\n"): stripped = line.strip() if stripped.startswith("\t") and ( stripped.endswith(".yaml") or "/" in stripped ): files.append(stripped.strip()) raise HTTPException( status_code=409, detail={ "error_code": "GIT_CHECKOUT_LOCAL_CHANGES", "message": ( f"Невозможно переключиться на ветку '{name}' — " f"локальные изменения будут перезаписаны. " f"Зафиксируйте или отложите изменения." ), "message_en": ( f"Cannot checkout branch '{name}' — " f"local changes would be overwritten. " f"Commit or stash your changes first." ), "files": files, "next_steps": [ "Зафиксируйте изменения (Commit) в текущей ветке перед переключением", "Отложите изменения через Stash (вручную: git stash)", "Отмените локальные изменения, если они не нужны", ], "next_steps_en": [ "Commit your changes in the current branch before switching", "Stash your changes manually: git stash", "Discard local changes if not needed", ], }, ) logger.error(f"[checkout_branch][Coherence:Failed] {e}") raise HTTPException( status_code=500, detail=f"Git checkout failed: {details}", ) # endregion checkout_branch # region commit_changes [C:4] [TYPE Function] [SEMANTICS git,commit,stage,lock] # @PURPOSE: Stage and commit changes (concurrent-safe). # @PARAM message (str) - Commit message. # @PARAM files (List[str]) - Optional list of specific files to stage. # @PRE Repository exists and has changes (dirty) or files are specified. # @POST Changes are staged and a new commit is created. def commit_changes(self, dashboard_id: int, message: str, files: list[str] = None): with self._locked(dashboard_id): with belief_scope("GitService.commit_changes"): repo = self.get_repo(dashboard_id) if not repo.is_dirty(untracked_files=True) and not files: logger.reason(f"No changes to commit for dashboard {dashboard_id}", extra={"src": "commit_changes"}) return if files: logger.reason(f"Staging files: {files}", extra={"src": "commit_changes"}) repo.index.add(files) else: logger.reason("Staging all changes", extra={"src": "commit_changes"}) repo.git.add(A=True) repo.index.commit(message) logger.info(f"[commit_changes][Coherence:OK] Committed changes with message: {message}") # endregion commit_changes # #endregion GitServiceBranchMixin # #endregion GitServiceBranchMixin