busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases 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 c53c3f77cc
commit 7e4124bc3f
19 changed files with 480 additions and 257 deletions
Download Patch File Download Diff File Expand all files Collapse all files

94
backend/src/services/auth_service.py
View File

@@ -1,5 +1,4 @@
# [DEF:backend.src.services.auth_service:Module]
#
# @COMPLEXITY: 5
# @SEMANTICS: auth, service, business-logic, login, jwt, adfs, jit-provisioning
# @PURPOSE: Orchestrates credential authentication and ADFS JIT user provisioning.
@@ -9,28 +8,29 @@
# @RELATION: [DEPENDS_ON] ->[backend.src.core.auth.jwt.create_access_token]
# @RELATION: [DEPENDS_ON] ->[backend.src.models.auth.User]
# @RELATION: [DEPENDS_ON] ->[backend.src.models.auth.Role]
#
# @INVARIANT: Authentication succeeds only for active users with valid credentials; issued sessions encode subject and scopes from assigned roles.
# @PRE: Core auth models and security utilities available.
# @POST: User identity verified and session tokens issued according to role scopes.
# @SIDE_EFFECT: Writes last login timestamps and JIT-provisions external users.
# @DATA_CONTRACT: [Credentials | ADFSClaims] -> [UserEntity | SessionToken]
# [SECTION: IMPORTS]
from typing import Dict, Any
from typing import Dict, Any, Optional, List
from datetime import datetime
from sqlalchemy.orm import Session
from ..models.auth import User, Role
from ..core.auth.repository import AuthRepository
from ..core.auth.security import verify_password
from ..core.auth.jwt import create_access_token
from ..core.auth.logger import log_security_event
from ..models.auth import User, Role
from ..core.logger import belief_scope
# [/SECTION]
# [DEF:AuthService:Class]
# @COMPLEXITY: 3
# @PURPOSE: Provides high-level authentication services.
class AuthService:
# [DEF:__init__:Function]
# [DEF:AuthService.__init__:Function]
# @COMPLEXITY: 1
# @PURPOSE: Initializes the authentication service with repository access over an active DB session.
# @PRE: db is a valid SQLAlchemy Session instance bound to the auth persistence context.
@@ -39,10 +39,11 @@ class AuthService:
# @DATA_CONTRACT: Input(Session) -> Model(AuthRepository)
# @PARAM: db (Session) - SQLAlchemy session.
def __init__(self, db: Session):
self.db = db
self.repo = AuthRepository(db)
# [/DEF:__init__:Function]
# [/DEF:AuthService.__init__:Function]
# [DEF:authenticate_user:Function]
# [DEF:AuthService.authenticate_user:Function]
# @COMPLEXITY: 3
# @PURPOSE: Validates credentials and account state for local username/password authentication.
# @PRE: username and password are non-empty credential inputs.
@@ -52,23 +53,24 @@ class AuthService:
# @PARAM: username (str) - The username.
# @PARAM: password (str) - The plain password.
# @RETURN: Optional[User] - The authenticated user or None.
def authenticate_user(self, username: str, password: str):
with belief_scope("AuthService.authenticate_user"):
def authenticate_user(self, username: str, password: str) -> Optional[User]:
with belief_scope("auth.authenticate_user"):
user = self.repo.get_user_by_username(username)
if not user:
if not user or not user.is_active:
return None
if not user.is_active:
return None
if not user.password_hash or not verify_password(password, user.password_hash):
if not verify_password(password, user.password_hash):
return None
self.repo.update_last_login(user)
# Update last login
user.last_login = datetime.utcnow()
self.db.commit()
self.db.refresh(user)
return user
# [/DEF:authenticate_user:Function]
# [/DEF:AuthService.authenticate_user:Function]
# [DEF:create_session:Function]
# [DEF:AuthService.create_session:Function]
# @COMPLEXITY: 3
# @PURPOSE: Issues an access token payload for an already authenticated user.
# @PRE: user is a valid User entity containing username and iterable roles with role.name values.
@@ -77,24 +79,16 @@ class AuthService:
# @DATA_CONTRACT: Input(User) -> Output(Dict[str, str]{access_token, token_type})
# @PARAM: user (User) - The authenticated user.
# @RETURN: Dict[str, str] - Session data.
def create_session(self, user) -> Dict[str, str]:
with belief_scope("AuthService.create_session"):
# Collect role names for scopes
scopes = [role.name for role in user.roles]
token_data = {
"sub": user.username,
"scopes": scopes
}
access_token = create_access_token(data=token_data)
return {
"access_token": access_token,
"token_type": "bearer"
}
# [/DEF:create_session:Function]
def create_session(self, user: User) -> Dict[str, str]:
with belief_scope("auth.create_session"):
roles = [role.name for role in user.roles]
access_token = create_access_token(
data={"sub": user.username, "scopes": roles}
)
return {"access_token": access_token, "token_type": "bearer"}
# [/DEF:AuthService.create_session:Function]
# [DEF:provision_adfs_user:Function]
# [DEF:AuthService.provision_adfs_user:Function]
# @COMPLEXITY: 3
# @PURPOSE: Performs ADFS Just-In-Time provisioning and role synchronization from AD group mappings.
# @PRE: user_info contains identity claims where at least one of 'upn' or 'email' is present; 'groups' may be absent.
@@ -104,32 +98,34 @@ class AuthService:
# @PARAM: user_info (Dict[str, Any]) - Claims from ADFS token.
# @RETURN: User - The provisioned user.
def provision_adfs_user(self, user_info: Dict[str, Any]) -> User:
with belief_scope("AuthService.provision_adfs_user"):
with belief_scope("auth.provision_adfs_user"):
username = user_info.get("upn") or user_info.get("email")
email = user_info.get("email")
ad_groups = user_info.get("groups", [])
groups = user_info.get("groups", [])
user = self.repo.get_user_by_username(username)
if not user:
user = User(
username=username,
email=email,
full_name=user_info.get("name"),
auth_source="ADFS",
is_active=True
is_active=True,
is_ad_user=True
)
self.repo.db.add(user)
# Update roles based on group mappings
from ..models.auth import ADGroupMapping
mapped_roles = self.repo.db.query(Role).join(ADGroupMapping).filter(
ADGroupMapping.ad_group.in_(ad_groups)
).all()
self.db.add(user)
log_security_event("USER_PROVISIONED", username, {"source": "ADFS"})
# Sync roles from AD groups
mapped_roles = self.repo.get_roles_by_ad_groups(groups)
user.roles = mapped_roles
self.repo.db.commit()
self.repo.db.refresh(user)
user.last_login = datetime.utcnow()
self.db.commit()
self.db.refresh(user)
return user
# [/DEF:provision_adfs_user:Function]
# [/DEF:AuthService.provision_adfs_user:Function]
# [/DEF:AuthService:Class]
# [/DEF:backend.src.services.auth_service:Module]

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]