busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

chore: update semantic contracts and git merge handling

Browse Source
This commit is contained in:
busya
2026-03-16 20:34:28 +03:00
parent 51a0973dc1
commit 30323d8273
19 changed files with 480 additions and 257 deletions
Download Patch File Download Diff File Expand all files Collapse all files

253
backend/src/services/git_service.py
View File

@@ -44,31 +44,35 @@ class GitService:
# @PARAM: base_path (str) - Root directory for all Git clones.
# @PRE: base_path is a valid string path.
# @POST: GitService is initialized; base_path directory exists.
# @RELATION: CALLS -> [GitService._resolve_base_path]
# @RELATION: CALLS -> [GitService._ensure_base_path_exists]
def __init__(self, base_path: str = "git_repos"):
with belief_scope("GitService.__init__"):
backend_root = Path(__file__).parents[2]
self.legacy_base_path = str((backend_root / "git_repos").resolve())
self.base_path = self._resolve_base_path(base_path)
self._ensure_base_path_exists()
# [/DEF:__init__:Function]
# [/DEF:backend.src.services.git_service.GitService.__init__:Function]
# [DEF:_ensure_base_path_exists:Function]
# [DEF:backend.src.services.git_service.GitService._ensure_base_path_exists:Function]
# @PURPOSE: Ensure the repositories root directory exists and is a directory.
# @PRE: self.base_path is resolved to filesystem path.
# @POST: self.base_path exists as directory or raises ValueError.
# @RETURN: None
# @RELATION: USES -> [self.base_path]
def _ensure_base_path_exists(self) -> None:
base = Path(self.base_path)
if base.exists() and not base.is_dir():
raise ValueError(f"Git repositories base path is not a directory: {self.base_path}")
base.mkdir(parents=True, exist_ok=True)
# [/DEF:_ensure_base_path_exists:Function]
# [/DEF:backend.src.services.git_service.GitService._ensure_base_path_exists:Function]
# [DEF:backend.src.services.git_service.GitService._resolve_base_path:Function]
# @PURPOSE: Resolve base repository directory from explicit argument or global storage settings.
# @PRE: base_path is a string path.
# @POST: Returns absolute path for Git repositories root.
# @RETURN: str
# @RELATION: USES -> [AppConfigRecord]
def _resolve_base_path(self, base_path: str) -> str:
# Resolve relative to backend directory for backward compatibility.
backend_root = Path(__file__).parents[2]
@@ -104,24 +108,26 @@ class GitService:
except Exception as e:
logger.warning(f"[_resolve_base_path][Coherence:Failed] Falling back to default path: {e}")
return fallback_path
# [/DEF:_resolve_base_path:Function]
# [/DEF:backend.src.services.git_service.GitService._resolve_base_path:Function]
# [DEF:_normalize_repo_key:Function]
# [DEF:backend.src.services.git_service.GitService._normalize_repo_key:Function]
# @PURPOSE: Convert user/dashboard-provided key to safe filesystem directory name.
# @PRE: repo_key can be None/empty.
# @POST: Returns normalized non-empty key.
# @RETURN: str
# @RELATION: USES -> [re.sub]
def _normalize_repo_key(self, repo_key: Optional[str]) -> str:
raw_key = str(repo_key or "").strip().lower()
normalized = re.sub(r"[^a-z0-9._-]+", "-", raw_key).strip("._-")
return normalized or "dashboard"
# [/DEF:_normalize_repo_key:Function]
# [/DEF:backend.src.services.git_service.GitService._normalize_repo_key:Function]
# [DEF:_update_repo_local_path:Function]
# [DEF:backend.src.services.git_service.GitService._update_repo_local_path:Function]
# @PURPOSE: Persist repository local_path in GitRepository table when record exists.
# @PRE: dashboard_id is valid integer.
# @POST: local_path is updated for existing record.
# @RETURN: None
# @RELATION: USES -> [GitRepository]
def _update_repo_local_path(self, dashboard_id: int, local_path: str) -> None:
try:
session = SessionLocal()
@@ -138,13 +144,14 @@ class GitService:
session.close()
except Exception as e:
logger.warning(f"[_update_repo_local_path][Coherence:Failed] {e}")
# [/DEF:_update_repo_local_path:Function]
# [/DEF:backend.src.services.git_service.GitService._update_repo_local_path:Function]
# [DEF:_migrate_repo_directory:Function]
# [DEF:backend.src.services.git_service.GitService._migrate_repo_directory:Function]
# @PURPOSE: Move legacy repository directory to target path and sync DB metadata.
# @PRE: source_path exists.
# @POST: Repository content available at target_path.
# @RETURN: str
# @RELATION: CALLS -> [GitService._update_repo_local_path]
def _migrate_repo_directory(self, dashboard_id: int, source_path: str, target_path: str) -> str:
source_abs = os.path.abspath(source_path)
target_abs = os.path.abspath(target_path)
@@ -168,13 +175,14 @@ class GitService:
f"[_migrate_repo_directory][Coherence:OK] Repository migrated for dashboard {dashboard_id}: {source_abs} -> {target_abs}"
)
return target_abs
# [/DEF:_migrate_repo_directory:Function]
# [/DEF:backend.src.services.git_service.GitService._migrate_repo_directory:Function]
# [DEF:_ensure_gitflow_branches:Function]
# [DEF:backend.src.services.git_service.GitService._ensure_gitflow_branches:Function]
# @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.
# @RETURN: None
# @RELATION: USES -> [Repo]
def _ensure_gitflow_branches(self, repo: Repo, dashboard_id: int) -> None:
with belief_scope("GitService._ensure_gitflow_branches"):
required_branches = ["main", "dev", "preprod"]
@@ -252,7 +260,7 @@ class GitService:
logger.warning(
f"[_ensure_gitflow_branches][Action] Could not checkout dev branch for dashboard {dashboard_id}: {e}"
)
# [/DEF:_ensure_gitflow_branches:Function]
# [/DEF:backend.src.services.git_service.GitService._ensure_gitflow_branches:Function]
# [DEF:backend.src.services.git_service.GitService._get_repo_path:Function]
# @PURPOSE: Resolves the local filesystem path for a dashboard's repository.
@@ -261,6 +269,9 @@ class GitService:
# @PRE: dashboard_id is an integer.
# @POST: Returns DB-local_path when present, otherwise base_path/<normalized repo_key>.
# @RETURN: str
# @RELATION: CALLS -> [GitService._normalize_repo_key]
# @RELATION: CALLS -> [GitService._migrate_repo_directory]
# @RELATION: CALLS -> [GitService._update_repo_local_path]
def _get_repo_path(self, dashboard_id: int, repo_key: Optional[str] = None) -> str:
with belief_scope("GitService._get_repo_path"):
if dashboard_id is None:
@@ -300,9 +311,9 @@ class GitService:
self._update_repo_local_path(dashboard_id, target_path)
return target_path
# [/DEF:_get_repo_path:Function]
# [/DEF:backend.src.services.git_service.GitService._get_repo_path:Function]
# [DEF:init_repo:Function]
# [DEF:backend.src.services.git_service.GitService.init_repo:Function]
# @PURPOSE: Initialize or clone a repository for a dashboard.
# @PARAM: dashboard_id (int)
# @PARAM: remote_url (str)
@@ -311,6 +322,8 @@ class GitService:
# @PRE: dashboard_id is int, remote_url is valid Git URL, pat is provided.
# @POST: Repository is cloned or opened at the local path.
# @RETURN: Repo - GitPython Repo object.
# @RELATION: CALLS -> [GitService._get_repo_path]
# @RELATION: CALLS -> [GitService._ensure_gitflow_branches]
def init_repo(self, dashboard_id: int, remote_url: str, pat: str, repo_key: Optional[str] = None) -> Repo:
with belief_scope("GitService.init_repo"):
self._ensure_base_path_exists()
@@ -344,13 +357,14 @@ class GitService:
repo = Repo.clone_from(auth_url, repo_path)
self._ensure_gitflow_branches(repo, dashboard_id)
return repo
# [/DEF:init_repo:Function]
# [/DEF:backend.src.services.git_service.GitService.init_repo:Function]
# [DEF:delete_repo:Function]
# [DEF:backend.src.services.git_service.GitService.delete_repo:Function]
# @PURPOSE: Remove local repository and DB binding for a dashboard.
# @PRE: dashboard_id is a valid integer.
# @POST: Local path is deleted when present and GitRepository row is removed.
# @RETURN: None
# @RELATION: CALLS -> [GitService._get_repo_path]
def delete_repo(self, dashboard_id: int) -> None:
with belief_scope("GitService.delete_repo"):
repo_path = self._get_repo_path(dashboard_id)
@@ -392,13 +406,14 @@ class GitService:
raise HTTPException(status_code=500, detail=f"Failed to delete repository: {str(e)}")
finally:
session.close()
# [/DEF:delete_repo:Function]
# [/DEF:backend.src.services.git_service.GitService.delete_repo:Function]
# [DEF:backend.src.services.git_service.GitService.get_repo:Function]
# @PURPOSE: Get Repo object for a dashboard.
# @PRE: Repository must exist on disk for the given dashboard_id.
# @POST: Returns a GitPython Repo instance for the dashboard.
# @RETURN: Repo
# @RELATION: CALLS -> [GitService._get_repo_path]
def get_repo(self, dashboard_id: int) -> Repo:
with belief_scope("GitService.get_repo"):
repo_path = self._get_repo_path(dashboard_id)
@@ -410,13 +425,14 @@ class GitService:
except Exception as e:
logger.error(f"[get_repo][Coherence:Failed] Failed to open repository at {repo_path}: {e}")
raise HTTPException(status_code=500, detail="Failed to open local Git repository")
# [/DEF:get_repo:Function]
# [/DEF:backend.src.services.git_service.GitService.get_repo:Function]
# [DEF:configure_identity:Function]
# [DEF:backend.src.services.git_service.GitService.configure_identity:Function]
# @PURPOSE: Configure repository-local Git committer identity for user-scoped operations.
# @PRE: dashboard_id repository exists; git_username/git_email may be empty.
# @POST: Repository config has user.name and user.email when both identity values are provided.
# @RETURN: None
# @RELATION: CALLS -> [GitService.get_repo]
def configure_identity(
self,
dashboard_id: int,
@@ -441,13 +457,14 @@ class GitService:
except Exception as e:
logger.error(f"[configure_identity][Coherence:Failed] Failed to configure git identity: {e}")
raise HTTPException(status_code=500, detail=f"Failed to configure git identity: {str(e)}")
# [/DEF:configure_identity:Function]
# [/DEF:backend.src.services.git_service.GitService.configure_identity:Function]
# [DEF:list_branches:Function]
# [DEF:backend.src.services.git_service.GitService.list_branches:Function]
# @PURPOSE: List all branches for a dashboard's repository.
# @PRE: Repository for dashboard_id exists.
# @POST: Returns a list of branch metadata dictionaries.
# @RETURN: List[dict]
# @RELATION: CALLS -> [GitService.get_repo]
def list_branches(self, dashboard_id: int) -> List[dict]:
with belief_scope("GitService.list_branches"):
repo = self.get_repo(dashboard_id)
@@ -495,14 +512,15 @@ class GitService:
})
return branches
# [/DEF:list_branches:Function]
# [/DEF:backend.src.services.git_service.GitService.list_branches:Function]
# [DEF:create_branch:Function]
# [DEF:backend.src.services.git_service.GitService.create_branch:Function]
# @PURPOSE: Create a new branch from an existing one.
# @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.
# @RELATION: CALLS -> [GitService.get_repo]
def create_branch(self, dashboard_id: int, name: str, from_branch: str = "main"):
with belief_scope("GitService.create_branch"):
repo = self.get_repo(dashboard_id)
@@ -531,25 +549,27 @@ class GitService:
except Exception as e:
logger.error(f"[create_branch][Coherence:Failed] {e}")
raise
# [/DEF:create_branch:Function]
# [/DEF:backend.src.services.git_service.GitService.create_branch:Function]
# [DEF:checkout_branch:Function]
# [DEF:backend.src.services.git_service.GitService.checkout_branch:Function]
# @PURPOSE: Switch to a specific branch.
# @PRE: Repository exists and the specified branch name exists.
# @POST: The repository working directory is updated to the specified branch.
# @RELATION: CALLS -> [GitService.get_repo]
def checkout_branch(self, dashboard_id: int, name: str):
with belief_scope("GitService.checkout_branch"):
repo = self.get_repo(dashboard_id)
logger.info(f"[checkout_branch][Action] Checking out branch {name}")
repo.git.checkout(name)
# [/DEF:checkout_branch:Function]
# [/DEF:backend.src.services.git_service.GitService.checkout_branch:Function]
# [DEF:commit_changes:Function]
# [DEF:backend.src.services.git_service.GitService.commit_changes:Function]
# @PURPOSE: Stage and commit changes.
# @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.
# @RELATION: CALLS -> [GitService.get_repo]
def commit_changes(self, dashboard_id: int, message: str, files: List[str] = None):
with belief_scope("GitService.commit_changes"):
repo = self.get_repo(dashboard_id)
@@ -568,13 +588,14 @@ class GitService:
repo.index.commit(message)
logger.info(f"[commit_changes][Coherence:OK] Committed changes with message: {message}")
# [/DEF:commit_changes:Function]
# [/DEF:backend.src.services.git_service.GitService.commit_changes:Function]
# [DEF:_extract_http_host:Function]
# [DEF:backend.src.services.git_service.GitService._extract_http_host:Function]
# @PURPOSE: Extract normalized host[:port] from HTTP(S) URL.
# @PRE: url_value may be empty.
# @POST: Returns lowercase host token or None.
# @RETURN: Optional[str]
# @RELATION: USES -> [urlparse]
def _extract_http_host(self, url_value: Optional[str]) -> Optional[str]:
normalized = str(url_value or "").strip()
if not normalized:
@@ -591,13 +612,14 @@ class GitService:
if parsed.port:
return f"{host.lower()}:{parsed.port}"
return host.lower()
# [/DEF:_extract_http_host:Function]
# [/DEF:backend.src.services.git_service.GitService._extract_http_host:Function]
# [DEF:_strip_url_credentials:Function]
# [DEF:backend.src.services.git_service.GitService._strip_url_credentials:Function]
# @PURPOSE: Remove credentials from URL while preserving scheme/host/path.
# @PRE: url_value may contain credentials.
# @POST: Returns URL without username/password.
# @RETURN: str
# @RELATION: USES -> [urlparse]
def _strip_url_credentials(self, url_value: str) -> str:
normalized = str(url_value or "").strip()
if not normalized:
@@ -612,13 +634,14 @@ class GitService:
if parsed.port:
host = f"{host}:{parsed.port}"
return parsed._replace(netloc=host).geturl()
# [/DEF:_strip_url_credentials:Function]
# [/DEF:backend.src.services.git_service.GitService._strip_url_credentials:Function]
# [DEF:_replace_host_in_url:Function]
# [DEF:backend.src.services.git_service.GitService._replace_host_in_url:Function]
# @PURPOSE: Replace source URL host with host from configured server URL.
# @PRE: source_url and config_url are HTTP(S) URLs.
# @POST: Returns source URL with updated host (credentials preserved) or None.
# @RETURN: Optional[str]
# @RELATION: USES -> [urlparse]
def _replace_host_in_url(self, source_url: Optional[str], config_url: Optional[str]) -> Optional[str]:
source = str(source_url or "").strip()
config = str(config_url or "").strip()
@@ -650,13 +673,16 @@ class GitService:
new_netloc = f"{auth_part}{target_host}"
return source_parsed._replace(netloc=new_netloc).geturl()
# [/DEF:_replace_host_in_url:Function]
# [/DEF:backend.src.services.git_service.GitService._replace_host_in_url:Function]
# [DEF:_align_origin_host_with_config:Function]
# [DEF:backend.src.services.git_service.GitService._align_origin_host_with_config:Function]
# @PURPOSE: Auto-align local origin host to configured Git server host when they drift.
# @PRE: origin remote exists.
# @POST: origin URL host updated and DB binding normalized when mismatch detected.
# @RETURN: Optional[str]
# @RELATION: CALLS -> [GitService._extract_http_host]
# @RELATION: CALLS -> [GitService._replace_host_in_url]
# @RELATION: CALLS -> [GitService._strip_url_credentials]
def _align_origin_host_with_config(
self,
dashboard_id: int,
@@ -716,12 +742,14 @@ class GitService:
)
return aligned_url
# [/DEF:_align_origin_host_with_config:Function]
# [/DEF:backend.src.services.git_service.GitService._align_origin_host_with_config:Function]
# [DEF:push_changes:Function]
# [DEF:backend.src.services.git_service.GitService.push_changes:Function]
# @PURPOSE: Push local commits to remote.
# @PRE: Repository exists and has an 'origin' remote.
# @POST: Local branch commits are pushed to origin.
# @RELATION: CALLS -> [GitService.get_repo]
# @RELATION: CALLS -> [GitService._align_origin_host_with_config]
def push_changes(self, dashboard_id: int):
with belief_scope("GitService.push_changes"):
repo = self.get_repo(dashboard_id)
@@ -829,12 +857,11 @@ class GitService:
except Exception as e:
logger.error(f"[push_changes][Coherence:Failed] Failed to push changes: {e}")
raise HTTPException(status_code=500, detail=f"Git push failed: {str(e)}")
# [/DEF:push_changes:Function]
# [/DEF:backend.src.services.git_service.GitService.push_changes:Function]
# [DEF:pull_changes:Function]
# @PURPOSE: Pull changes from remote.
# @PRE: Repository exists and has an 'origin' remote.
# @POST: Changes from origin are pulled and merged into the active branch.
# [DEF:backend.src.services.git_service.GitService._read_blob_text:Function]
# @PURPOSE: Read text from a Git blob.
# @RELATION: USES -> [Blob]
def _read_blob_text(self, blob: Blob) -> str:
with belief_scope("GitService._read_blob_text"):
if blob is None:
@@ -843,14 +870,22 @@ class GitService:
return blob.data_stream.read().decode("utf-8", errors="replace")
except Exception:
return ""
# [/DEF:backend.src.services.git_service.GitService._read_blob_text:Function]
# [DEF:backend.src.services.git_service.GitService._get_unmerged_file_paths:Function]
# @PURPOSE: List files with merge conflicts.
# @RELATION: USES -> [Repo]
def _get_unmerged_file_paths(self, repo: Repo) -> List[str]:
with belief_scope("GitService._get_unmerged_file_paths"):
try:
return sorted(list(repo.index.unmerged_blobs().keys()))
except Exception:
return []
# [/DEF:backend.src.services.git_service.GitService._get_unmerged_file_paths:Function]
# [DEF:backend.src.services.git_service.GitService._build_unfinished_merge_payload:Function]
# @PURPOSE: Build payload for unfinished merge state.
# @RELATION: CALLS -> [GitService._get_unmerged_file_paths]
def _build_unfinished_merge_payload(self, repo: Repo) -> Dict[str, Any]:
with belief_scope("GitService._build_unfinished_merge_payload"):
merge_head_path = os.path.join(repo.git_dir, "MERGE_HEAD")
@@ -900,7 +935,12 @@ class GitService:
"git merge --abort",
],
}
# [/DEF:backend.src.services.git_service.GitService._build_unfinished_merge_payload:Function]
# [DEF:backend.src.services.git_service.GitService.get_merge_status:Function]
# @PURPOSE: Get current merge status for a dashboard repository.
# @RELATION: CALLS -> [GitService.get_repo]
# @RELATION: CALLS -> [GitService._build_unfinished_merge_payload]
def get_merge_status(self, dashboard_id: int) -> Dict[str, Any]:
with belief_scope("GitService.get_merge_status"):
repo = self.get_repo(dashboard_id)
@@ -930,7 +970,12 @@ class GitService:
"merge_message_preview": payload["merge_message_preview"],
"conflicts_count": int(payload.get("conflicts_count") or 0),
}
# [/DEF:backend.src.services.git_service.GitService.get_merge_status:Function]
# [DEF:backend.src.services.git_service.GitService.get_merge_conflicts:Function]
# @PURPOSE: List all files with conflicts and their contents.
# @RELATION: CALLS -> [GitService.get_repo]
# @RELATION: CALLS -> [GitService._read_blob_text]
def get_merge_conflicts(self, dashboard_id: int) -> List[Dict[str, Any]]:
with belief_scope("GitService.get_merge_conflicts"):
repo = self.get_repo(dashboard_id)
@@ -952,7 +997,11 @@ class GitService:
}
)
return sorted(conflicts, key=lambda item: item["file_path"])
# [/DEF:backend.src.services.git_service.GitService.get_merge_conflicts:Function]
# [DEF:backend.src.services.git_service.GitService.resolve_merge_conflicts:Function]
# @PURPOSE: Resolve conflicts using specified strategy.
# @RELATION: CALLS -> [GitService.get_repo]
def resolve_merge_conflicts(self, dashboard_id: int, resolutions: List[Dict[str, Any]]) -> List[str]:
with belief_scope("GitService.resolve_merge_conflicts"):
repo = self.get_repo(dashboard_id)
@@ -986,7 +1035,11 @@ class GitService:
resolved_files.append(file_path)
return resolved_files
# [/DEF:backend.src.services.git_service.GitService.resolve_merge_conflicts:Function]
# [DEF:backend.src.services.git_service.GitService.abort_merge:Function]
# @PURPOSE: Abort ongoing merge.
# @RELATION: CALLS -> [GitService.get_repo]
def abort_merge(self, dashboard_id: int) -> Dict[str, Any]:
with belief_scope("GitService.abort_merge"):
repo = self.get_repo(dashboard_id)
@@ -999,7 +1052,12 @@ class GitService:
return {"status": "no_merge_in_progress"}
raise HTTPException(status_code=409, detail=f"Cannot abort merge: {details}")
return {"status": "aborted"}
# [/DEF:backend.src.services.git_service.GitService.abort_merge:Function]
# [DEF:backend.src.services.git_service.GitService.continue_merge:Function]
# @PURPOSE: Finalize merge after conflict resolution.
# @RELATION: CALLS -> [GitService.get_repo]
# @RELATION: CALLS -> [GitService._get_unmerged_file_paths]
def continue_merge(self, dashboard_id: int, message: Optional[str] = None) -> Dict[str, Any]:
with belief_scope("GitService.continue_merge"):
repo = self.get_repo(dashboard_id)
@@ -1032,7 +1090,14 @@ class GitService:
except Exception:
commit_hash = ""
return {"status": "committed", "commit_hash": commit_hash}
# [/DEF:backend.src.services.git_service.GitService.continue_merge:Function]
# [DEF:backend.src.services.git_service.GitService.pull_changes:Function]
# @PURPOSE: Pull changes from remote.
# @PRE: Repository exists and has an 'origin' remote.
# @POST: Changes from origin are pulled and merged into the active branch.
# @RELATION: CALLS -> [GitService.get_repo]
# @RELATION: CALLS -> [GitService._build_unfinished_merge_payload]
def pull_changes(self, dashboard_id: int):
with belief_scope("GitService.pull_changes"):
repo = self.get_repo(dashboard_id)
@@ -1110,13 +1175,14 @@ class GitService:
except Exception as e:
logger.error(f"[pull_changes][Coherence:Failed] Failed to pull changes: {e}")
raise HTTPException(status_code=500, detail=f"Git pull failed: {str(e)}")
# [/DEF:pull_changes:Function]
# [/DEF:backend.src.services.git_service.GitService.pull_changes:Function]
# [DEF:backend.src.services.git_service.GitService.get_status:Function]
# @PURPOSE: Get current repository status (dirty files, untracked, etc.)
# @PRE: Repository for dashboard_id exists.
# @POST: Returns a dictionary representing the Git status.
# @RETURN: dict
# @RELATION: CALLS -> [GitService.get_repo]
def get_status(self, dashboard_id: int) -> dict:
with belief_scope("GitService.get_status"):
repo = self.get_repo(dashboard_id)
@@ -1186,15 +1252,16 @@ class GitService:
"is_diverged": is_diverged,
"sync_state": sync_state,
}
# [/DEF:get_status:Function]
# [/DEF:backend.src.services.git_service.GitService.get_status:Function]
# [DEF:get_diff:Function]
# [DEF:backend.src.services.git_service.GitService.get_diff:Function]
# @PURPOSE: Generate diff for a file or the whole repository.
# @PARAM: file_path (str) - Optional specific file.
# @PARAM: staged (bool) - Whether to show staged changes.
# @PRE: Repository for dashboard_id exists.
# @POST: Returns the diff text as a string.
# @RETURN: str
# @RELATION: CALLS -> [GitService.get_repo]
def get_diff(self, dashboard_id: int, file_path: str = None, staged: bool = False) -> str:
with belief_scope("GitService.get_diff"):
repo = self.get_repo(dashboard_id)
@@ -1205,14 +1272,15 @@ class GitService:
if file_path:
return repo.git.diff(*diff_args, "--", file_path)
return repo.git.diff(*diff_args)
# [/DEF:get_diff:Function]
# [/DEF:backend.src.services.git_service.GitService.get_diff:Function]
# [DEF:get_commit_history:Function]
# [DEF:backend.src.services.git_service.GitService.get_commit_history:Function]
# @PURPOSE: Retrieve commit history for a repository.
# @PARAM: limit (int) - Max number of commits to return.
# @PRE: Repository for dashboard_id exists.
# @POST: Returns a list of dictionaries for each commit in history.
# @RETURN: List[dict]
# @RELATION: CALLS -> [GitService.get_repo]
def get_commit_history(self, dashboard_id: int, limit: int = 50) -> List[dict]:
with belief_scope("GitService.get_commit_history"):
repo = self.get_repo(dashboard_id)
@@ -1235,9 +1303,9 @@ class GitService:
logger.warning(f"[get_commit_history][Action] Could not retrieve commit history for dashboard {dashboard_id}: {e}")
return []
return commits
# [/DEF:get_commit_history:Function]
# [/DEF:backend.src.services.git_service.GitService.get_commit_history:Function]
# [DEF:test_connection:Function]
# [DEF:backend.src.services.git_service.GitService.test_connection:Function]
# @PURPOSE: Test connection to Git provider using PAT.
# @PARAM: provider (GitProvider)
# @PARAM: url (str)
@@ -1245,6 +1313,7 @@ class GitService:
# @PRE: provider is valid; url is a valid HTTP(S) URL; pat is provided.
# @POST: Returns True if connection to the provider's API succeeds.
# @RETURN: bool
# @RELATION: USES -> [httpx.AsyncClient]
async def test_connection(self, provider: GitProvider, url: str, pat: str) -> bool:
with belief_scope("GitService.test_connection"):
# Check for offline mode or local-only URLs
@@ -1285,9 +1354,9 @@ class GitService:
except Exception as e:
logger.error(f"[test_connection][Coherence:Failed] Error testing git connection: {e}")
return False
# [/DEF:test_connection:Function]
# [/DEF:backend.src.services.git_service.GitService.test_connection:Function]
# [DEF:_normalize_git_server_url:Function]
# [DEF:backend.src.services.git_service.GitService._normalize_git_server_url:Function]
# @PURPOSE: Normalize Git server URL for provider API calls.
# @PRE: raw_url is non-empty.
# @POST: Returns URL without trailing slash.
@@ -1297,9 +1366,9 @@ class GitService:
if not normalized:
raise HTTPException(status_code=400, detail="Git server URL is required")
return normalized.rstrip("/")
# [/DEF:_normalize_git_server_url:Function]
# [/DEF:backend.src.services.git_service.GitService._normalize_git_server_url:Function]
# [DEF:_gitea_headers:Function]
# [DEF:backend.src.services.git_service.GitService._gitea_headers:Function]
# @PURPOSE: Build Gitea API authorization headers.
# @PRE: pat is provided.
# @POST: Returns headers with token auth.
@@ -1313,13 +1382,15 @@ class GitService:
"Content-Type": "application/json",
"Accept": "application/json",
}
# [/DEF:_gitea_headers:Function]
# [/DEF:backend.src.services.git_service.GitService._gitea_headers:Function]
# [DEF:_gitea_request:Function]
# [DEF:backend.src.services.git_service.GitService._gitea_request:Function]
# @PURPOSE: Execute HTTP request against Gitea API with stable error mapping.
# @PRE: method and endpoint are valid.
# @POST: Returns decoded JSON payload.
# @RETURN: Any
# @RELATION: CALLS -> [GitService._normalize_git_server_url]
# @RELATION: CALLS -> [GitService._gitea_headers]
async def _gitea_request(
self,
method: str,
@@ -1361,26 +1432,28 @@ class GitService:
if response.status_code == 204:
return None
return response.json()
# [/DEF:_gitea_request:Function]
# [/DEF:backend.src.services.git_service.GitService._gitea_request:Function]
# [DEF:get_gitea_current_user:Function]
# [DEF:backend.src.services.git_service.GitService.get_gitea_current_user:Function]
# @PURPOSE: Resolve current Gitea user for PAT.
# @PRE: server_url and pat are valid.
# @POST: Returns current username.
# @RETURN: str
# @RELATION: CALLS -> [GitService._gitea_request]
async def get_gitea_current_user(self, server_url: str, pat: str) -> str:
payload = await self._gitea_request("GET", server_url, pat, "/user")
username = payload.get("login") or payload.get("username")
if not username:
raise HTTPException(status_code=500, detail="Failed to resolve Gitea username")
return str(username)
# [/DEF:get_gitea_current_user:Function]
# [/DEF:backend.src.services.git_service.GitService.get_gitea_current_user:Function]
# [DEF:list_gitea_repositories:Function]
# [DEF:backend.src.services.git_service.GitService.list_gitea_repositories:Function]
# @PURPOSE: List repositories visible to authenticated Gitea user.
# @PRE: server_url and pat are valid.
# @POST: Returns repository list from Gitea.
# @RETURN: List[dict]
# @RELATION: CALLS -> [GitService._gitea_request]
async def list_gitea_repositories(self, server_url: str, pat: str) -> List[dict]:
payload = await self._gitea_request(
"GET",
@@ -1391,13 +1464,14 @@ class GitService:
if not isinstance(payload, list):
return []
return payload
# [/DEF:list_gitea_repositories:Function]
# [/DEF:backend.src.services.git_service.GitService.list_gitea_repositories:Function]
# [DEF:create_gitea_repository:Function]
# [DEF:backend.src.services.git_service.GitService.create_gitea_repository:Function]
# @PURPOSE: Create repository in Gitea for authenticated user.
# @PRE: name is non-empty and PAT has repo creation permission.
# @POST: Returns created repository payload.
# @RETURN: dict
# @RELATION: CALLS -> [GitService._gitea_request]
async def create_gitea_repository(
self,
server_url: str,
@@ -1427,12 +1501,13 @@ class GitService:
if not isinstance(created, dict):
raise HTTPException(status_code=500, detail="Unexpected Gitea response while creating repository")
return created
# [/DEF:create_gitea_repository:Function]
# [/DEF:backend.src.services.git_service.GitService.create_gitea_repository:Function]
# [DEF:delete_gitea_repository:Function]
# [DEF:backend.src.services.git_service.GitService.delete_gitea_repository:Function]
# @PURPOSE: Delete repository in Gitea.
# @PRE: owner and repo_name are non-empty.
# @POST: Repository deleted on Gitea server.
# @RELATION: CALLS -> [GitService._gitea_request]
async def delete_gitea_repository(
self,
server_url: str,
@@ -1448,13 +1523,14 @@ class GitService:
pat,
f"/repos/{owner}/{repo_name}",
)
# [/DEF:delete_gitea_repository:Function]
# [/DEF:backend.src.services.git_service.GitService.delete_gitea_repository:Function]
# [DEF:_gitea_branch_exists:Function]
# [DEF:backend.src.services.git_service.GitService._gitea_branch_exists:Function]
# @PURPOSE: Check whether a branch exists in Gitea repository.
# @PRE: owner/repo/branch are non-empty.
# @POST: Returns True when branch exists, False when 404.
# @RETURN: bool
# @RELATION: CALLS -> [GitService._gitea_request]
async def _gitea_branch_exists(
self,
server_url: str,
@@ -1473,13 +1549,14 @@ class GitService:
if exc.status_code == 404:
return False
raise
# [/DEF:_gitea_branch_exists:Function]
# [/DEF:backend.src.services.git_service.GitService._gitea_branch_exists:Function]
# [DEF:_build_gitea_pr_404_detail:Function]
# [DEF:backend.src.services.git_service.GitService._build_gitea_pr_404_detail:Function]
# @PURPOSE: Build actionable error detail for Gitea PR 404 responses.
# @PRE: owner/repo/from_branch/to_branch are provided.
# @POST: Returns specific branch-missing message when detected.
# @RETURN: Optional[str]
# @RELATION: CALLS -> [GitService._gitea_branch_exists]
async def _build_gitea_pr_404_detail(
self,
server_url: str,
@@ -1508,13 +1585,14 @@ class GitService:
if not target_exists:
return f"Gitea branch not found: target branch '{to_branch}' in {owner}/{repo}"
return None
# [/DEF:_build_gitea_pr_404_detail:Function]
# [/DEF:backend.src.services.git_service.GitService._build_gitea_pr_404_detail:Function]
# [DEF:create_github_repository:Function]
# [DEF:backend.src.services.git_service.GitService.create_github_repository:Function]
# @PURPOSE: Create repository in GitHub or GitHub Enterprise.
# @PRE: PAT has repository create permission.
# @POST: Returns created repository payload.
# @RETURN: dict
# @RELATION: CALLS -> [GitService._normalize_git_server_url]
async def create_github_repository(
self,
server_url: str,
@@ -1560,13 +1638,14 @@ class GitService:
pass
raise HTTPException(status_code=response.status_code, detail=f"GitHub API error: {detail}")
return response.json()
# [/DEF:create_github_repository:Function]
# [/DEF:backend.src.services.git_service.GitService.create_github_repository:Function]
# [DEF:create_gitlab_repository:Function]
# [DEF:backend.src.services.git_service.GitService.create_gitlab_repository:Function]
# @PURPOSE: Create repository(project) in GitLab.
# @PRE: PAT has api scope.
# @POST: Returns created repository payload.
# @RETURN: dict
# @RELATION: CALLS -> [GitService._normalize_git_server_url]
async def create_gitlab_repository(
self,
server_url: str,
@@ -1620,13 +1699,14 @@ class GitService:
if "full_name" not in data:
data["full_name"] = data.get("path_with_namespace") or data.get("name")
return data
# [/DEF:create_gitlab_repository:Function]
# [/DEF:backend.src.services.git_service.GitService.create_gitlab_repository:Function]
# [DEF:_parse_remote_repo_identity:Function]
# [DEF:backend.src.services.git_service.GitService._parse_remote_repo_identity:Function]
# @PURPOSE: Parse owner/repo from remote URL for Git server API operations.
# @PRE: remote_url is a valid git URL.
# @POST: Returns owner/repo tokens.
# @RETURN: Dict[str, str]
# @RELATION: USES -> [urlparse]
def _parse_remote_repo_identity(self, remote_url: str) -> Dict[str, str]:
normalized = str(remote_url or "").strip()
if not normalized:
@@ -1655,13 +1735,14 @@ class GitService:
"namespace": namespace,
"full_name": f"{namespace}/{repo}",
}
# [/DEF:_parse_remote_repo_identity:Function]
# [/DEF:backend.src.services.git_service.GitService._parse_remote_repo_identity:Function]
# [DEF:backend.src.services.git_service.GitService._derive_server_url_from_remote:Function]
# @PURPOSE: Build API base URL from remote repository URL without credentials.
# @PRE: remote_url may be any git URL.
# @POST: Returns normalized http(s) base URL or None when derivation is impossible.
# @RETURN: Optional[str]
# @RELATION: USES -> [urlparse]
def _derive_server_url_from_remote(self, remote_url: str) -> Optional[str]:
normalized = str(remote_url or "").strip()
if not normalized or normalized.startswith("git@"):
@@ -1677,13 +1758,14 @@ class GitService:
if parsed.port:
netloc = f"{netloc}:{parsed.port}"
return f"{parsed.scheme}://{netloc}".rstrip("/")
# [/DEF:_derive_server_url_from_remote:Function]
# [/DEF:backend.src.services.git_service.GitService._derive_server_url_from_remote:Function]
# [DEF:promote_direct_merge:Function]
# [DEF:backend.src.services.git_service.GitService.promote_direct_merge:Function]
# @PURPOSE: Perform direct merge between branches in local repo and push target branch.
# @PRE: Repository exists and both branches are valid.
# @POST: Target branch contains merged changes from source branch.
# @RETURN: Dict[str, Any]
# @RELATION: CALLS -> [GitService.get_repo]
def promote_direct_merge(
self,
dashboard_id: int,
@@ -1742,13 +1824,18 @@ class GitService:
"to_branch": target,
"status": "merged",
}
# [/DEF:promote_direct_merge:Function]
# [/DEF:backend.src.services.git_service.GitService.promote_direct_merge:Function]
# [DEF:backend.src.services.git_service.GitService.create_gitea_pull_request:Function]
# @PURPOSE: Create pull request in Gitea.
# @PRE: Config and remote URL are valid.
# @POST: Returns normalized PR metadata.
# @RETURN: Dict[str, Any]
# @RELATION: CALLS -> [GitService._parse_remote_repo_identity]
# @RELATION: CALLS -> [GitService._gitea_request]
# @RELATION: CALLS -> [GitService._derive_server_url_from_remote]
# @RELATION: CALLS -> [GitService._normalize_git_server_url]
# @RELATION: CALLS -> [GitService._build_gitea_pr_404_detail]
async def create_gitea_pull_request(
self,
server_url: str,
@@ -1830,13 +1917,15 @@ class GitService:
"url": data.get("html_url") or data.get("url"),
"status": data.get("state") or "open",
}
# [/DEF:create_gitea_pull_request:Function]
# [/DEF:backend.src.services.git_service.GitService.create_gitea_pull_request:Function]
# [DEF:backend.src.services.git_service.GitService.create_github_pull_request:Function]
# @PURPOSE: Create pull request in GitHub or GitHub Enterprise.
# @PRE: Config and remote URL are valid.
# @POST: Returns normalized PR metadata.
# @RETURN: Dict[str, Any]
# @RELATION: CALLS -> [GitService._parse_remote_repo_identity]
# @RELATION: CALLS -> [GitService._normalize_git_server_url]
async def create_github_pull_request(
self,
server_url: str,
@@ -1884,13 +1973,15 @@ class GitService:
"url": data.get("html_url") or data.get("url"),
"status": data.get("state") or "open",
}
# [/DEF:create_github_pull_request:Function]
# [/DEF:backend.src.services.git_service.GitService.create_github_pull_request:Function]
# [DEF:backend.src.services.git_service.GitService.create_gitlab_merge_request:Function]
# @PURPOSE: Create merge request in GitLab.
# @PRE: Config and remote URL are valid.
# @POST: Returns normalized MR metadata.
# @RETURN: Dict[str, Any]
# @RELATION: CALLS -> [GitService._parse_remote_repo_identity]
# @RELATION: CALLS -> [GitService._normalize_git_server_url]
async def create_gitlab_merge_request(
self,
server_url: str,
@@ -1938,7 +2029,7 @@ class GitService:
"url": data.get("web_url") or data.get("url"),
"status": data.get("state") or "opened",
}
# [/DEF:create_gitlab_merge_request:Function]
# [/DEF:backend.src.services.git_service.GitService.create_gitlab_merge_request:Function]
# [/DEF:GitService:Class]
# [/DEF:backend.src.services.git_service.GitService:Class]
# [/DEF:backend.src.services.git_service:Module]