chore(lint): apply ruff --fix (4443 auto-fixes)

Auto-fixed categories:
- F401: unused imports removed
- I001: import blocks sorted
- W293: trailing whitespace stripped
- UP035: deprecated typing imports replaced
- SIM: simplify suggestions applied
- ARG: unused args prefixed with underscore
- T201: print statements removed
- F841: unused variables removed
- RUF059: unpacked variables prefixed

Remaining ~1500 unfixable errors (C901, B904, N806, E402) require manual work.

Backend smoke tests: 13/13 passed.
This commit is contained in:
2026-05-14 11:20:17 +03:00
parent a9a5eff518
commit c6189876b3
337 changed files with 4677 additions and 4515 deletions

View File

@@ -19,12 +19,12 @@
from ._base import GitServiceBase
from ._branch import GitServiceBranchMixin
from ._sync import GitServiceSyncMixin
from ._status import GitServiceStatusMixin
from ._merge import GitServiceMergeMixin
from ._url import GitServiceUrlMixin
from ._gitea import GitServiceGiteaMixin
from ._merge import GitServiceMergeMixin
from ._remote_providers import GitServiceGithubMixin, GitServiceGitlabMixin
from ._status import GitServiceStatusMixin
from ._sync import GitServiceSyncMixin
from ._url import GitServiceUrlMixin
__all__ = ["GitService"]

View File

@@ -10,14 +10,15 @@ import os
import re
import shutil
from pathlib import Path
from typing import Any, Dict, List, Optional
from fastapi import HTTPException
from git import Repo
from git.exc import InvalidGitRepositoryError, NoSuchPathError
from fastapi import HTTPException
from src.core.logger import logger, belief_scope
from src.models.git import GitRepository
from src.models.config import AppConfigRecord
from src.core.database import SessionLocal
from src.core.logger import belief_scope, logger
from src.models.config import AppConfigRecord
from src.models.git import GitRepository
# #region GitServiceBase [C:3] [TYPE Class]
@@ -94,7 +95,7 @@ class GitServiceBase:
# @PRE: repo_key can be None/empty.
# @POST: Returns normalized non-empty key.
# @RETURN: str
def _normalize_repo_key(self, repo_key: Optional[str]) -> str:
def _normalize_repo_key(self, repo_key: str | None) -> str:
raw_key = str(repo_key or "").strip().lower()
normalized = re.sub(r"[^a-z0-9._-]+", "-", raw_key).strip("._-")
return normalized or "dashboard"
@@ -154,7 +155,7 @@ class GitServiceBase:
# @PRE: dashboard_id is an integer.
# @POST: Returns DB-local_path when present, otherwise base_path/<normalized repo_key>.
# @RETURN: str
def _get_repo_path(self, dashboard_id: int, repo_key: Optional[str] = None) -> str:
def _get_repo_path(self, dashboard_id: int, repo_key: str | None = None) -> str:
with belief_scope("GitService._get_repo_path"):
if dashboard_id is None:
raise ValueError("dashboard_id cannot be None")
@@ -199,7 +200,7 @@ class GitServiceBase:
# @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
def init_repo(self, dashboard_id: int, remote_url: str, pat: str, repo_key: Optional[str] = None) -> Repo:
def init_repo(self, dashboard_id: int, remote_url: str, pat: str, repo_key: str | None = None) -> Repo:
with belief_scope("GitService.init_repo"):
self._ensure_base_path_exists()
repo_path = self._get_repo_path(dashboard_id, repo_key=repo_key or str(dashboard_id))
@@ -269,7 +270,7 @@ class GitServiceBase:
except Exception as e:
session.rollback()
logger.error(f"[delete_repo][Coherence:Failed] Failed to delete repository for dashboard {dashboard_id}: {e}")
raise HTTPException(status_code=500, detail=f"Failed to delete repository: {str(e)}")
raise HTTPException(status_code=500, detail=f"Failed to delete repository: {e!s}")
finally:
session.close()
# endregion delete_repo
@@ -296,7 +297,7 @@ class GitServiceBase:
# @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.
def configure_identity(self, dashboard_id: int, git_username: Optional[str], git_email: Optional[str]) -> None:
def configure_identity(self, dashboard_id: int, git_username: str | None, git_email: str | None) -> None:
with belief_scope("GitService.configure_identity"):
normalized_username = str(git_username or "").strip()
normalized_email = str(git_email or "").strip()
@@ -310,7 +311,7 @@ class GitServiceBase:
logger.reason(f"Applied repository-local git identity for dashboard {dashboard_id}", extra={"src": "configure_identity"})
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)}")
raise HTTPException(status_code=500, detail=f"Failed to configure git identity: {e!s}")
# endregion configure_identity
# #endregion GitServiceBase
# #endregion GitServiceBase

