105 lines
3.5 KiB
Python
105 lines
3.5 KiB
Python
import hashlib
|
|
import logging
|
|
import subprocess
|
|
from pathlib import Path
|
|
from urllib.parse import quote, urlsplit, urlunsplit
|
|
|
|
from app.repository.security import validate_repository_url
|
|
from app.repository.tools import RepositoryTools
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class RepositoryService:
|
|
def __init__(self, url: str, branch: str, path: Path, token: str | None, max_file_bytes: int):
|
|
self.url, self.branch, self.path, self.token, self.max_file_bytes = (
|
|
url,
|
|
branch,
|
|
path,
|
|
token,
|
|
max_file_bytes,
|
|
)
|
|
|
|
def _authenticated_url(self) -> str:
|
|
if not self.token:
|
|
return self.url
|
|
parsed = urlsplit(self.url)
|
|
if parsed.scheme != "https":
|
|
raise ValueError("Private repository URL must use HTTPS")
|
|
return urlunsplit(
|
|
(
|
|
parsed.scheme,
|
|
f"oauth2:{quote(self.token, safe='')}@{parsed.netloc}",
|
|
parsed.path,
|
|
parsed.query,
|
|
"",
|
|
)
|
|
)
|
|
|
|
def sync(self) -> str:
|
|
if not self.url:
|
|
raise RuntimeError("REPOSITORY_URL is not configured")
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
if not (self.path / ".git").exists():
|
|
clone_args = [
|
|
"git",
|
|
"clone",
|
|
"--filter=blob:none",
|
|
"--no-checkout",
|
|
"--depth",
|
|
"50",
|
|
]
|
|
if self.branch != "HEAD":
|
|
clone_args.extend(["--branch", self.branch])
|
|
clone_args.extend([self._authenticated_url(), str(self.path)])
|
|
subprocess.run(
|
|
clone_args,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
else:
|
|
fetch_args = ["git", "fetch", "origin", "--depth", "50"]
|
|
if self.branch != "HEAD":
|
|
fetch_args.append(self.branch)
|
|
subprocess.run(
|
|
fetch_args,
|
|
cwd=self.path,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
target = "origin/HEAD" if self.branch == "HEAD" else f"origin/{self.branch}"
|
|
subprocess.run(
|
|
["git", "reset", "--soft", target],
|
|
cwd=self.path,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
commit = RepositoryTools(self.path, self.max_file_bytes).current_commit()
|
|
logger.info("repository_synced branch=%s commit=%s", self.branch, commit)
|
|
return commit
|
|
|
|
def tools(self) -> RepositoryTools:
|
|
if not (self.path / ".git").exists():
|
|
raise RuntimeError("Repository is not synced")
|
|
return RepositoryTools(self.path, self.max_file_bytes)
|
|
|
|
|
|
class RepositoryManager:
|
|
def __init__(self, cache_root: Path, token: str | None, max_file_bytes: int):
|
|
self.cache_root = cache_root.resolve()
|
|
self.token, self.max_file_bytes = token, max_file_bytes
|
|
|
|
def service_for(
|
|
self, url: str, branch: str = "HEAD", cache_path: str | None = None
|
|
) -> RepositoryService:
|
|
normalized = validate_repository_url(url)
|
|
cache = (
|
|
Path(cache_path)
|
|
if cache_path
|
|
else self.cache_root / hashlib.sha256(normalized.encode()).hexdigest()
|
|
)
|
|
return RepositoryService(normalized, branch, cache, self.token, self.max_file_bytes)
|