Coverage for app/services/git_service.py: 32.69%
208 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-07-25 15:51 +0000
1"""Git service for repository management and release information retrieval."""
3import logging
4import re
5import shutil
6from pathlib import Path
7from typing import Any
8from urllib.parse import quote
10import git
11import requests
12from requests.adapters import HTTPAdapter
13from urllib3.util.retry import Retry
15from ..config import settings
17logger = logging.getLogger(__name__)
20class GitService:
21 """Service for Git operations and release management."""
23 # Sparse-checkout allowlist for the variable scan (non-cone mode,
24 # ``.gitignore`` pathspecs). ``packer/`` matches the whole subtree so
25 # both the flat layout (``packer/variables.pkr.hcl``) and per-template
26 # layout (``packer/<key>/variables.pkr.hcl``) plus the
27 # ``template.pkr.hcl`` files are fetched. ``terraform/variables.tf``
28 # is a single file since only the variables file is needed.
29 SPARSE_CHECKOUT_FILES = [
30 'terraform/variables.tf',
31 'packer/',
32 ]
34 def __init__(self) -> None:
35 """Initialize Git service with settings."""
36 self.base_path = Path(settings.TEMP_REPO_BASE_PATH)
37 self.token = settings.GIT_ACCESS_TOKEN
38 self._session = self._create_http_session()
40 def _create_http_session(self) -> requests.Session:
41 """Create requests session with retry strategy."""
42 session = requests.Session()
43 retry = Retry(
44 total=3,
45 backoff_factor=1,
46 status_forcelist=[429, 500, 502, 503, 504],
47 )
48 adapter = HTTPAdapter(max_retries=retry)
49 session.mount("http://", adapter)
50 session.mount("https://", adapter)
51 return session
53 def _get_authenticated_url(self, git_url: str) -> str:
54 """Convert Git URL to HTTPS format with token authentication."""
55 url = git_url
56 if url.startswith('git@'):
57 url = url.replace('git@', '').replace(':', '/', 1)
58 elif url.startswith(('https://', 'http://')):
59 url = re.sub(r'^https?://', '', url)
61 if '@' in url:
62 url = url.split('@', 1)[1]
64 if self.token:
65 return f"https://{self.token}@{url}"
66 return f"https://{url}"
68 def _parse_git_url(self, git_url: str) -> dict[str, str] | None:
69 """Parse Git URL and extract components."""
70 patterns = [
71 r'^git@([^:]+):([^/]+)/(.+?)(?:\.git)?$',
72 r'^https?://([^/]+)/([^/]+)/(.+?)(?:\.git)?$',
73 ]
75 for pattern in patterns:
76 match = re.match(pattern, git_url)
77 if match:
78 host, owner, repo = match.groups()
79 platform = 'github' if 'github' in host.lower() else 'gitlab' if 'gitlab' in host.lower() else 'unknown'
80 return {'host': host, 'owner': owner, 'repo': repo, 'platform': platform}
82 logger.warning(f"Could not parse git URL: {git_url}")
83 return None
85 def _request_tags(self, parsed: dict[str, str]) -> requests.Response:
86 """Build and perform the tags request for the parsed repository.
88 Constructs the provider-specific tags URL and headers (GitHub vs.
89 GitLab, including auth token handling) and issues the GET request,
90 returning the raw response for the caller to interpret.
91 """
92 if parsed['platform'] == 'github':
93 api_url = f"https://api.github.com/repos/{parsed['owner']}/{parsed['repo']}/tags"
94 headers = {
95 'Authorization': f"token {self.token}",
96 'Accept': 'application/vnd.github.v3+json',
97 }
98 else: # gitlab
99 project_path = quote(f"{parsed['owner']}/{parsed['repo']}", safe='')
100 api_url = f"https://{parsed['host']}/api/v4/projects/{project_path}/repository/tags"
101 headers = {'PRIVATE-TOKEN': self.token}
103 return self._session.get(api_url, headers=headers, timeout=10)
105 def _fetch_github_tags(self, parsed: dict[str, str]) -> list[dict[str, Any]]:
106 """Fetch all tags from GitHub API."""
107 response = self._request_tags(parsed)
109 if response.status_code == 404:
110 return []
112 response.raise_for_status()
113 return [{'version': tag['name'], 'commit': tag['commit']['sha'][:8], 'type': 'tag'} for tag in response.json()]
115 def _fetch_github_releases(self, parsed: dict[str, str]) -> dict[str, dict[str, Any]]:
116 """Fetch releases from GitHub API."""
117 api_url = f"https://api.github.com/repos/{parsed['owner']}/{parsed['repo']}/releases"
118 response = self._session.get(
119 api_url,
120 headers={'Authorization': f"token {self.token}", 'Accept': 'application/vnd.github.v3+json'},
121 timeout=10
122 )
124 if response.status_code == 404:
125 return {}
127 response.raise_for_status()
128 return {
129 release['tag_name']: {
130 'name': release.get('name') or release['tag_name'],
131 'description': release.get('body', ''),
132 'author': release.get('author', {}).get('login', ''),
133 'published_at': release.get('published_at', ''),
134 'prerelease': str(release.get('prerelease', False)),
135 'html_url': release.get('html_url', ''),
136 }
137 for release in response.json() if release.get('tag_name')
138 }
140 def _fetch_gitlab_tags(self, parsed: dict[str, str]) -> list[dict[str, Any]]:
141 """Fetch all tags from GitLab API."""
142 response = self._request_tags(parsed)
144 if response.status_code == 404:
145 return []
147 response.raise_for_status()
148 return [{'version': tag['name'], 'commit': tag['commit']['id'][:8], 'type': 'tag'} for tag in response.json()]
150 def _fetch_gitlab_releases(self, parsed: dict[str, str]) -> dict[str, dict[str, Any]]:
151 """Fetch releases from GitLab API."""
152 project_path = quote(f"{parsed['owner']}/{parsed['repo']}", safe='')
153 api_url = f"https://{parsed['host']}/api/v4/projects/{project_path}/releases"
154 response = self._session.get(
155 api_url,
156 headers={'PRIVATE-TOKEN': self.token},
157 timeout=10
158 )
160 if response.status_code == 404:
161 return {}
163 response.raise_for_status()
164 return {
165 release['tag_name']: {
166 'name': release.get('name') or release['tag_name'],
167 'description': release.get('description', ''),
168 'author': release.get('author', {}).get('username', ''),
169 'published_at': release.get('released_at', ''),
170 'prerelease': 'False',
171 'html_url': release.get('_links', {}).get('self', ''),
172 }
173 for release in response.json() if release.get('tag_name')
174 }
176 def _accept_github_invite(self, parsed: dict[str, str]) -> dict[str, Any]:
177 """Check for and accept pending GitHub repository invitations."""
178 try:
179 repo_full_name = f"{parsed['owner']}/{parsed['repo']}"
181 # Get all pending invitations
182 api_url = "https://api.github.com/user/repository_invitations"
183 response = self._session.get(
184 api_url,
185 headers={
186 'Authorization': f'token {self.token}',
187 'Accept': 'application/vnd.github.v3+json'
188 },
189 timeout=10
190 )
191 response.raise_for_status()
192 invitations = response.json()
194 # Find invitation for this repository
195 invitation = next(
196 (inv for inv in invitations if inv.get('repository', {}).get('full_name') == repo_full_name),
197 None
198 )
200 if not invitation:
201 return {
202 'success': False,
203 'message': f"No pending invitation found for repository {repo_full_name}. Please ask the repository owner to invite the app store user."
204 }
206 # Accept the invitation
207 invitation_id = invitation['id']
208 accept_url = f"https://api.github.com/user/repository_invitations/{invitation_id}"
209 accept_response = self._session.patch(
210 accept_url,
211 headers={
212 'Authorization': f'token {self.token}',
213 'Accept': 'application/vnd.github.v3+json'
214 },
215 timeout=10
216 )
218 if accept_response.status_code == 204:
219 logger.info(f"Successfully accepted GitHub invitation for {repo_full_name}")
220 return {
221 'success': True,
222 'message': f"Invitation accepted successfully for {repo_full_name}"
223 }
224 else:
225 return {
226 'success': False,
227 'message': f"Failed to accept invitation (HTTP {accept_response.status_code})"
228 }
230 except Exception as e:
231 logger.error(f"Failed to accept GitHub invitation: {str(e)}", exc_info=True)
232 return {
233 'success': False,
234 'message': f"Error processing invitation: {str(e)}"
235 }
237 def _accept_gitlab_invite(self, parsed: dict[str, str]) -> dict[str, Any]:
238 """Check for and accept pending GitLab project invitations."""
239 try:
240 project_path = quote(f"{parsed['owner']}/{parsed['repo']}", safe='')
242 # Try to get project ID first
243 project_url = f"https://{parsed['host']}/api/v4/projects/{project_path}"
244 project_response = self._session.get(
245 project_url,
246 headers={'PRIVATE-TOKEN': self.token},
247 timeout=10
248 )
250 if project_response.status_code == 404:
251 return {
252 'success': False,
253 'message': f"Repository not found or no pending invitation exists. Please ask the repository owner to invite the app store user to: {parsed['owner']}/{parsed['repo']}"
254 }
256 # GitLab automatically grants access when invited, so if we can access the project, we're good
257 if project_response.status_code == 200:
258 logger.info(f"GitLab access already granted for {parsed['owner']}/{parsed['repo']}")
259 return {
260 'success': True,
261 'message': f"Access granted for {parsed['owner']}/{parsed['repo']}"
262 }
264 return {
265 'success': False,
266 'message': "No pending invitation found. Please ask the repository owner to add the app store user as a member."
267 }
269 except Exception as e:
270 logger.error(f"Failed to check GitLab invitation: {str(e)}", exc_info=True)
271 return {
272 'success': False,
273 'message': f"Error processing invitation: {str(e)}"
274 }
276 def verify_repository_access(self, git_url: str) -> dict[str, Any]:
277 """
278 Verify that the app store user has read access to the repository.
279 Returns dict with 'success' bool and 'message' str.
280 """
281 try:
282 parsed = self._parse_git_url(git_url)
284 if not parsed or parsed['platform'] == 'unknown':
285 return {
286 'success': False,
287 'message': f"Unable to parse repository URL or unsupported platform. Supported: GitHub, GitLab. URL: {git_url}"
288 }
290 if not self.token:
291 return {
292 'success': False,
293 'message': "Git access token is not configured. Please contact the administrator."
294 }
296 # Try to fetch tags to verify access
297 logger.info(f"Verifying access to {git_url} via {parsed['platform'].upper()} API")
299 response = self._request_tags(parsed)
301 if response.status_code == 401:
302 return {
303 'success': False,
304 'message': "Authentication failed. The Git access token is invalid or expired."
305 }
306 elif response.status_code == 403 or response.status_code == 404:
307 # Try to accept invitation automatically
308 logger.info(f"Access denied, attempting to accept invitation for {git_url}")
310 if parsed['platform'] == 'github':
311 invite_result = self._accept_github_invite(parsed)
312 else: # gitlab
313 invite_result = self._accept_gitlab_invite(parsed)
315 if not invite_result['success']:
316 return invite_result
318 # Retry access check after accepting invitation
319 logger.info(f"Invitation accepted, retrying access check for {git_url}")
320 retry_response = self._request_tags(parsed)
322 if retry_response.status_code >= 400:
323 return {
324 'success': False,
325 'message': "Invitation was accepted but access still denied. Please wait a moment and try again."
326 }
328 logger.info(f"Access verified successfully after accepting invitation for {git_url}")
329 return {
330 'success': True,
331 'message': "Repository invitation accepted and access verified successfully"
332 }
334 elif response.status_code >= 400:
335 return {
336 'success': False,
337 'message': f"Failed to access repository (HTTP {response.status_code}). Please check the repository URL and permissions."
338 }
340 response.raise_for_status()
341 logger.info(f"Access verified successfully for {git_url}")
342 return {
343 'success': True,
344 'message': "Repository access verified successfully"
345 }
347 except requests.exceptions.Timeout:
348 return {
349 'success': False,
350 'message': "Request timeout. The Git provider did not respond in time. Please try again."
351 }
352 except requests.exceptions.RequestException as e:
353 logger.error(f"Failed to verify repository access: {str(e)}", exc_info=True)
354 return {
355 'success': False,
356 'message': f"Failed to connect to Git provider: {str(e)}"
357 }
358 except Exception as e:
359 logger.error(f"Unexpected error during repository access verification: {str(e)}", exc_info=True)
360 return {
361 'success': False,
362 'message': f"Unexpected error: {str(e)}"
363 }
365 def clone_release_vars(self, git_url: str, tag: str, deployment_id: str) -> str:
366 """Clone specific files from a release using sparse checkout."""
367 repo_path = self.base_path / f"deploy_{deployment_id}"
369 if repo_path.exists():
370 shutil.rmtree(repo_path)
372 try:
373 auth_url = self._get_authenticated_url(git_url)
374 logger.info(f"Cloning {git_url} tag {tag} (sparse checkout)")
376 repo = git.Repo.init(repo_path)
377 origin = repo.create_remote('origin', auth_url)
379 # Configure sparse checkout
380 git_dir = repo_path / '.git'
381 sparse_file = git_dir / 'info' / 'sparse-checkout'
382 sparse_file.parent.mkdir(parents=True, exist_ok=True)
383 sparse_file.write_text('\n'.join(self.SPARSE_CHECKOUT_FILES) + '\n')
385 with (git_dir / 'config').open('a') as f:
386 f.write('[core]\n\tsparseCheckout = true\n')
388 origin.fetch(refspec=f'refs/tags/{tag}:refs/tags/{tag}', depth=1)
389 repo.git.checkout(f'refs/tags/{tag}', force=True)
391 logger.info(f"Cloned successfully: {repo.head.commit.hexsha[:8]}")
392 return str(repo_path)
394 except Exception as e:
395 if repo_path.exists():
396 shutil.rmtree(repo_path)
397 raise Exception(f"Failed to clone: {str(e)}") from e
399 def get_versions(self, git_url: str) -> list[dict[str, Any]]:
400 """Get all versions via REST API."""
401 parsed = self._parse_git_url(git_url)
403 if not parsed or parsed['platform'] == 'unknown':
404 raise Exception(f"Unable to parse URL or unsupported platform: {git_url}")
406 if not self.token:
407 raise Exception("Git access token not configured")
409 try:
410 logger.info(f"Fetching versions from {parsed['platform'].upper()} API")
412 if parsed['platform'] == 'github':
413 versions = self._fetch_github_tags(parsed)
414 releases = self._fetch_github_releases(parsed)
415 else: # gitlab
416 versions = self._fetch_gitlab_tags(parsed)
417 releases = self._fetch_gitlab_releases(parsed)
419 # Merge release info into versions
420 for v in versions:
421 if v['version'] in releases:
422 v.update(releases[v['version']])
424 # Sort by semantic versioning
425 def sort_key(v):
426 try:
427 return tuple(map(int, v['version'].lstrip('v').split('.')))
428 except (ValueError, AttributeError):
429 return (v['version'],)
431 versions.sort(key=sort_key, reverse=True)
432 logger.info(f"Found {len(versions)} versions")
433 return versions
435 except Exception as e:
436 logger.error(f"Failed to fetch versions: {str(e)}", exc_info=True)
437 raise Exception(f"Failed to fetch versions: {str(e)}") from e
439 def cleanup_repository(self, repo_path: str) -> None:
440 """Clean up cloned repository."""
441 path = Path(repo_path)
442 if path.exists():
443 logger.info(f"Cleaning up {repo_path}")
444 shutil.rmtree(path)
447# Singleton instance
448git_service = GitService()