View File

@@ -4,12 +4,12 @@
# @RELATION USED_BY -> [GitService]
import os
from typing import Any, Dict, List, Optional
from datetime import datetime
from git import Repo
from git.exc import GitCommandError
from fastapi import HTTPException
from src.core.logger import logger, belief_scope
from git import Repo
from src.core.logger import belief_scope, logger
# #region GitServiceBranchMixin [C:3] [TYPE Class]
@@ -77,7 +77,7 @@ class GitServiceBranchMixin:
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: {str(e)}",
detail=f"Failed to create default branch '{branch_name}' on remote: {e!s}",
)
try:
repo.git.checkout("dev")
@@ -91,7 +91,7 @@ class GitServiceBranchMixin:
# @PRE: Repository for dashboard_id exists.
# @POST: Returns a list of branch metadata dictionaries.
# @RETURN: List[dict]
def list_branches(self, dashboard_id: int) -> List[dict]:
def list_branches(self, dashboard_id: int) -> list[dict]:
with 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"})
@@ -174,7 +174,7 @@ class GitServiceBranchMixin:
# @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):
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)
if not repo.is_dirty(untracked_files=True) and not files:

View File

@@ -3,11 +3,13 @@
# @BRIEF Gitea API operations for GitService — connection testing, repository CRUD, and pull request creation.
# @RELATION USED_BY -> [GitService]
import httpx
from typing import Any, Dict, List, Optional
from typing import Any
from urllib.parse import quote
import httpx
from fastapi import HTTPException
from src.core.logger import logger, belief_scope
from src.core.logger import belief_scope, logger
from src.models.git import GitProvider
@@ -61,7 +63,7 @@ class GitServiceGiteaMixin:
# @PRE: pat is provided.
# @POST: Returns headers with token auth.
# @RETURN: Dict[str, str]
def _gitea_headers(self, pat: str) -> Dict[str, str]:
def _gitea_headers(self, pat: str) -> dict[str, str]:
token = (pat or "").strip()
if not token:
raise HTTPException(status_code=400, detail="Git PAT is required for Gitea operations")
@@ -79,7 +81,7 @@ class GitServiceGiteaMixin:
# @RETURN: Any
async def _gitea_request(
self, method: str, server_url: str, pat: str, endpoint: str,
payload: Optional[Dict[str, Any]] = None,
payload: dict[str, Any] | None = None,
) -> Any:
base_url = self._normalize_git_server_url(server_url)
url = f"{base_url}/api/v1{endpoint}"
@@ -89,7 +91,7 @@ class GitServiceGiteaMixin:
response = await client.request(method=method, url=url, headers=headers, json=payload)
except Exception as e:
logger.error(f"[gitea_request][Coherence:Failed] Network error: {e}")
raise HTTPException(status_code=503, detail=f"Gitea API is unavailable: {str(e)}")
raise HTTPException(status_code=503, detail=f"Gitea API is unavailable: {e!s}")
if response.status_code >= 400:
detail = response.text
try:
@@ -122,7 +124,7 @@ class GitServiceGiteaMixin:
# @PRE: server_url and pat are valid.
# @POST: Returns repository list from Gitea.
# @RETURN: List[dict]
async def list_gitea_repositories(self, server_url: str, pat: str) -> List[dict]:
async def list_gitea_repositories(self, server_url: str, pat: str) -> list[dict]:
payload = await self._gitea_request("GET", server_url, pat, "/user/repos?limit=100&page=1")
if not isinstance(payload, list):
return []
@@ -136,9 +138,9 @@ class GitServiceGiteaMixin:
# @RETURN: dict
async def create_gitea_repository(
self, server_url: str, pat: str, name: str, private: bool = True,
description: Optional[str] = None, auto_init: bool = True, default_branch: Optional[str] = "main",
) -> Dict[str, Any]:
payload: Dict[str, Any] = {"name": name, "private": bool(private), "auto_init": bool(auto_init)}
description: str | None = None, auto_init: bool = True, default_branch: str | None = "main",
) -> dict[str, Any]:
payload: dict[str, Any] = {"name": name, "private": bool(private), "auto_init": bool(auto_init)}
if description:
payload["description"] = description
if default_branch:
@@ -184,7 +186,7 @@ class GitServiceGiteaMixin:
# @RETURN: Optional[str]
async def _build_gitea_pr_404_detail(
self, server_url: str, pat: str, owner: str, repo: str, from_branch: str, to_branch: str,
) -> Optional[str]:
) -> str | None:
source_exists = await self._gitea_branch_exists(
server_url=server_url, pat=pat, owner=owner, repo=repo, branch=from_branch,
)
@@ -205,8 +207,8 @@ class GitServiceGiteaMixin:
# @RETURN: Dict[str, Any]
async def create_gitea_pull_request(
self, server_url: str, pat: str, remote_url: str, from_branch: str, to_branch: str,
title: str, description: Optional[str] = None,
) -> Dict[str, Any]:
title: str, description: str | None = None,
) -> dict[str, Any]:
identity = self._parse_remote_repo_identity(remote_url)
req_payload = {"title": title, "head": from_branch, "base": to_branch, "body": description or ""}
endpoint = f"/repos/{identity['owner']}/{identity['repo']}/pulls"

