Coverage for lynceus/devops/gitlab_devops_analyzer.py: 94%
237 statements
« prev ^ index » next coverage.py v7.10.0, created at 2025-07-29 08:46 +0000
« prev ^ index » next coverage.py v7.10.0, created at 2025-07-29 08:46 +0000
1from collections import defaultdict
2from datetime import date, datetime, timedelta, timezone
3from pathlib import Path
4from typing import Callable
6import gitlab
7from gitlab.const import AccessLevel
9from lynceus.core.config import DATETIME_FORMAT_SHORT
10from lynceus.core.exchange.lynceus_exchange import LynceusExchange
11from lynceus.core.lynceus import LynceusSession
12from lynceus.devops.devops_analyzer import DevOpsAnalyzer
13from lynceus.utils import flatten, parse_string_to_datetime
14from lynceus.utils.lynceus_dict import LynceusDict
17def gitlab_exception_handler(func):
18 """
19 Decorator to handle GitLab-specific exceptions and convert them to standard exceptions.
21 Parameters
22 ----------
23 func : callable
24 Function to wrap with exception handling
26 Returns
27 -------
28 callable
29 Wrapped function that handles GitLab exceptions
31 Raises
32 ------
33 PermissionError
34 For 401/403 HTTP status codes
35 NameError
36 For 404 HTTP status codes
37 """
39 def func_wrapper(*args, **kwargs):
40 """
41 Internal wrapper function that handles GitLab API errors.
43 Parameters
44 ----------
45 *args : tuple
46 Positional arguments passed to the wrapped function
47 **kwargs
48 Keyword arguments passed to the wrapped function
50 Returns
51 -------
52 object
53 Result of the wrapped function call
55 Raises
56 ------
57 PermissionError
58 For 401/403 HTTP status codes
59 NameError
60 For 404 HTTP status codes
61 """
62 try:
63 return func(*args, **kwargs)
64 except gitlab.exceptions.GitlabError as error:
65 # Intercepts permission error.
66 if error.response_code in (401, 403):
67 raise PermissionError(
68 "You don't have enough permission to perform this operation on Gitlab."
69 ) from error
70 if error.response_code == 404:
71 raise NameError("Unable to find requested Object.") from error
73 # Raises any other error.
74 raise
76 return func_wrapper
79def get_list_from_paginated_and_count(
80 plist_func: Callable, count: int | None = None, **kwargs
81) -> list:
82 """
83 Helper function to get a list from GitLab paginated results with optional count limit.
85 Parameters
86 ----------
87 plist_func : Callable
88 Function that returns paginated results
89 count : int, optional
90 Maximum number of items to retrieve
91 **kwargs
92 Additional arguments to pass to plist_func
94 Returns
95 -------
96 list
97 List of items from the paginated result
98 """
99 if count is not None and count:
100 kwargs = {"per_page": count, "page": 1} | kwargs
101 else:
102 kwargs = {"all": True} | kwargs
104 return list(plist_func(**kwargs))
107# See: https://python-gitlab.readthedocs.io/en/stable/api-usage.html
108# See: https://docs.gitlab.com/ee/api/README.html
109# See: https://docs.gitlab.com/ee/api/api_resources.html
110class GitlabDevOpsAnalyzer(DevOpsAnalyzer):
111 """
112 GitLab-specific implementation of the DevOps analyzer.
114 This class provides concrete implementations for all DevOps operations
115 specific to GitLab, including authentication, user/group/project management,
116 repository operations, and statistics gathering. It uses the python-gitlab
117 library to interact with GitLab's REST API.
119 Attributes:
120 IMPORT_EXPORT_STATUS_SUCCESS: Status indicating successful import/export
121 IMPORT_EXPORT_STATUS_FAILED: Status indicating failed import/export
122 IMPORT_EXPORT_STATUS_NONE: Status indicating no import/export operation
123 """
125 # See: https://docs.gitlab.com/ce/api/project_import_export.html#export-status
126 # See: https://docs.gitlab.com/ce/api/project_import_export.html#import-status
127 IMPORT_EXPORT_STATUS_SUCCESS: str = "finished"
128 IMPORT_EXPORT_STATUS_FAILED: str = "failed"
129 IMPORT_EXPORT_STATUS_NONE: str = "none"
131 def __init__(
132 self,
133 lynceus_session: LynceusSession,
134 uri: str,
135 token: str,
136 lynceus_exchange: LynceusExchange,
137 ):
138 """
139 Initialize the GitLab DevOps analyzer.
141 Parameters
142 ----------
143 lynceus_session : LynceusSession
144 The Lynceus session instance
145 uri : str
146 The GitLab instance URI
147 token : str
148 Personal access token for GitLab authentication
149 lynceus_exchange : LynceusExchange
150 Exchange instance for data communication
151 """
152 super().__init__(lynceus_session, uri, "gitlab", lynceus_exchange)
153 self._manager = gitlab.Gitlab(uri, private_token=token)
154 self.__current_token: dict | None = None
156 # The following methods are only for uniformity and coherence.
157 def _extract_user_info(self, user) -> LynceusDict:
158 # Extracts information which are always available.
159 user_info = {
160 "id": user.id,
161 "name": user.name,
162 "login": user.username,
163 "username": user.username,
164 }
166 # Extracts extra information if available.
167 for extra_info_key, extra_info_attr in (
168 ("e-mail", "public_email"),
169 ("avatar_url", "avatar_url"),
170 ("bio", "bio"),
171 ):
172 value = (
173 user.attributes[extra_info_attr]
174 if extra_info_attr in user.attributes
175 else self.INFO_UNDEFINED
176 )
177 user_info.update({extra_info_key: value})
179 return LynceusDict(user_info)
181 def _extract_group_info(self, group) -> LynceusDict:
182 return LynceusDict(
183 {"id": group.id, "name": group.name, "path": group.full_path}
184 )
186 def _extract_project_info(self, project) -> LynceusDict:
187 return LynceusDict(
188 {
189 "id": project.id,
190 "name": project.name,
191 "path": project.path_with_namespace,
192 "web_url": project.web_url,
193 }
194 )
196 def _extract_member_info(self, member) -> LynceusDict:
197 member_info = {
198 "id": member.id,
199 "name": member.name,
200 "login": member.username,
201 "username": member.username,
202 "state": member.state,
203 }
205 # Extracts extra information if available.
206 for extra_info_key, extra_info_attr in ("parent_id", "group_id"), (
207 "parent_id",
208 "project_id",
209 ):
210 value = (
211 member.attributes[extra_info_attr]
212 if extra_info_attr in member.attributes
213 else None
214 )
215 if value:
216 member_info.update({extra_info_key: value})
218 return LynceusDict(member_info)
220 def _extract_issue_event_info(self, issue_event, **kwargs) -> LynceusDict:
221 return LynceusDict(
222 {
223 "id": issue_event.id,
224 "issue_id": issue_event.target_iid,
225 "action": issue_event.action_name,
226 "target_type": issue_event.target_type,
227 "created_at": parse_string_to_datetime(
228 datetime_str=issue_event.created_at
229 ),
230 "author": issue_event.author["name"],
231 "title": issue_event.target_title,
232 "issue_web_url": kwargs["project_web_url"] + f"/-/issues/{issue_event.target_iid}",
233 # N.B.: project information is unable, and must be added by caller via kwargs.
234 }
235 | kwargs
236 )
238 def _extract_commit_info(self, commit) -> LynceusDict:
239 return LynceusDict(
240 {
241 "id": commit.id,
242 "short_id": commit.short_id,
243 "parent_ids": commit.parent_ids,
244 "message": commit.message,
245 "created_at": commit.created_at,
246 "author_name": commit.author_name,
247 "author_email": commit.author_email,
248 "committer_name": commit.committer_name,
249 "committer_email": commit.committer_email,
250 }
251 )
253 def _extract_branch_info(self, branch) -> LynceusDict:
254 return LynceusDict(
255 {
256 "name": branch.name,
257 "merged": branch.merged,
258 "commit_id": branch.commit["id"],
259 "commit_short_id": branch.commit["short_id"],
260 "created_at": parse_string_to_datetime(
261 datetime_str=branch.commit["created_at"]
262 ),
263 # Not available for other DevOps: 'project_id': branch.project_id,
264 }
265 )
267 def _extract_tag_info(self, tag) -> LynceusDict:
268 return LynceusDict(
269 {
270 "name": tag.name,
271 "commit_id": tag.commit["id"],
272 "commit_short_id": tag.commit["short_id"],
273 "created_at": parse_string_to_datetime(
274 datetime_str=tag.commit["created_at"]
275 ),
276 # Not available for other DevOps: 'project_id': branch.project_id,
277 }
278 )
280 # The following methods are only performing read access on DevOps backend.
281 @gitlab_exception_handler
282 def authenticate(self):
283 """
284 Authenticate with the GitLab instance using the configured credentials.
286 Perform authentication with the GitLab API using the access token
287 or other credentials provided during initialization.
288 """
289 self._manager.auth()
291 @gitlab_exception_handler
292 def _do_get_current_user(self):
293 return self._manager.user
295 @gitlab_exception_handler
296 def _do_get_user_without_cache(
297 self, *, username: str = None, email: str = None, **kwargs
298 ):
299 if "user_id" in kwargs:
300 return self._manager.users.get(id=kwargs["user_id"])
302 # Note: there is still no API route to search an user directly
303 # from username or email (but only from only known id)
304 # - https://python-gitlab.readthedocs.io/en/stable/gl_objects/users.html#users-examples
305 # - https://docs.gitlab.com/api/users/
306 users = self._manager.users.list(
307 iterator=True, username=username, email=email, **kwargs
308 )
309 try:
310 user = next(users)
311 self._logger.debug(
312 f'Successfully lookup user with parameters: "{username=}"/"{email=}".'
313 )
314 return user
315 except StopIteration:
316 # pylint: disable=raise-missing-from
317 raise NameError(
318 f'User "{username=}"/"{email=}" has not been found with specified parameters ({kwargs if kwargs else "none"}).'
319 )
321 @gitlab_exception_handler
322 def _do_get_groups(self, *, count: int | None = None, **kwargs):
323 return get_list_from_paginated_and_count(
324 self._manager.groups.list, count, **kwargs
325 )
327 @gitlab_exception_handler
328 def _do_get_group_without_cache(self, *, full_path: str, **kwargs):
329 groups = self._manager.groups.list(
330 iterator=True, search=full_path, search_namespaces=True, **kwargs
331 )
332 if not groups:
333 raise NameError(
334 f'Group "{full_path}" has not been found with specified parameters ({kwargs if kwargs else "none"}).'
335 )
337 # Checks if match must be exact (which is now the case by default ... check if has been defined to False manually).
338 if not bool(kwargs.get("exact_match", True)):
339 if len(groups) > 1:
340 self._logger.warning(
341 f'There are more than one group matching requested path "{full_path}"'
342 + f'(add "exact_match" parameter to ensure getting only one group): {groups} '
343 )
344 return next(groups)
346 try:
347 group = next(
348 filter(
349 lambda grp: str.lower(grp.full_path) == str.lower(full_path), groups
350 )
351 )
352 self._logger.debug(
353 f'Successfully lookup group with full path "{full_path}".'
354 )
355 return group
356 except StopIteration:
357 # pylint: disable=raise-missing-from
358 raise NameError(
359 f'Group "{full_path}" has not been found with specified parameters ({kwargs if kwargs else "none"}).'
360 + f" Maybe you are looking for one of the following paths: {[group.full_path for group in groups]}"
361 )
363 @gitlab_exception_handler
364 def _do_get_projects(self, *, count: int | None = None, **kwargs):
365 return get_list_from_paginated_and_count(
366 self._manager.projects.list, count, **kwargs
367 )
369 @gitlab_exception_handler
370 def _do_get_project_without_cache(self, *, full_path: str, **kwargs):
371 try:
372 project = self._manager.projects.get(
373 full_path, search_namespaces=True, **kwargs
374 )
375 self._logger.debug(
376 f'Successfully lookup project with full path "{full_path}".'
377 )
378 return project
379 except gitlab.exceptions.GitlabError as error:
380 if error.response_code != 404:
381 # Lets the @gitlab_exception_handler manages the exception.
382 raise
384 # pylint: disable=raise-missing-from
385 raise NameError(
386 f'Project "{full_path}" has not been found with specified parameters ({kwargs if kwargs else "none"}).'
387 )
389 @gitlab_exception_handler
390 def __get_current_token(self):
391 """
392 Get information about the current personal access token.
394 Returns
395 -------
396 dict
397 Token information including ID, name, scopes, and expiration
398 """
399 if not self.__current_token:
400 # Sample: {'id': 60, 'name': 'Lynceus CI/CD', 'revoked': False, 'created_at': '2020-10-22T15:50:13.516Z',
401 # 'scopes': ['api', 'admin_mode'], 'user_id': 5, 'last_used_at': '2023-04-12T09:22:19.495Z',
402 # 'active': True, 'expires_at': None}
403 self.__current_token = self._manager.personal_access_tokens.get("self")
405 return self.__current_token
407 @gitlab_exception_handler
408 # pylint: disable=too-many-return-statements
409 def check_permissions_on_project(
410 self,
411 *,
412 full_path: str,
413 get_metadata: bool,
414 pull: bool,
415 push: bool = False,
416 maintain: bool = False,
417 admin: bool = False,
418 **kwargs,
419 ):
420 """
421 Check user permissions on a specific GitLab project.
423 Verify that the authenticated user has the requested permissions
424 on the specified project by checking their access level and membership.
426 Parameters
427 ----------
428 full_path : str
429 Full path to the GitLab project (e.g., 'group/project')
430 get_metadata : bool
431 Whether metadata access is required
432 pull : bool
433 Whether pull/read access is required
434 push : bool, optional
435 Whether push/write access is required (default: False)
436 maintain : bool, optional
437 Whether maintainer access is required (default: False)
438 admin : bool, optional
439 Whether admin access is required (default: False)
440 **kwargs
441 Additional project lookup arguments
443 Returns
444 -------
445 bool
446 Permission check results with granted access levels
448 Raises
449 ------
450 PermissionError
451 If required permissions are not granted
452 """
453 try:
454 # First of all, checks get_metadata permission which is required anyway.
455 _ = self._do_get_project(full_path=full_path, **kwargs)
457 # From here, we consider get_metadata permission is OK.
459 # See: https://docs.gitlab.com/ee/user/permissions.html
460 # See: https://gitlab.com/gitlab-org/gitlab/-/blob/e97357824bedf007e75f8782259fe07435b64fbb/lib/gitlab/access.rb#L12-18
462 # Retrieves current use, and corresponding membership information if any, for specified project.
463 current_user = self._do_get_current_user()
464 members = self._do_get_project_members(full_path=full_path, recursive=True)
466 member = next(filter(lambda grp: grp.id == current_user.id, members))
467 access_level = member.access_level
469 # According to my tests, required role (aka access_level):
470 # - at least REPORTER role to get access to repository metadata (name, tags, branches ...)
471 # - at least DEVELOPER role to be able to pull the repository (the read_repository scope is NOT enough here)
472 # - at least MAINTAINER role to get access to project statistics
474 # Checks scopes of the token.
475 token = self.__get_current_token()
476 token_scopes = token.scopes
477 api_scope = "api" in token_scopes
478 admin_scope = "admin" in token_scopes
479 read_repository_scope = "read_repository" in token_scopes
480 write_repository_scope = "write_repository" in token_scopes
482 if pull and (
483 (not api_scope and not read_repository_scope)
484 or access_level < AccessLevel.DEVELOPER
485 ):
486 return False
488 if push and (
489 (not api_scope and not write_repository_scope)
490 or access_level < AccessLevel.DEVELOPER
491 ):
492 return False
494 if maintain and (
495 (not api_scope and not write_repository_scope)
496 or access_level < AccessLevel.MAINTAINER
497 ):
498 return False
500 if admin and (
501 (not api_scope and not admin_scope) or access_level < AccessLevel.OWNER
502 ):
503 return False
505 # All permission checks OK.
506 return True
507 except NameError:
508 # Returns True if there were NO permission at all to check ...
509 return (
510 not get_metadata
511 and not pull
512 and not push
513 and not maintain
514 and not admin
515 )
516 except StopIteration:
517 # Returns True if there were NO more permission to check ...
518 return not pull and not push and not maintain and not admin
520 @gitlab_exception_handler
521 def _do_get_project_commits(
522 self, *, full_path: str, git_ref_name: str, count: int | None = None, **kwargs
523 ):
524 project = self._do_get_project(full_path=full_path, **kwargs)
525 return get_list_from_paginated_and_count(
526 project.commits.list, count, ref_name=git_ref_name, **kwargs
527 )
529 @gitlab_exception_handler
530 def _do_get_project_branches(
531 self, *, full_path: str, count: int | None = None, **kwargs
532 ):
533 project = self._do_get_project(full_path=full_path, **kwargs)
534 return get_list_from_paginated_and_count(project.branches.list, count, **kwargs)
536 @gitlab_exception_handler
537 def _do_get_project_tags(
538 self, *, full_path: str, count: int | None = None, **kwargs
539 ):
540 project = self._do_get_project(full_path=full_path, **kwargs)
541 return get_list_from_paginated_and_count(project.tags.list, count, **kwargs)
543 @gitlab_exception_handler
544 def _do_get_project_members(
545 self, *, full_path: str, count: int | None = None, **kwargs
546 ):
547 project = self._do_get_project(full_path=full_path, **kwargs)
549 if not bool(kwargs.get("recursive", False)):
550 return get_list_from_paginated_and_count(
551 project.members.list, count, **kwargs
552 )
554 return get_list_from_paginated_and_count(
555 project.members_all.list, count, **kwargs
556 )
558 @gitlab_exception_handler
559 def _do_get_group_members(
560 self, *, full_path: str, count: int | None = None, **kwargs
561 ):
562 group = self._do_get_group(full_path=full_path, **kwargs)
564 if not bool(kwargs.get("recursive", False)):
565 return get_list_from_paginated_and_count(
566 group.members.list, count, **kwargs
567 )
569 return get_list_from_paginated_and_count(
570 group.members_all.list, count, **kwargs
571 )
573 @gitlab_exception_handler
574 def _do_get_project_issue_events(
575 self,
576 *,
577 full_path: str,
578 action: str | None = None,
579 from_date: datetime | None = None,
580 to_date: datetime | None = None,
581 count: int | None = None,
582 **kwargs,
583 ):
584 # See: https://python-gitlab.readthedocs.io/en/stable/api/gitlab.v4.html?highlight=events#gitlab.v4.objects.ProjectEventManager
585 # Object listing filters
586 # action => https://docs.gitlab.com/ee/user/profile/index.html#user-contribution-events
587 # target_type => https://docs.gitlab.com/ee/api/events.html#target-types
588 # sort
589 project = self._do_get_project(full_path=full_path, **kwargs)
591 # Adds filters if needed.
592 if action:
593 kwargs["action"] = action
595 # before & after => https://docs.gitlab.com/ee/api/events.html#date-formatting
596 if from_date:
597 kwargs["after"] = from_date.strftime("%Y-%m-%d")
599 if to_date:
600 kwargs["before"] = to_date.strftime("%Y-%m-%d")
602 return get_list_from_paginated_and_count(
603 project.events.list, count, target_type="issue", **kwargs
604 )
606 @gitlab_exception_handler
607 def _do_get_project_issues(
608 self, *, full_path: str, count: int | None = None, **kwargs
609 ):
610 project = self._do_get_project(full_path=full_path, **kwargs)
611 return get_list_from_paginated_and_count(project.issues.list, count, **kwargs)
613 @gitlab_exception_handler
614 def _do_get_project_merge_requests(
615 self, *, full_path: str, count: int | None = None, **kwargs
616 ):
617 project = self._do_get_project(full_path=full_path, **kwargs)
618 return get_list_from_paginated_and_count(
619 project.mergerequests.list, count, **kwargs
620 )
622 @gitlab_exception_handler
623 def _do_get_project_milestones(
624 self, *, full_path: str, count: int | None = None, **kwargs
625 ):
626 project = self._do_get_project(full_path=full_path, **kwargs)
627 return get_list_from_paginated_and_count(
628 project.milestones.list, count, **kwargs
629 )
631 @gitlab_exception_handler
632 def __get_recursive_groups(
633 self, *, full_path: str, count: int | None = None, **kwargs
634 ):
635 """
636 Recursively get all subgroups for a given group.
638 Parameters
639 ----------
640 full_path : str
641 Full path of the parent group
642 count : int, optional
643 Maximum number of groups to retrieve
644 **kwargs
645 Additional filtering parameters
647 Returns
648 -------
649 list
650 List of all groups including the parent and all recursive subgroups
651 """
652 # TODO: find a way to implement count properly.
653 group = self._do_get_group(full_path=full_path, **kwargs)
654 subgroups = get_list_from_paginated_and_count(
655 group.subgroups.list, count, **kwargs
656 )
657 if not subgroups:
658 return [group]
659 return flatten(
660 self.__get_recursive_groups(full_path=subgroup.full_path, **kwargs)
661 for subgroup in subgroups
662 )
664 @gitlab_exception_handler
665 def _do_get_group_projects(
666 self, *, full_path: str, count: int | None = None, **kwargs
667 ):
668 group = self._do_get_group(full_path=full_path, **kwargs)
670 # Checks if recursive is requested.
671 if not bool(kwargs.get("recursive", False)):
672 project_list = get_list_from_paginated_and_count(
673 group.projects.list, count, **kwargs
674 )
675 else:
676 # TODO: find a way to implement count
677 all_groups = {group} | set(
678 self.__get_recursive_groups(full_path=full_path, **kwargs)
679 )
680 project_list = flatten(
681 group.projects.list(all=True) for group in all_groups
682 )
684 return list(
685 map(
686 lambda gp: self._do_get_project(full_path=gp.path_with_namespace),
687 project_list,
688 )
689 )
691 @gitlab_exception_handler
692 def _do_get_group_milestones(
693 self, *, full_path: str, count: int | None = None, **kwargs
694 ):
695 group = self._do_get_group(full_path=full_path, **kwargs)
696 return get_list_from_paginated_and_count(group.milestones.list, count, **kwargs)
698 @gitlab_exception_handler
699 def get_user_stats_commit_activity(
700 self,
701 *,
702 group_full_path: str = None,
703 project_full_path: str = None,
704 since: datetime = None,
705 keep_empty_stats: bool = False,
706 count: int | None = None,
707 ):
708 """
709 Get commit activity statistics for the authenticated user in GitLab.
711 Parameters
712 ----------
713 group_full_path : str, optional
714 Group path to limit statistics to
715 project_full_path : str, optional
716 Project path to include in statistics
717 since : datetime, optional
718 Start date for statistics (defaults to 365 days ago)
719 keep_empty_stats : bool, optional
720 Whether to include days with zero commits (default: False)
721 count : int, optional
722 Maximum number of projects to analyze
724 Returns
725 -------
726 dict
727 Mapping of dates to commit counts
728 """
729 # TODO: find a way to implement count
730 # See: https://docs.gitlab.com/ee/api/project_statistics.html#get-the-statistics-of-the-last-30-days
732 # Defines projects on which to perform statistics.
733 if group_full_path is None:
734 projects = self._do_get_projects(count=count)
735 else:
736 projects = self._do_get_group_projects(
737 full_path=group_full_path, count=count
738 )
739 if project_full_path is not None:
740 projects.append(self._do_get_project(full_path=project_full_path))
742 # Defines threshold date.
743 contributions_since: datetime = (
744 since if since else datetime.now(tz=timezone.utc) - timedelta(days=365)
745 )
746 stats_user_commit_activity: dict[date, int] = defaultdict(int)
747 for _project in projects:
748 for commit_activity in _project.additionalstatistics.get().fetches["days"]:
749 day_date = parse_string_to_datetime(
750 datetime_str=commit_activity["date"],
751 datetime_format=DATETIME_FORMAT_SHORT,
752 )
754 # Ignores oldest statistics.
755 if day_date < contributions_since:
756 continue
758 # Ignores 0 stats but if wanted.
759 if not keep_empty_stats and not commit_activity["count"]:
760 continue
762 stats_user_commit_activity[day_date.date()] += commit_activity["count"]
764 return stats_user_commit_activity
766 @gitlab_exception_handler
767 # pylint: disable=unused-argument
768 def get_user_contributions(
769 self,
770 *,
771 since: datetime = None,
772 keep_empty_stats: bool = False,
773 count: int | None = None,
774 ):
775 """
776 Get user contribution statistics (not implemented for GitLab).
778 This method is not implemented for GitLab as the equivalent functionality
779 is not readily available through the GitLab API.
781 Parameters
782 ----------
783 since : datetime, optional
784 Start date for contributions (unused)
785 keep_empty_stats : bool, optional
786 Whether to include empty statistics (unused) (default: False)
787 count : int, optional
788 Maximum number of projects to analyze (unused)
790 Returns
791 -------
792 dict
793 Empty dictionary
794 """
795 self._logger.warning(
796 "get_user_contributions is not implemented yet for GitLab DevOps Analyzer"
797 )
798 return {}
800 @gitlab_exception_handler
801 def get_user_stats_code_frequency(self, *, count: int | None = None):
802 """
803 Get code frequency statistics (not implemented for GitLab).
805 This method is not implemented for GitLab as the equivalent functionality
806 is not readily available through the GitLab API.
808 Parameters
809 ----------
810 count : int, optional
811 Maximum number of projects to analyze (unused)
813 Returns
814 -------
815 dict
816 Empty dictionary
817 """
818 self._logger.warning(
819 "get_user_stats_commit_activity is not implemented yet for GitLab DevOps Analyzer"
820 )
821 return {}
823 @gitlab_exception_handler
824 def _do_download_repository(
825 self,
826 *,
827 project_full_path: str,
828 dest_path: Path,
829 reference: str = None,
830 chunk_size: int = 1024,
831 **kwargs,
832 ):
833 # See: https://docs.gitlab.com/ee/api/repositories.html#get-file-archive
834 # See: https://github.com/python-gitlab/python-gitlab/blob/main/gitlab/v4/objects.py#L4465
835 project = self._do_get_project(full_path=project_full_path, **kwargs)
836 with open(dest_path, "wb") as dest_file:
837 project.repository_archive(
838 sha=reference,
839 streamed=True,
840 action=dest_file.write,
841 chunk_size=chunk_size,
842 **kwargs,
843 )