Coverage for lynceus/devops/devops_analyzer.py: 98%
172 statements
« prev ^ index » next coverage.py v7.10.0, created at 2025-07-29 08:46 +0000
1import tarfile
2from datetime import datetime
3from pathlib import Path
5from lynceus.core.exchange.lynceus_exchange import LynceusExchange
6from lynceus.core.lynceus import LynceusSession
7from lynceus.core.lynceus_client import LynceusClientClass
8from lynceus.lynceus_exceptions import LynceusError
9from lynceus.utils.lynceus_dict import LynceusDict
12# TODO: add accessrequests/permissions listing feature
15# pylint: disable=too-many-public-methods
16class DevOpsAnalyzer(LynceusClientClass):
17 """
18 Abstract base class for DevOps platform analyzers (GitLab, GitHub).
20 This class provides a unified interface for interacting with different DevOps platforms,
21 enabling read-only operations such as retrieving users, projects, groups, commits, and
22 various statistics. It includes caching mechanisms for improved performance and defines
23 abstract methods that must be implemented by platform-specific subclasses.
25 The class supports:
26 - User, group, and project management
27 - Repository operations (branches, tags, commits)
28 - Issue and merge request tracking
29 - Statistics and analytics
30 - Repository downloads
31 - Permission checks
33 Attributes:
34 STATUS_ACTIVE: Constant representing active status
35 INFO_UNDEFINED: Constant representing undefined information
36 CACHE_USER_TYPE: Cache type identifier for users
37 CACHE_GROUP_TYPE: Cache type identifier for groups
38 CACHE_PROJECT_TYPE: Cache type identifier for projects
39 """
41 STATUS_ACTIVE: str = "active"
42 INFO_UNDEFINED: str = "undefined"
44 CACHE_USER_TYPE: str = "user"
45 CACHE_GROUP_TYPE: str = "group"
46 CACHE_PROJECT_TYPE: str = "project"
48 def __init__(
49 self,
50 lynceus_session: LynceusSession,
51 uri: str,
52 logger_name: str,
53 lynceus_exchange: LynceusExchange | None,
54 ):
55 """
56 Initialize the DevOps analyzer.
58 Parameters
59 ----------
60 lynceus_session : LynceusSession
61 The Lynceus session instance
62 uri : str
63 The URI of the DevOps platform
64 logger_name : str
65 Name for the logger instance
66 lynceus_exchange : LynceusExchange
67 Exchange instance for data communication
68 """
69 super().__init__(lynceus_session, logger_name, lynceus_exchange)
70 self._uri: str = uri
72 # Initializes cache system.
73 self.__cache: dict[str, dict[tuple, object]] = {
74 self.CACHE_USER_TYPE: {},
75 self.CACHE_GROUP_TYPE: {},
76 self.CACHE_PROJECT_TYPE: {},
77 }
79 # The following methods are only for uniformity and coherence.
80 def _extract_user_info(self, user) -> LynceusDict:
81 """
82 Extract standardized user information from platform-specific user object.
84 Parameters
85 ----------
86 user : object
87 Platform-specific user object
89 Returns
90 -------
91 LynceusDict
92 Standardized user information dictionary
94 Raises
95 ------
96 NotImplementedError
97 Must be implemented by subclasses
98 """
99 raise NotImplementedError()
101 def _extract_group_info(self, group) -> LynceusDict:
102 """
103 Extract standardized group information from platform-specific group object.
105 Parameters
106 ----------
107 group : object
108 Platform-specific group object
110 Returns
111 -------
112 LynceusDict
113 Standardized group information dictionary
115 Raises
116 ------
117 NotImplementedError
118 Must be implemented by subclasses
119 """
120 raise NotImplementedError()
122 def _extract_project_info(self, project) -> LynceusDict:
123 """
124 Extract standardized project information from platform-specific project object.
126 Parameters
127 ----------
128 project : object
129 Platform-specific project object
131 Returns
132 -------
133 LynceusDict
134 Standardized project information dictionary
136 Raises
137 ------
138 NotImplementedError
139 Must be implemented by subclasses
140 """
141 raise NotImplementedError()
143 def _extract_member_info(self, member) -> LynceusDict:
144 """
145 Extract standardized member information from platform-specific member object.
147 Parameters
148 ----------
149 member : object
150 Platform-specific member object
152 Returns
153 -------
154 LynceusDict
155 Standardized member information dictionary
157 Raises
158 ------
159 NotImplementedError
160 Must be implemented by subclasses
161 """
162 raise NotImplementedError()
164 def _extract_issue_event_info(self, issue_event, **kwargs) -> LynceusDict:
165 """
166 Extract standardized issue event information from platform-specific issue event object.
168 Parameters
169 ----------
170 issue_event : object
171 Platform-specific issue event object
172 **kwargs
173 Additional context information (e.g., project metadata)
175 Returns
176 -------
177 LynceusDict
178 Standardized issue event information dictionary
180 Raises
181 ------
182 NotImplementedError
183 Must be implemented by subclasses
184 """
185 raise NotImplementedError()
187 def _extract_commit_info(self, commit) -> LynceusDict:
188 """
189 Extract standardized commit information from platform-specific commit object.
191 Parameters
192 ----------
193 commit : object
194 Platform-specific commit object
196 Returns
197 -------
198 LynceusDict
199 Standardized commit information dictionary
201 Raises
202 ------
203 NotImplementedError
204 Must be implemented by subclasses
205 """
206 raise NotImplementedError()
208 def _extract_branch_info(self, branch) -> str:
209 """
210 Extract standardized branch information from platform-specific branch object.
212 Parameters
213 ----------
214 branch : object
215 Platform-specific branch object
217 Returns
218 -------
219 str
220 Standardized branch information
222 Raises
223 ------
224 NotImplementedError
225 Must be implemented by subclasses
226 """
227 raise NotImplementedError()
229 def _extract_tag_info(self, tag) -> str:
230 """
231 Extract standardized tag information from platform-specific tag object.
233 Parameters
234 ----------
235 tag : object
236 Platform-specific tag object
238 Returns
239 -------
240 str
241 Standardized tag information
243 Raises
244 ------
245 NotImplementedError
246 Must be implemented by subclasses
247 """
248 raise NotImplementedError()
250 def __get_from_cache(self, *, cache_type: str, cache_key: tuple, log_access: bool = True) -> object | None:
251 """
252 Retrieve an object from the internal cache.
254 Parameters
255 ----------
256 cache_type : str
257 Type of cache (user, group, or project)
258 cache_key : tuple
259 Key to look up in the cache
260 log_access : bool, optional
261 Whether to log the cache access (default: True)
263 Returns
264 -------
265 object or None
266 The cached object if found, None otherwise
267 """
268 if log_access:
269 self._logger.debug(
270 f'Checking if an instance is registered in cache, for type "{cache_type}" and key "{cache_key}".'
271 )
272 return self.__cache[cache_type].get(cache_key)
274 def __register_in_cache(self, *, cache_type: str, cache_key: tuple, obj: object, obj_short: LynceusDict):
275 """
276 Register an object in the internal cache.
278 Parameters
279 ----------
280 cache_type : str
281 Type of cache (user, group, or project)
282 cache_key : tuple
283 Key to store the object under
284 obj : object
285 The complete object to cache
286 obj_short : object
287 Short representation of the object for logging
288 """
289 # Safe-guard: checks if it has already been registered in cache.
290 from_cache = self.__get_from_cache(
291 cache_type=cache_type, cache_key=cache_key, log_access=False
292 )
293 if from_cache:
294 self._logger.warning(
295 f'An instance of type "{cache_type}" has already been registered in cache, for key "{cache_key}". It will be overridden.'
296 )
298 self._logger.debug(
299 f'Registering a complete/long instance of type "{cache_type}" in cache, for key "{cache_key}", whose short version is: "{obj_short}".'
300 )
301 self.__cache[cache_type][cache_key] = obj
303 # The following methods are only performing read access on DevOps backend.
304 def authenticate(self):
305 """
306 Authenticate with the DevOps platform.
308 Raises
309 ------
310 NotImplementedError
311 Must be implemented by subclasses
312 """
313 raise NotImplementedError()
315 def _do_get_current_user(self):
316 """
317 Get the current authenticated user from the platform.
319 Returns
320 -------
321 object
322 Platform-specific user object
324 Raises
325 ------
326 NotImplementedError
327 Must be implemented by subclasses
328 """
329 raise NotImplementedError()
331 def get_current_user(self):
332 """
333 Get standardized information about the current authenticated user.
335 Returns
336 -------
337 LynceusDict
338 Standardized current user information
339 """
340 self._logger.debug("Retrieving current user information.")
341 user = self._do_get_current_user()
343 return self._extract_user_info(user)
345 def _do_get_user_without_cache(
346 self, *, username: str = None, email: str = None, **kwargs
347 ):
348 """
349 Get a user from the platform without using cache.
351 Parameters
352 ----------
353 username : str, optional
354 Username to search for
355 email : str, optional
356 Email to search for
357 **kwargs
358 Additional search parameters
360 Returns
361 -------
362 object
363 Platform-specific user object
365 Raises
366 ------
367 NotImplementedError
368 Must be implemented by subclasses
369 """
370 raise NotImplementedError()
372 def _do_get_user(self, *, username: str = None, email: str = None, **kwargs):
373 """
374 Get a user from the platform with caching support.
376 Parameters
377 ----------
378 username : str, optional
379 Username to search for
380 email : str, optional
381 Email to search for
382 **kwargs
383 Additional search parameters
385 Returns
386 -------
387 object
388 Platform-specific user object (from cache or fresh)
389 """
390 # Checks if available in cache.
391 cache_key: tuple = (username, email)
392 user = self.__get_from_cache(
393 cache_type=self.CACHE_USER_TYPE, cache_key=cache_key
394 )
396 # Retrieves it if not available in cache.
397 if not user:
398 user = self._do_get_user_without_cache(
399 username=username, email=email, **kwargs
400 )
402 # Registers in cache.
403 self.__register_in_cache(
404 cache_type=self.CACHE_USER_TYPE,
405 cache_key=cache_key,
406 obj=user,
407 obj_short=self._extract_user_info(user),
408 )
410 return user
412 def get_user(self, *, username: str = None, email: str = None, **kwargs):
413 """
414 Get standardized information about a specific user.
416 Parameters
417 ----------
418 username : str, optional
419 Username to search for
420 email : str, optional
421 Email to search for
422 **kwargs
423 Additional search parameters
425 Returns
426 -------
427 LynceusDict
428 Standardized user information
429 """
430 self._logger.debug(f'Retrieving user "{username=}" with "{email}" ({kwargs=}).')
431 user = self._do_get_user(username=username, email=email, **kwargs)
432 return self._extract_user_info(user)
434 def _do_get_groups(self, *, count: int | None = None, **kwargs):
435 """
436 Get groups from the platform.
438 Parameters
439 ----------
440 count : int, optional
441 Maximum number of groups to retrieve
442 **kwargs
443 Additional filtering parameters
445 Returns
446 -------
447 list
448 List of platform-specific group objects
450 Raises
451 ------
452 NotImplementedError
453 Must be implemented by subclasses
454 """
455 raise NotImplementedError()
457 def get_groups(self, *, count: int | None = None, **kwargs):
458 """
459 Get standardized information about groups.
461 Parameters
462 ----------
463 count : int, optional
464 Maximum number of groups to retrieve
465 **kwargs
466 Additional filtering parameters
468 Returns
469 -------
470 List[LynceusDict]
471 List of standardized group information
472 """
473 self._logger.debug(f"Retrieving groups ({count=}; {kwargs=}).")
474 groups = self._do_get_groups(count=count, **kwargs)
475 return [self._extract_group_info(group) for group in groups]
477 def _do_get_group_without_cache(self, *, full_path: str, **kwargs):
478 """
479 Get a group from the platform without using cache.
481 Parameters
482 ----------
483 full_path : str
484 Full path of the group
485 **kwargs
486 Additional search parameters
488 Returns
489 -------
490 object
491 Platform-specific group object
493 Raises
494 ------
495 NotImplementedError
496 Must be implemented by subclasses
497 """
498 raise NotImplementedError()
500 def _do_get_group(self, *, full_path: str, **kwargs):
501 """
502 Get a group from the platform with caching support.
504 Parameters
505 ----------
506 full_path : str
507 Full path of the group
508 **kwargs
509 Additional search parameters
511 Returns
512 -------
513 object
514 Platform-specific group object (from cache or fresh)
515 """
516 # Checks if available in cache.
517 cache_key: tuple = (full_path,)
518 group = self.__get_from_cache(
519 cache_type=self.CACHE_GROUP_TYPE, cache_key=cache_key
520 )
522 # Retrieves it if not available in cache.
523 if not group:
524 group = self._do_get_group_without_cache(full_path=full_path, **kwargs)
526 # Registers in cache.
527 self.__register_in_cache(
528 cache_type=self.CACHE_GROUP_TYPE,
529 cache_key=cache_key,
530 obj=group,
531 obj_short=self._extract_group_info(group),
532 )
534 return group
536 def get_group(self, *, full_path: str, **kwargs):
537 """
538 Get standardized information about a specific group.
540 Parameters
541 ----------
542 full_path : str
543 Full path of the group
544 **kwargs
545 Additional search parameters
547 Returns
548 -------
549 LynceusDict
550 Standardized group information
551 """
552 self._logger.debug(f'Retrieving group "{full_path=}" ({kwargs=}).')
553 group = self._do_get_group(full_path=full_path, **kwargs)
554 return self._extract_group_info(group)
556 def _do_get_projects(self, *, count: int | None = None, **kwargs):
557 """
558 Get projects from the platform.
560 Parameters
561 ----------
562 count : int, optional
563 Maximum number of projects to retrieve
564 **kwargs
565 Additional filtering parameters
567 Returns
568 -------
569 list
570 List of platform-specific project objects
572 Raises
573 ------
574 NotImplementedError
575 Must be implemented by subclasses
576 """
577 raise NotImplementedError()
579 def get_projects(self, *, count: int | None = None, **kwargs):
580 """
581 Get standardized information about projects.
583 Parameters
584 ----------
585 count : int, optional
586 Maximum number of projects to retrieve
587 **kwargs
588 Additional filtering parameters
590 Returns
591 -------
592 List[LynceusDict]
593 List of standardized project information
594 """
595 self._logger.debug(f"Retrieving projects ({count=}; {kwargs=}).")
596 projects = self._do_get_projects(count=count, **kwargs)
597 return [self._extract_project_info(project) for project in projects]
599 def _do_get_project_without_cache(self, *, full_path: str, **kwargs):
600 """
601 Get a project from the platform without using cache.
603 Parameters
604 ----------
605 full_path : str
606 Full path of the project (including namespace)
607 **kwargs
608 Additional search parameters
610 Returns
611 -------
612 object
613 Platform-specific project object
615 Raises
616 ------
617 NotImplementedError
618 Must be implemented by subclasses
619 """
620 raise NotImplementedError()
622 def _do_get_project(self, *, full_path: str, **kwargs):
623 """
624 Get a project from the platform with caching support.
626 Parameters
627 ----------
628 full_path : str
629 Full path of the project (including namespace)
630 **kwargs
631 Additional search parameters
633 Returns
634 -------
635 object
636 Platform-specific project object (from cache or fresh)
637 """
638 # Checks if available in cache.
639 cache_key: tuple = (full_path,)
640 project = self.__get_from_cache(
641 cache_type=self.CACHE_PROJECT_TYPE, cache_key=cache_key
642 )
644 # Retrieves it if not available in cache.
645 if not project:
646 project = self._do_get_project_without_cache(full_path=full_path, **kwargs)
648 # Registers in cache.
649 self.__register_in_cache(
650 cache_type=self.CACHE_PROJECT_TYPE,
651 cache_key=cache_key,
652 obj=project,
653 obj_short=self._extract_project_info(project),
654 )
656 return project
658 def get_project(self, *, full_path: str, **kwargs):
659 """
660 Get standardized information about a specific project.
662 Parameters
663 ----------
664 full_path : str
665 Full path of the project (including namespace)
666 **kwargs
667 Additional search parameters
669 Returns
670 -------
671 LynceusDict
672 Standardized project information
673 """
674 self._logger.debug(f'Retrieving project "{full_path=}" ({kwargs=}).')
675 project = self._do_get_project(full_path=full_path, **kwargs)
676 return self._extract_project_info(project)
678 def check_permissions_on_project(
679 self,
680 *,
681 full_path: str,
682 get_metadata: bool,
683 pull: bool,
684 push: bool = False,
685 maintain: bool = False,
686 admin: bool = False,
687 **kwargs,
688 ):
689 """
690 Check permissions of authenticated user on specific project.
692 Parameters
693 ----------
694 full_path : str
695 Full path of the project (including namespace)
696 get_metadata : bool
697 Check 'get project metadata' permission
698 pull : bool
699 Check 'pull' permission
700 push : bool, optional
701 Check 'push' permission (default: False)
702 maintain : bool, optional
703 Check 'maintain' permission (default: False)
704 admin : bool, optional
705 Check 'admin' permission (default: False)
706 **kwargs
707 Optional additional parameters
709 Returns
710 -------
711 bool
712 True if the authenticated user has requested permission, False otherwise
713 """
714 raise NotImplementedError()
716 def _do_get_project_members(
717 self, *, full_path: str, count: int | None = None, **kwargs
718 ):
719 """
720 Get project members from the platform.
722 Parameters
723 ----------
724 full_path : str
725 Full path of the project (including namespace)
726 count : int, optional
727 Maximum number of members to retrieve
728 **kwargs
729 Additional filtering parameters
731 Returns
732 -------
733 list
734 List of platform-specific member objects
736 Raises
737 ------
738 NotImplementedError
739 Must be implemented by subclasses
740 """
741 raise NotImplementedError()
743 def get_project_commits(
744 self, *, full_path: str, git_ref_name: str, count: int | None = None, **kwargs
745 ):
746 """
747 Get standardized information about project commits.
749 Parameters
750 ----------
751 full_path : str
752 Full path of the project (including namespace)
753 git_ref_name : str
754 Git reference name (branch, tag, or commit)
755 count : int, optional
756 Maximum number of commits to retrieve
757 **kwargs
758 Additional filtering parameters
760 Returns
761 -------
762 List[LynceusDict]
763 List of standardized commit information
764 """
765 self._logger.debug(
766 f'Retrieving commits from git_reference "{git_ref_name}" of project "{full_path}" ({count=}; {kwargs=}).'
767 )
768 commits = self._do_get_project_commits(
769 full_path=full_path, git_ref_name=git_ref_name, count=count, **kwargs
770 )
771 return [self._extract_commit_info(commit) for commit in commits]
773 def _do_get_project_commits(
774 self, *, full_path: str, git_ref_name: str, count: int | None = None, **kwargs
775 ):
776 """
777 Get project commits from the platform.
779 Parameters
780 ----------
781 full_path : str
782 Full path of the project (including namespace)
783 git_ref_name : str
784 Git reference name (branch, tag, or commit)
785 count : int, optional
786 Maximum number of commits to retrieve
787 **kwargs
788 Additional filtering parameters
790 Returns
791 -------
792 list
793 List of platform-specific commit objects
795 Raises
796 ------
797 NotImplementedError
798 Must be implemented by subclasses
799 """
800 raise NotImplementedError()
802 def get_project_branches(
803 self, *, full_path: str, count: int | None = None, **kwargs
804 ):
805 """
806 Get standardized information about project branches.
808 Parameters
809 ----------
810 full_path : str
811 Full path of the project (including namespace)
812 count : int, optional
813 Maximum number of branches to retrieve
814 **kwargs
815 Additional filtering parameters
817 Returns
818 -------
819 List[LynceusDict]
820 List of standardized branch information
821 """
822 self._logger.debug(
823 f'Retrieving branches of project "{full_path=}" ({count=}; {kwargs=}).'
824 )
825 branches = self._do_get_project_branches(
826 full_path=full_path, count=count, **kwargs
827 )
828 return [self._extract_branch_info(branch) for branch in branches]
830 def _do_get_project_branches(
831 self, *, full_path: str, count: int | None = None, **kwargs
832 ):
833 """
834 Get project branches from the platform.
836 Parameters
837 ----------
838 full_path : str
839 Full path of the project (including namespace)
840 count : int, optional
841 Maximum number of branches to retrieve
842 **kwargs
843 Additional filtering parameters
845 Returns
846 -------
847 list
848 List of platform-specific branch objects
850 Raises
851 ------
852 NotImplementedError
853 Must be implemented by subclasses
854 """
855 raise NotImplementedError()
857 def get_project_tags(self, *, full_path: str, count: int | None = None, **kwargs):
858 """
859 Get standardized information about project tags.
861 Parameters
862 ----------
863 full_path : str
864 Full path of the project (including namespace)
865 count : int, optional
866 Maximum number of tags to retrieve
867 **kwargs
868 Additional filtering parameters
870 Returns
871 -------
872 List[LynceusDict]
873 List of standardized tag information
874 """
875 self._logger.debug(
876 f'Retrieving tags of project "{full_path=}" ({count=}; {kwargs=}).'
877 )
878 tags = self._do_get_project_tags(full_path=full_path, count=count, **kwargs)
879 return [self._extract_tag_info(tag) for tag in tags]
881 def _do_get_project_tags(
882 self, *, full_path: str, count: int | None = None, **kwargs
883 ):
884 """
885 Get project tags from the platform.
887 Parameters
888 ----------
889 full_path : str
890 Full path of the project (including namespace)
891 count : int, optional
892 Maximum number of tags to retrieve
893 **kwargs
894 Additional filtering parameters
896 Returns
897 -------
898 list
899 List of platform-specific tag objects
901 Raises
902 ------
903 NotImplementedError
904 Must be implemented by subclasses
905 """
906 raise NotImplementedError()
908 def get_project_members(
909 self, *, full_path: str, count: int | None = None, **kwargs
910 ):
911 """
912 Get standardized information about project members.
914 Parameters
915 ----------
916 full_path : str
917 Full path of the project (including namespace)
918 count : int, optional
919 Maximum number of members to retrieve
920 **kwargs
921 Additional filtering parameters
923 Returns
924 -------
925 List[LynceusDict]
926 List of standardized member information
927 """
928 self._logger.debug(
929 f'Retrieving members of project "{full_path=}" ({count=}; {kwargs=}).'
930 )
931 project_members = self._do_get_project_members(
932 full_path=full_path, count=count, **kwargs
933 )
934 return [self._extract_member_info(member) for member in project_members]
936 def _do_get_group_members(
937 self, *, full_path: str, count: int | None = None, **kwargs
938 ):
939 """
940 Get group members from the platform.
942 Parameters
943 ----------
944 full_path : str
945 Full path of the group
946 count : int, optional
947 Maximum number of members to retrieve
948 **kwargs
949 Additional filtering parameters
951 Returns
952 -------
953 list
954 List of platform-specific member objects
956 Raises
957 ------
958 NotImplementedError
959 Must be implemented by subclasses
960 """
961 raise NotImplementedError()
963 def get_group_members(self, *, full_path: str, count: int | None = None, **kwargs):
964 """
965 Return all members of group (aka Organization).
967 Parameters
968 ----------
969 full_path : str
970 Full path of the group (including namespace)
971 count : int, optional
972 Maximum number of members to retrieve
973 **kwargs
974 Additional filtering parameters
976 Returns
977 -------
978 List[LynceusDict]
979 List of standardized member information
980 """
981 self._logger.debug(
982 f'Retrieving members of group "{full_path=}" ({count=}; {kwargs=}).'
983 )
984 members = self._do_get_group_members(full_path=full_path, count=count, **kwargs)
985 return [self._extract_member_info(member) for member in members]
987 def _do_get_project_issue_events(
988 self,
989 *,
990 full_path: str,
991 action: str | None = None,
992 from_date: datetime | None = None,
993 to_date: datetime | None = None,
994 count: int | None = None,
995 **kwargs,
996 ):
997 """
998 Get project issue events from the platform.
1000 Parameters
1001 ----------
1002 full_path : str
1003 Full path of the project (including namespace)
1004 action : str, optional
1005 Issue event action to filter on
1006 from_date : datetime, optional
1007 Datetime from which to consider issue events
1008 to_date : datetime, optional
1009 Datetime until which to consider issue events
1010 count : int, optional
1011 Maximum number of events to retrieve
1012 **kwargs
1013 Additional filtering parameters
1015 Returns
1016 -------
1017 list
1018 List of platform-specific issue event objects
1020 Raises
1021 ------
1022 NotImplementedError
1023 Must be implemented by subclasses
1024 """
1025 raise NotImplementedError()
1027 def get_project_issue_events(
1028 self,
1029 *,
1030 full_path: str,
1031 action: str | None = None,
1032 from_date: datetime | None = None,
1033 to_date: datetime | None = None,
1034 count: int | None = None,
1035 **kwargs,
1036 ):
1037 """
1038 Return project issue events filtered according to specified parameters.
1040 Parameters
1041 ----------
1042 full_path : str
1043 Path of the project (including namespace)
1044 action : str, optional
1045 Issue event action to filter on, most of them are common to DevOps.
1046 See GitLab specs: https://docs.gitlab.com/ee/user/profile/index.html#user-contribution-events
1047 See GitHub specs: https://docs.github.com/en/developers/webhooks-and-events/events/github-event-types#event-payload-object-6
1048 from_date : datetime, optional
1049 Datetime from which to consider issue events
1050 to_date : datetime, optional
1051 Datetime until which to consider issue events
1052 count : int, optional
1053 Maximum number of events to retrieve
1054 **kwargs
1055 Additional filtering parameters
1057 Returns
1058 -------
1059 List[LynceusDict]
1060 Filtered issue events of the project
1061 """
1062 self._logger.debug(
1063 f'Retrieving issue events of project "{full_path=}" ({action=}; {from_date=}; {to_date=}; {count=}; {kwargs=}).'
1064 )
1065 project_events = self._do_get_project_issue_events(
1066 full_path=full_path,
1067 action=action,
1068 from_date=from_date,
1069 to_date=to_date,
1070 count=count,
1071 **kwargs,
1072 )
1074 # Important: in Github there is no project (either id or name) information attached to event, and on Gitlab there is only the id.
1075 # To return consistent result, we add both here.
1076 project = self.get_project(full_path=full_path)
1077 project_metadata = {
1078 "project_id": project.id,
1079 "project_name": project.name,
1080 "project_web_url": project.web_url,
1081 }
1083 return [
1084 self._extract_issue_event_info(event, **project_metadata)
1085 for event in project_events
1086 ]
1088 def _do_get_project_issues(
1089 self, *, full_path: str, count: int | None = None, **kwargs
1090 ):
1091 """
1092 Get project issues from the platform.
1094 Parameters
1095 ----------
1096 full_path : str
1097 Full path of the project (including namespace)
1098 count : int, optional
1099 Maximum number of issues to retrieve
1100 **kwargs
1101 Additional filtering parameters
1103 Returns
1104 -------
1105 list
1106 List of platform-specific issue objects
1108 Raises
1109 ------
1110 NotImplementedError
1111 Must be implemented by subclasses
1112 """
1113 raise NotImplementedError()
1115 def get_project_issues(self, *, full_path: str, count: int | None = None, **kwargs):
1116 """
1117 Get all issues of a project.
1119 Parameters
1120 ----------
1121 full_path : str
1122 Full path of the project (including namespace)
1123 count : int, optional
1124 Maximum number of issues to retrieve
1125 **kwargs
1126 Additional filtering parameters
1128 Returns
1129 -------
1130 list
1131 List of platform-specific issue objects
1132 """
1133 self._logger.debug(
1134 f'Retrieving issues of project "{full_path=}" ({count=}; {kwargs=}).'
1135 )
1136 return self._do_get_project_issues(full_path=full_path, count=count, **kwargs)
1138 def get_project_pull_requests(
1139 self, *, full_path: str, count: int | None = None, **kwargs
1140 ):
1141 """
1142 Get all pull requests of a project (alias for merge requests).
1144 Parameters
1145 ----------
1146 full_path : str
1147 Full path of the project (including namespace)
1148 count : int, optional
1149 Maximum number of pull requests to retrieve
1150 **kwargs
1151 Additional filtering parameters
1153 Returns
1154 -------
1155 list
1156 List of platform-specific pull/merge request objects
1157 """
1158 return self.get_project_merge_requests(
1159 full_path=full_path, count=count, **kwargs
1160 )
1162 def _do_get_project_merge_requests(
1163 self, *, full_path: str, count: int | None = None, **kwargs
1164 ):
1165 """
1166 Get project merge requests from the platform.
1168 Parameters
1169 ----------
1170 full_path : str
1171 Full path of the project (including namespace)
1172 count : int, optional
1173 Maximum number of merge requests to retrieve
1174 **kwargs
1175 Additional filtering parameters
1177 Returns
1178 -------
1179 list
1180 List of platform-specific merge request objects
1182 Raises
1183 ------
1184 NotImplementedError
1185 Must be implemented by subclasses
1186 """
1187 raise NotImplementedError()
1189 def get_project_merge_requests(
1190 self, *, full_path: str, count: int | None = None, **kwargs
1191 ):
1192 """
1193 Get all merge requests of a project.
1195 Parameters
1196 ----------
1197 full_path : str
1198 Full path of the project (including namespace)
1199 count : int, optional
1200 Maximum number of merge requests to retrieve
1201 **kwargs
1202 Additional filtering parameters
1204 Returns
1205 -------
1206 list
1207 List of platform-specific merge request objects
1208 """
1209 self._logger.debug(
1210 f'Retrieving merge/pull requests of project "{full_path=}" ({count=}; {kwargs=}).'
1211 )
1212 return self._do_get_project_merge_requests(
1213 full_path=full_path, count=count, **kwargs
1214 )
1216 def _do_get_project_milestones(
1217 self, *, full_path: str, count: int | None = None, **kwargs
1218 ):
1219 """
1220 Get project milestones from the platform.
1222 Parameters
1223 ----------
1224 full_path : str
1225 Full path of the project (including namespace)
1226 count : int, optional
1227 Maximum number of milestones to retrieve
1228 **kwargs
1229 Additional filtering parameters
1231 Returns
1232 -------
1233 list
1234 List of platform-specific milestone objects
1236 Raises
1237 ------
1238 NotImplementedError
1239 Must be implemented by subclasses
1240 """
1241 raise NotImplementedError()
1243 def get_project_milestones(
1244 self, *, full_path: str, count: int | None = None, **kwargs
1245 ):
1246 """
1247 Get all milestones (sprints) of a project.
1249 Parameters
1250 ----------
1251 full_path : str
1252 Full path of the project (including namespace)
1253 count : int, optional
1254 Maximum number of milestones to retrieve
1255 **kwargs
1256 Additional filtering parameters
1258 Returns
1259 -------
1260 list
1261 List of platform-specific milestone objects
1262 """
1263 self._logger.debug(
1264 f'Retrieving milestones of project "{full_path=}" ({count=}; {kwargs=}).'
1265 )
1266 return self._do_get_project_milestones(
1267 full_path=full_path, count=count, **kwargs
1268 )
1270 def _do_get_group_projects(
1271 self, *, full_path: str, count: int | None = None, **kwargs
1272 ):
1273 """
1274 Get all projects of a group from the platform.
1276 Parameters
1277 ----------
1278 full_path : str
1279 Full path of the group
1280 count : int, optional
1281 Maximum number of projects to retrieve
1282 **kwargs
1283 Additional filtering parameters
1285 Returns
1286 -------
1287 list
1288 List of platform-specific project objects
1290 Raises
1291 ------
1292 NotImplementedError
1293 Must be implemented by subclasses
1294 """
1295 raise NotImplementedError()
1297 def get_group_projects(self, *, full_path: str, count: int | None = None, **kwargs):
1298 """
1299 Get standardized information about all projects in a group.
1301 Parameters
1302 ----------
1303 full_path : str
1304 Full path of the group
1305 count : int, optional
1306 Maximum number of projects to retrieve
1307 **kwargs
1308 Additional filtering parameters
1310 Returns
1311 -------
1312 List[LynceusDict]
1313 List of standardized project information
1314 """
1315 self._logger.debug(
1316 f'Retrieving projects of group "{full_path=}" ({count=}; {kwargs=}).'
1317 )
1318 projects = self._do_get_group_projects(
1319 full_path=full_path, count=count, **kwargs
1320 )
1321 return [self._extract_project_info(project) for project in projects]
1323 def _do_get_group_milestones(
1324 self, *, full_path: str, count: int | None = None, **kwargs
1325 ):
1326 """
1327 Get group milestones from the platform.
1329 Parameters
1330 ----------
1331 full_path : str
1332 Full path of the group
1333 count : int, optional
1334 Maximum number of milestones to retrieve
1335 **kwargs
1336 Additional filtering parameters
1338 Returns
1339 -------
1340 list
1341 List of platform-specific milestone objects
1343 Raises
1344 ------
1345 NotImplementedError
1346 Must be implemented by subclasses
1347 """
1348 raise NotImplementedError()
1350 def get_group_milestones(
1351 self, *, full_path: str, count: int | None = None, **kwargs
1352 ):
1353 """
1354 Get all milestones (sprints) of a group.
1356 Parameters
1357 ----------
1358 full_path : str
1359 Full path of the group
1360 count : int, optional
1361 Maximum number of milestones to retrieve
1362 **kwargs
1363 Additional filtering parameters
1365 Returns
1366 -------
1367 list
1368 List of platform-specific milestone objects
1369 """
1370 self._logger.debug(
1371 f'Retrieving milestones of group "{full_path=}" ({count=}; {kwargs=}).'
1372 )
1373 return self._do_get_group_milestones(full_path=full_path, count=count, **kwargs)
1375 def get_user_stats_commit_activity(
1376 self,
1377 *,
1378 group_full_path: str = None,
1379 project_full_path: str = None,
1380 since: datetime | None = None,
1381 keep_empty_stats: bool = False,
1382 count: int | None = None,
1383 ):
1384 """
1385 Compute commit activity statistics for the authenticated user.
1387 Compute total statistics of all accessible projects of authenticated user,
1388 or all projects of specified group and/or of specified project.
1390 Parameters
1391 ----------
1392 group_full_path : str, optional
1393 Group path - statistics will be computed on all projects of this group
1394 project_full_path : str, optional
1395 Project path - statistics will be computed on this project
1396 since : datetime, optional
1397 Start date from which statistics must be computed
1398 keep_empty_stats : bool, optional
1399 Whether to include days with zero commits in results (default: False)
1400 count : int, optional
1401 Maximum number of projects to analyze
1403 Returns
1404 -------
1405 dict
1406 Dictionary mapping dates to commit counts
1408 Raises
1409 ------
1410 NotImplementedError
1411 Must be implemented by subclasses
1412 """
1413 raise NotImplementedError()
1415 def get_user_contributions(
1416 self,
1417 *,
1418 since: datetime = None,
1419 keep_empty_stats: bool = False,
1420 count: int | None = None,
1421 ):
1422 """
1423 Get user contribution statistics including additions, deletions, and commits.
1425 Parameters
1426 ----------
1427 since : datetime, optional
1428 Start date from which contributions must be computed
1429 keep_empty_stats : bool, optional
1430 Whether to include periods with zero contributions (default: False)
1431 count : int, optional
1432 Maximum number of projects to analyze
1434 Returns
1435 -------
1436 dict
1437 Dictionary mapping dates to contribution statistics
1439 Raises
1440 ------
1441 NotImplementedError
1442 Must be implemented by subclasses
1443 """
1444 raise NotImplementedError()
1446 def get_user_stats_code_frequency(self, *, count: int | None = None):
1447 """
1448 Get code frequency statistics showing additions and deletions over time.
1450 Parameters
1451 ----------
1452 count : int, optional
1453 Maximum number of projects to analyze
1455 Returns
1456 -------
1457 dict
1458 Dictionary mapping time periods to code frequency data
1460 Raises
1461 ------
1462 NotImplementedError
1463 Must be implemented by subclasses
1464 """
1465 raise NotImplementedError()
1467 def download_repository(
1468 self,
1469 *,
1470 project_full_path: str,
1471 dest_path: Path,
1472 reference: str = None,
1473 chunk_size: int = 1024,
1474 **kwargs,
1475 ):
1476 """
1477 Download a repository as an archive file.
1479 Parameters
1480 ----------
1481 project_full_path : str
1482 Full path of the project to download
1483 dest_path : Path
1484 Destination path for the downloaded archive
1485 reference : str, optional
1486 Git reference to download (branch, tag, commit). Uses default if None
1487 chunk_size : int, optional
1488 Size of chunks for streaming download (default: 1024 bytes)
1489 **kwargs
1490 Additional parameters for the download
1492 Raises
1493 ------
1494 ValueError
1495 If the git reference cannot be accessed or downloaded
1496 NameError
1497 If the project or reference is not found
1498 """
1499 self._logger.debug(
1500 f'Starting repository download of project "{project_full_path}" with git reference "{reference}", to destination file "{dest_path}" ...'
1501 )
1502 try:
1503 self._do_download_repository(
1504 project_full_path=project_full_path,
1505 dest_path=dest_path,
1506 reference=reference,
1507 chunk_size=chunk_size,
1508 **kwargs,
1509 )
1510 self._logger.info(
1511 f'Successfully downloaded repository of project "{project_full_path}" with git reference "{reference}", to destination file "{dest_path}".'
1512 )
1513 except NameError as exc:
1514 git_reference_str: str = (
1515 f'git reference "{reference}"'
1516 if reference is not None
1517 else "default git reference"
1518 )
1519 error_message: str = (
1520 f'Unable to download/access {git_reference_str} of project "{project_full_path}" (ensure your Token has permissions enough).'
1521 )
1522 self._logger.warning(error_message)
1523 raise ValueError(error_message) from exc
1525 def _do_download_repository(
1526 self,
1527 *,
1528 project_full_path: str,
1529 dest_path: Path,
1530 reference: str = None,
1531 chunk_size: int = 1024,
1532 **kwargs,
1533 ):
1534 """
1535 Platform-specific implementation for downloading a repository.
1537 Parameters
1538 ----------
1539 project_full_path : str
1540 Full path of the project to download
1541 dest_path : Path
1542 Destination path for the downloaded archive
1543 reference : str, optional
1544 Git reference to download (branch, tag, commit)
1545 chunk_size : int, optional
1546 Size of chunks for streaming download (default: 1024)
1547 **kwargs
1548 Additional platform-specific parameters
1550 Raises
1551 ------
1552 NotImplementedError
1553 Must be implemented by subclasses
1554 """
1555 raise NotImplementedError()
1557 def uncompress_tarball(
1558 self,
1559 *,
1560 tmp_dir: Path,
1561 tarball_path: Path,
1562 devops_project_downloaded: bool = True,
1563 ) -> Path:
1564 """
1565 Uncompress a tarball archive to a temporary directory.
1567 Parameters
1568 ----------
1569 tmp_dir : Path
1570 Temporary directory to extract the tarball into
1571 tarball_path : Path
1572 Path to the tarball file to extract
1573 devops_project_downloaded : bool, optional
1574 Whether this is a DevOps project download
1575 (affects validation of extracted structure) (default: True)
1577 Returns
1578 -------
1579 Path
1580 Path to the root directory of the extracted content
1582 Raises
1583 ------
1584 ValueError
1585 If the tarball structure is unexpected for a DevOps project
1586 """
1587 tarball_uncompress_path: Path = tmp_dir / Path("uncompress")
1588 self._logger.debug(
1589 f'Uncompressing the "{tarball_path}" tarball under "{tarball_uncompress_path}" ...'
1590 )
1591 try:
1592 with tarfile.open(tarball_path) as tarball:
1593 tarball.extractall(tarball_uncompress_path, filter='data')
1594 self._logger.info(f'Successfully uncompressed the "{tarball_path}" tarball ...')
1595 except Exception as exc:
1596 raise LynceusError(f'An error occured while uncompressing "{tarball_path}"') from exc
1598 # Retrieves the root directory of the project.
1599 files_in_root_uncompress_dir = list(tarball_uncompress_path.iterdir())
1600 if devops_project_downloaded:
1601 if len(files_in_root_uncompress_dir) > 1:
1602 raise ValueError(
1603 f"Only one root directory is expected in downloaded repository tarball ({len(files_in_root_uncompress_dir)} found)."
1604 )
1606 return files_in_root_uncompress_dir[0]