View File

@@ -5,12 +5,14 @@
import os
from pathlib import Path
from typing import Any, Dict, List, Optional
from typing import Any
from fastapi import HTTPException
from git import Repo
from git.exc import GitCommandError
from git.objects.blob import Blob
from fastapi import HTTPException
from src.core.logger import logger, belief_scope
from src.core.logger import belief_scope
# #region GitServiceMergeMixin [C:3] [TYPE Class]
@@ -30,7 +32,7 @@ class GitServiceMergeMixin:
# region _get_unmerged_file_paths [TYPE Function]
# @PURPOSE: List files with merge conflicts.
def _get_unmerged_file_paths(self, repo: Repo) -> List[str]:
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()))
@@ -40,7 +42,7 @@ class GitServiceMergeMixin:
# region _build_unfinished_merge_payload [TYPE Function]
# @PURPOSE: Build payload for unfinished merge state.
def _build_unfinished_merge_payload(self, repo: Repo) -> Dict[str, Any]:
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")
merge_head_value = ""
@@ -87,7 +89,7 @@ class GitServiceMergeMixin:
# region get_merge_status [TYPE Function]
# @PURPOSE: Get current merge status for a dashboard repository.
def get_merge_status(self, dashboard_id: int) -> Dict[str, Any]:
def get_merge_status(self, dashboard_id: int) -> dict[str, Any]:
with belief_scope("GitService.get_merge_status"):
repo = self.get_repo(dashboard_id)
merge_head_path = os.path.join(repo.git_dir, "MERGE_HEAD")
@@ -120,7 +122,7 @@ class GitServiceMergeMixin:
# region get_merge_conflicts [TYPE Function]
# @PURPOSE: List all files with conflicts and their contents.
def get_merge_conflicts(self, dashboard_id: int) -> List[Dict[str, Any]]:
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)
conflicts = []
@@ -143,10 +145,10 @@ class GitServiceMergeMixin:
# region resolve_merge_conflicts [TYPE Function]
# @PURPOSE: Resolve conflicts using specified strategy.
def resolve_merge_conflicts(self, dashboard_id: int, resolutions: List[Dict[str, Any]]) -> List[str]:
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)
resolved_files: List[str] = []
resolved_files: list[str] = []
repo_root = os.path.abspath(str(repo.working_tree_dir or ""))
if not repo_root:
raise HTTPException(status_code=500, detail="Repository working tree directory is unavailable")
@@ -176,7 +178,7 @@ class GitServiceMergeMixin:
# region abort_merge [TYPE Function]
# @PURPOSE: Abort ongoing merge.
def abort_merge(self, dashboard_id: int) -> Dict[str, Any]:
def abort_merge(self, dashboard_id: int) -> dict[str, Any]:
with belief_scope("GitService.abort_merge"):
repo = self.get_repo(dashboard_id)
try:
@@ -192,7 +194,7 @@ class GitServiceMergeMixin:
# region continue_merge [TYPE Function]
# @PURPOSE: Finalize merge after conflict resolution.
def continue_merge(self, dashboard_id: int, message: Optional[str] = None) -> Dict[str, Any]:
def continue_merge(self, dashboard_id: int, message: str | None = None) -> dict[str, Any]:
with belief_scope("GitService.continue_merge"):
repo = self.get_repo(dashboard_id)
unmerged_files = self._get_unmerged_file_paths(repo)
@@ -230,7 +232,7 @@ class GitServiceMergeMixin:
# @PRE: Repository exists and both branches are valid.
# @POST: Target branch contains merged changes from source branch.
# @RETURN: Dict[str, Any]
def promote_direct_merge(self, dashboard_id: int, from_branch: str, to_branch: str) -> Dict[str, Any]:
def promote_direct_merge(self, dashboard_id: int, from_branch: str, to_branch: str) -> dict[str, Any]:
with belief_scope("GitService.promote_direct_merge"):
if not from_branch or not to_branch:
raise HTTPException(status_code=400, detail="from_branch and to_branch are required")

View File

@@ -3,11 +3,11 @@
# @BRIEF GitHub and GitLab provider operations for GitService — repository creation and PR/MR creation.
# @RELATION USED_BY -> [GitService]
import httpx
from typing import Any, Dict, Optional
from typing import Any
from urllib.parse import quote
import httpx
from fastapi import HTTPException
from src.core.logger import logger, belief_scope
# #region GitServiceGithubMixin [C:3] [TYPE Class]
@@ -20,8 +20,8 @@ class GitServiceGithubMixin:
# @RETURN: dict
async def create_github_repository(
self, server_url: str, pat: str, name: str, private: bool = True,
description: Optional[str] = None, auto_init: bool = True, default_branch: Optional[str] = "main",
) -> Dict[str, Any]:
description: str | None = None, auto_init: bool = True, default_branch: str | None = "main",
) -> dict[str, Any]:
base_url = self._normalize_git_server_url(server_url)
if "github.com" in base_url:
api_url = "https://api.github.com/user/repos"
@@ -32,7 +32,7 @@ class GitServiceGithubMixin:
"Content-Type": "application/json",
"Accept": "application/vnd.github+json",
}
payload: Dict[str, Any] = {"name": name, "private": bool(private), "auto_init": bool(auto_init)}
payload: dict[str, Any] = {"name": name, "private": bool(private), "auto_init": bool(auto_init)}
if description:
payload["description"] = description
if default_branch:
@@ -41,7 +41,7 @@ class GitServiceGithubMixin:
async with httpx.AsyncClient(timeout=20.0) as client:
response = await client.post(api_url, headers=headers, json=payload)
except Exception as e:
raise HTTPException(status_code=503, detail=f"GitHub API is unavailable: {str(e)}")
raise HTTPException(status_code=503, detail=f"GitHub API is unavailable: {e!s}")
if response.status_code >= 400:
detail = response.text
try:
@@ -60,8 +60,8 @@ class GitServiceGithubMixin:
# @RETURN: Dict[str, Any]
async def create_github_pull_request(
self, server_url: str, pat: str, remote_url: str, from_branch: str, to_branch: str,
title: str, description: Optional[str] = None, draft: bool = False,
) -> Dict[str, Any]:
title: str, description: str | None = None, draft: bool = False,
) -> dict[str, Any]:
identity = self._parse_remote_repo_identity(remote_url)
base_url = self._normalize_git_server_url(server_url)
if "github.com" in base_url:
@@ -81,7 +81,7 @@ class GitServiceGithubMixin:
async with httpx.AsyncClient(timeout=20.0) as client:
response = await client.post(api_url, headers=headers, json=payload)
except Exception as e:
raise HTTPException(status_code=503, detail=f"GitHub API is unavailable: {str(e)}")
raise HTTPException(status_code=503, detail=f"GitHub API is unavailable: {e!s}")
if response.status_code >= 400:
detail = response.text
try:
@@ -104,12 +104,12 @@ class GitServiceGitlabMixin:
# @RETURN: dict
async def create_gitlab_repository(
self, server_url: str, pat: str, name: str, private: bool = True,
description: Optional[str] = None, auto_init: bool = True, default_branch: Optional[str] = "main",
) -> Dict[str, Any]:
description: str | None = None, auto_init: bool = True, default_branch: str | None = "main",
) -> dict[str, Any]:
base_url = self._normalize_git_server_url(server_url)
api_url = f"{base_url}/api/v4/projects"
headers = {"PRIVATE-TOKEN": pat.strip(), "Content-Type": "application/json", "Accept": "application/json"}
payload: Dict[str, Any] = {
payload: dict[str, Any] = {
"name": name, "visibility": "private" if private else "public",
"initialize_with_readme": bool(auto_init),
}
@@ -121,7 +121,7 @@ class GitServiceGitlabMixin:
async with httpx.AsyncClient(timeout=20.0) as client:
response = await client.post(api_url, headers=headers, json=payload)
except Exception as e:
raise HTTPException(status_code=503, detail=f"GitLab API is unavailable: {str(e)}")
raise HTTPException(status_code=503, detail=f"GitLab API is unavailable: {e!s}")
if response.status_code >= 400:
detail = response.text
try:
@@ -150,8 +150,8 @@ class GitServiceGitlabMixin:
# @RETURN: Dict[str, Any]
async def create_gitlab_merge_request(
self, server_url: str, pat: str, remote_url: str, from_branch: str, to_branch: str,
title: str, description: Optional[str] = None, remove_source_branch: bool = False,
) -> Dict[str, Any]:
title: str, description: str | None = None, remove_source_branch: bool = False,
) -> dict[str, Any]:
identity = self._parse_remote_repo_identity(remote_url)
base_url = self._normalize_git_server_url(server_url)
project_id = quote(identity["full_name"], safe="")
@@ -165,7 +165,7 @@ class GitServiceGitlabMixin:
async with httpx.AsyncClient(timeout=20.0) as client:
response = await client.post(api_url, headers=headers, json=payload)
except Exception as e:
raise HTTPException(status_code=503, detail=f"GitLab API is unavailable: {str(e)}")
raise HTTPException(status_code=503, detail=f"GitLab API is unavailable: {e!s}")
if response.status_code >= 400:
detail = response.text
try:

View File

@@ -3,10 +3,9 @@
# @BRIEF Status, diff, and commit history operations for GitService using git status --porcelain to avoid --cached flag issues.
# @RELATION USED_BY -> [GitService]
from typing import Any, Dict, List
from datetime import datetime
from git import Repo
from src.core.logger import logger, belief_scope
from src.core.logger import belief_scope, logger
# #region GitServiceStatusMixin [C:3] [TYPE Class]
@@ -135,7 +134,7 @@ class GitServiceStatusMixin:
# @PRE: Repository for dashboard_id exists.
# @POST: Returns a list of dictionaries for each commit in history.
# @RETURN: List[dict]
def get_commit_history(self, dashboard_id: int, limit: int = 50) -> List[dict]:
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)
commits = []

View File

@@ -5,12 +5,12 @@
# @RELATION DEPENDS_ON -> [GitServiceUrlMixin]
import os
from typing import Optional
from git import Repo
from git.exc import GitCommandError
from fastapi import HTTPException
from git.exc import GitCommandError
from src.core.database import SessionLocal
from src.core.logger import logger, belief_scope
from src.core.logger import belief_scope, logger
from src.models.git import GitRepository, GitServerConfig
@@ -107,7 +107,7 @@ class GitServiceSyncMixin:
raise HTTPException(status_code=500, detail=f"Git push failed: {details}")
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)}")
raise HTTPException(status_code=500, detail=f"Git push failed: {e!s}")
# endregion push_changes
# region pull_changes [TYPE Function]
@@ -167,7 +167,7 @@ class GitServiceSyncMixin:
raise
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)}")
raise HTTPException(status_code=500, detail=f"Git pull failed: {e!s}")
# endregion pull_changes
# #endregion GitServiceSyncMixin
# #endregion GitServiceSyncMixin

View File

@@ -5,13 +5,14 @@
# @RELATION USED_BY -> [GitServiceGiteaMixin]
# @RELATION USED_BY -> [GitServiceRemoteMixin]
import os
from typing import Any, Dict, List, Optional
from urllib.parse import quote, urlparse
from fastapi import HTTPException
from src.core.database import SessionLocal
from src.core.logger import logger, belief_scope
from src.models.git import GitRepository, GitServerConfig
from src.core.logger import logger
from src.models.git import GitRepository
# #region GitServiceUrlMixin [C:3] [TYPE Class]
# @BRIEF URL helper methods for GitService — extract host, strip credentials, replace host, align origin, parse remote identity, derive server URL, normalize server URL.
@@ -21,7 +22,7 @@ class GitServiceUrlMixin:
# @PRE: url_value may be empty.
# @POST: Returns lowercase host token or None.
# @RETURN: Optional[str]
def _extract_http_host(self, url_value: Optional[str]) -> Optional[str]:
def _extract_http_host(self, url_value: str | None) -> str | None:
normalized = str(url_value or "").strip()
if not normalized:
return None
@@ -65,7 +66,7 @@ class GitServiceUrlMixin:
# @PRE: source_url and config_url are HTTP(S) URLs.
# @POST: Returns source URL with updated host (credentials preserved) or None.
# @RETURN: Optional[str]
def _replace_host_in_url(self, source_url: Optional[str], config_url: Optional[str]) -> Optional[str]:
def _replace_host_in_url(self, source_url: str | None, config_url: str | None) -> str | None:
source = str(source_url or "").strip()
config = str(config_url or "").strip()
if not source or not config:
@@ -101,10 +102,10 @@ class GitServiceUrlMixin:
self,
dashboard_id: int,
origin,
config_url: Optional[str],
current_origin_url: Optional[str],
binding_remote_url: Optional[str],
) -> Optional[str]:
config_url: str | None,
current_origin_url: str | None,
binding_remote_url: str | None,
) -> str | None:
config_host = self._extract_http_host(config_url)
source_origin_url = str(current_origin_url or "").strip() or str(binding_remote_url or "").strip()
origin_host = self._extract_http_host(source_origin_url)
@@ -153,7 +154,7 @@ class GitServiceUrlMixin:
# @PRE: remote_url is a valid git URL.
# @POST: Returns owner/repo tokens.
# @RETURN: Dict[str, str]
def _parse_remote_repo_identity(self, remote_url: str) -> Dict[str, str]:
def _parse_remote_repo_identity(self, remote_url: str) -> dict[str, str]:
normalized = str(remote_url or "").strip()
if not normalized:
raise HTTPException(status_code=400, detail="Repository remote_url is empty")
@@ -179,7 +180,7 @@ class GitServiceUrlMixin:
# @PRE: remote_url may be any git URL.
# @POST: Returns normalized http(s) base URL or None when derivation is impossible.
# @RETURN: Optional[str]
def _derive_server_url_from_remote(self, remote_url: str) -> Optional[str]:
def _derive_server_url_from_remote(self, remote_url: str) -> str | None:
normalized = str(remote_url or "").strip()
if not normalized or normalized.startswith("git@"):
return None