from datetime import datetime, date from deprecated import deprecated from functools import reduce from typing import ClassVar, Dict, Iterable, List, Optional, Set, Union from gitlab import Gitlab, MAINTAINER_ACCESS from gitlab.v4.objects import Issue, Project, ProjectCommit, ProjectMilestone, User from config import Config class GitlabSession: __instance: ClassVar[Gitlab] = None @classmethod def get_session(cls) -> Gitlab: if cls.__instance is None: cls.__instance = Gitlab(Config.gitlab_url, private_token=Config.gitlab_api_key) return cls.__instance class GitlabUser: git_user: User def __init__(self, user: Union[int, str, User]): """ Creates a User object, populating the backing git_user instance with the appropriate gitlab.User object :param user: must be either a gitlab.User object, an integer representing the user ID, or a string containing the username """ super().__init__() if isinstance(user, int): # by user id self.git_user = GitlabSession.get_session().users.get(user) elif isinstance(user, str): # by username self.git_user = GitlabSession.get_session().users.list(username=user)[0] else: self.git_user = user def get_user_id(self) -> int: return self.git_user.id def get_name(self) -> str: return self.git_user.name def get_username(self) -> str: return self.git_user.username def get_email(self) -> Optional[str]: # return self.git_user.emails.list()[0] # list() is forbidden try: return self.git_user.email except AttributeError: return None def get_site(self) -> str: return self.git_user.web_url def get_projects(self) -> List["GitlabProject"]: gitlab_projects = self.git_user.projects.list(all=True) projects: List[GitlabProject] = [] for project in gitlab_projects: projects.append(GitlabProject(project)) return projects def __repr__(self) -> str: username = self.get_username() return f'@{username}' def __eq__(self, other: "GitlabUser") -> bool: # if isinstance(other, GitlabUser): return self.get_username() == other.get_username() # else: # return False def __ne__(self, other: "GitlabUser") -> bool: return not self.__eq__(other) class GitlabIssue: git_issue: Issue def __init__(self, issue: Issue): """ Creates an Issue object, populating the backing git_issue instance with the appropriate gitlab.Issue object :param issue: must be a gitlab.Issue object """ super().__init__() self.git_issue = issue def get_universal_issue_id(self) -> int: """ :return: universally-unique identifier """ return self.git_issue.id def get_project_issue_id(self) -> int: """ :return: project-unique identifier """ return self.git_issue.iid def get_title(self) -> str: """ :return: issue's title """ return self.git_issue.title def get_description(self) -> str: """ :return: issue's description """ return self.git_issue.description @deprecated(reason='Use is_open() or is_closed()') def get_state(self) -> str: # TODO, delete after we're sure there are no uses """ :return: opened or closed """ return self.git_issue.state def is_open(self) -> bool: """ :return: True if the issue is open; False if the issue is closed """ return self.git_issue.state == 'opened' def is_closed(self) -> bool: """ :return: True if the issue is closed; False if the issue is open """ return self.git_issue.state == 'closed' def get_created_at(self) -> datetime: """ :return: an "aware" datetime object representing the creation date/time """ created_at: str = self.git_issue.created_at if created_at[-1] in ('z', 'Z'): # Didn't encounter this problem with created_at created_at = created_at[:-1] + '+00:00' return datetime.fromisoformat(created_at) def get_updated_at(self) -> datetime: """ :return: an "aware" datetime object representing the last date/time the issue was updated """ updated_at: str = self.git_issue.updated_at if updated_at[-1] in ('z', 'Z'): # Didn't encounter this problem with updated_at updated_at = updated_at[:-1] + '+00:00' return datetime.fromisoformat(updated_at) def get_closed_at(self) -> Optional[datetime]: """ :return: an "aware" datetime object representing the last date/time the issue was closed, or None if the issue is open """ closed_at: str = self.git_issue.closed_at if closed_at is None: return None else: if closed_at[-1] in ('z', 'Z'): # Did encounter this problem with closed_at closed_at = closed_at[:-1] + '+00:00' return datetime.fromisoformat(closed_at) def get_labels(self) -> Set[str]: """ :return: set of label names """ return set(self.git_issue.labels) # return self.git_issue.labels.list(all=True) def get_page(self) -> str: """ :return: HTTPS URL to issue's page """ return self.git_issue.web_url def get_assignee(self) -> Optional[GitlabUser]: assignee = self.git_issue.assignee if assignee is None: return None else: return GitlabUser(assignee['username']) def get_participants(self) -> List[GitlabUser]: participants = [] for participant in self.git_issue.participants(): participants.append(GitlabUser(participant['username'])) return participants def close(self) -> None: self.git_issue.state_event = 'close' self.git_issue.save() def __repr__(self) -> str: issue_number = self.get_project_issue_id() title = self.get_title() return f'{issue_number}. {title}' def __eq__(self, other: "GitlabIssue") -> bool: return self.get_universal_issue_id() == other.get_universal_issue_id() def __ne__(self, other: "GitlabIssue") -> bool: return not self.__eq__(other) def __hash__(self) -> int: return self.get_universal_issue_id() # other git_issue fields: # project_id # title # description # closed_by user # milestone # assignees list of users # author user # user_notes_count # merge_requests_count # upvotes # downvotes # due_date same date format, or None # confidential # discussion_locked # time_stats # task_completion_status # has_tasks # _links # notes # award_emoji # project # subscribed class GitlabCommit: git_commit: ProjectCommit # noinspection PyShadowingNames def __init__(self, commit: ProjectCommit): super().__init__() self.git_commit = commit def get_author(self) -> Dict[str, str]: return {'name': self.git_commit.author_name, 'email': self.git_commit.author_email} def get_timestamp(self) -> datetime: return datetime.fromisoformat(self.git_commit.created_at) def get_message(self) -> str: return self.git_commit.message def is_merge(self) -> bool: return len(self.git_commit.parent_ids) > 1 def is_revert(self) -> bool: return self.get_message().startswith('Revert') def get_id(self) -> str: return self.git_commit.id def get_short_id(self) -> str: return self.git_commit.short_id # noinspection PyShadowingNames def get_diffs(self) -> List[Dict[str, Union[str, int]]]: diffs: List[Dict[str, Union[str, int]]] = [] gitlab_diffs: List[Dict[str, str]] = self.git_commit.diff() for diff in gitlab_diffs: diffs.append({'file': diff['new_path'], 'text': diff['diff'], '+': diff['diff'].count('\n+'), '-': diff['diff'].count('\n-')}) return diffs # noinspection PyShadowingNames def get_diff_size(self) -> int: insertions = 0 deletions = 0 if not self.is_merge(): for diff in self.get_diffs(): insertions += diff['+'] deletions += diff['-'] return max(insertions, deletions) # noinspection PyShadowingNames def get_diff_size_by_filetype(self) -> Dict[str, int]: binary_like_text_files = {"css", "csv", "fxml", "html"} # TODO soft-code this # noinspection SpellCheckingInspection binary_files = {"docx", "gif", "jpg", "jpeg", "pdf", "png", "pptx", "svg", "xlsx"} # TODO this, too return_diff: Dict[str, int] = {} insert_diff: Dict[str, int] = {} delete_diff: Dict[str, int] = {} if not self.is_merge(): for diff in self.get_diffs(): file = diff['file'].split('.') if len(file) == 1: filetype = 'no type' else: filetype = file[-1] if filetype not in return_diff.keys(): return_diff[filetype] = 0 insert_diff[filetype] = 0 delete_diff[filetype] = 0 if filetype in binary_files.union(binary_like_text_files): insert_diff[filetype] += 1 else: insert_diff[filetype] += diff['+'] delete_diff[filetype] += diff['-'] for key in return_diff.keys(): return_diff[key] = max(insert_diff[key], delete_diff[key]) return return_diff @staticmethod def _number_of_lines_too_long(lines, subject_line_length, message_line_length): # noinspection PyUnusedLocal lines_too_long: int if len(lines) == 1: lines_too_long = 0 if len(lines[0]) <= message_line_length else 1 else: lines_too_long = 0 if len(lines[0]) <= subject_line_length else 1 lines_too_long += reduce((lambda x, y: x + y), list(map(lambda line: 0 if len(line) <= message_line_length else 1, lines[1:]))) return lines_too_long @staticmethod def _has_blank_line_after_subject(lines): if len(lines) == 1: return True else: return lines[1] == '' def is_well_formatted(self, subject_line_length=72, message_line_length=72) -> bool: lines: List[str] = self.get_message().rstrip('\n').split('\n') return self._number_of_lines_too_long(lines, subject_line_length, message_line_length) == 0 \ and self._has_blank_line_after_subject(lines) def detail_formatting_problems(self, subject_line_length=72, message_line_length=72) -> str: lines: List[str] = self.get_message().rstrip('\n').split('\n') commit_id: str = f'Commit {self.get_short_id()}' if self.is_well_formatted(): return f'{commit_id} is well-formatted' else: if self._has_blank_line_after_subject(lines): blank_line_comment = '' else: blank_line_comment = 'is missing a blank line after the subject' overlong_lines = self._number_of_lines_too_long(lines, subject_line_length, message_line_length) if overlong_lines == 0: overlong_line_comment = '' elif overlong_lines == 1: overlong_line_comment = 'has 1 line too long' else: overlong_line_comment = f'has {overlong_lines} lines too long' if blank_line_comment == '' or overlong_line_comment == '': conjunction = '' else: conjunction = ' and ' return f'{commit_id} {blank_line_comment}{conjunction}{overlong_line_comment}.' def __eq__(self, other: "GitlabCommit") -> bool: return self.get_id() == other.get_id() def __ne__(self, other: "GitlabCommit") -> bool: return not self.__eq__(other) def __hash__(self) -> int: return hash(self.get_id()) # git_commit fields: # comments # discussions # manager # statuses # attributes: # id # short_id # created_at # parent_ids # title # message # author_name # author_email # authored_date # committer_name # committer_email # committed_date # project_id class GitlabMilestone: git_milestone: ProjectMilestone def __init__(self, milestone: ProjectMilestone): super().__init__() self.git_milestone = milestone def get_issues(self) -> List[GitlabIssue]: """ :return: List of Issue objects representing project's issues """ gitlab_issues: Iterable[Issue] = self.git_milestone.issues() issues: List[GitlabIssue] = [] for issue in gitlab_issues: issues.append(GitlabIssue(issue)) return issues def is_active(self) -> bool: """ :return: True if the milestone is active; False if the milestone is closed """ return self.git_milestone.state == 'active' def is_closed(self) -> bool: """ :return: True if the milestone is closed; False if the milestone is active """ return self.git_milestone.state == 'closed' def close(self) -> None: self.git_milestone.state_event = 'close' self.git_milestone.save() def get_title(self) -> str: """ :return: The milestone's title """ return self.git_milestone.title def get_description(self) -> str: """ :return: The milestone's description """ return self.git_milestone.description def get_start_date(self) -> date: """ :return: a date object representing the start date """ date_segments: List[str] = self.git_milestone.start_date.split('-') year = int(date_segments[0]) month = int(date_segments[1]) day = int(date_segments[2]) return date(year, month, day) # return date.fromisoformat(date_segments) # TODO: requires Python 3.7; I'm using 3.7 - why can't I use this? def get_due_date(self) -> date: """ :return: a date object representing the due date """ date_segments: List[str] = self.git_milestone.due_date.split('-') year = int(date_segments[0]) month = int(date_segments[1]) day = int(date_segments[2]) return date(year, month, day) # return date.fromisoformat(date_segments) # TODO: requires Python 3.7; I'm using 3.7 - why can't I use this? def __repr__(self) -> str: return self.get_title() def __eq__(self, other: "GitlabMilestone") -> bool: return self.git_milestone.id == other.git_milestone.id def __ne__(self, other: "GitlabMilestone") -> bool: return not self.__eq__(other) def __hash__(self) -> int: return self.git_milestone.id # other git_milestone fields: # id # iid # project_id # updated_at # created_at class GitlabProject: git_project: Project def __init__(self, project: Union[int, str, Project]): """ Creates a Project object, populating the backing git_project instance with the appropriate gitlab.Project object :param project: must be either a gitlab.Project object, an integer representing the project ID, or a string containing the project's path (namespace and name, such as 'csce_361/sandbox/HelloWorld') """ super().__init__() if isinstance(project, int): # by project id self.git_project = GitlabSession.get_session().projects.get(project) elif isinstance(project, str): # by path self.git_project = GitlabSession.get_session().projects.get(project) else: # self.git_project = project # for some reason, many attributes (including members) might be lost self.git_project = GitlabSession.get_session().projects.get(project.id) @staticmethod def get_projects_by_group(group: Union[int, str]) -> List["GitlabProject"]: """ :param group: must be either an integer representing the group ID, or a string containing the group's namespace :return: list of projects in the specified group """ if isinstance(group, int): # by group id gitlab_projects: Iterable[Project] = GitlabSession.get_session().groups.get(group).projects.list(all=True) else: # isinstance(group, str): # by path gitlab_projects = GitlabSession.get_session().groups.get(group).projects.list(all=True) projects: List[GitlabProject] = [] for project in gitlab_projects: projects.append(GitlabProject(project)) return projects @staticmethod def get_projects_by_keyword(search_term: str) -> List["GitlabProject"]: gitlab_projects: Iterable[Project] = GitlabSession.get_session().projects.list(search=search_term, all=True) projects: List[GitlabProject] = [] for project in gitlab_projects: projects.append(GitlabProject(project)) return projects @staticmethod def create_project(project_name: str) -> "GitlabProject": return GitlabProject(GitlabSession.get_session().projects.create({'name': project_name})) @staticmethod def create_project_in_group(group_name: str, project_name: str) -> "GitlabProject": group_id = GitlabSession.get_session().groups.get(group_name).id return GitlabProject(GitlabSession.get_session().projects.create({'name': project_name, 'namespace_id': group_id})) def get_project_id(self) -> int: return self.git_project.id def get_name(self) -> str: """ :return: project name without namespace """ return self.git_project.name def get_name_with_namespace(self) -> str: """ :return: project name with namespace, spaces around slashes """ return self.git_project.name_with_namespace def get_path(self) -> str: """ :return: path without namespace (may differ from name if name has spaces) """ return self.git_project.path def get_path_with_namespace(self) -> str: """ :return: path with namespace, no spaces around slashes """ return self.git_project.path_with_namespace def get_cloning_url(self) -> str: """ :return: SSH URL to clone repository """ return self.git_project.ssh_url_to_repo def get_site(self) -> str: """ :return: HTTPS URL to git site """ return self.git_project.web_url def get_readme_url(self) -> str: """ :return: HTTPS URL to README.md """ return self.git_project.readme_url def get_visibility(self) -> str: """ :return: 'private', 'public', or 'internal'. """ return self.git_project.visibility def get_creator(self) -> GitlabUser: """ :return: User object backed by the gitlab.User object representing the user who created the repo """ return GitlabUser(self.git_project.creator_id) def get_created_at(self) -> datetime: """ :return: an "aware" datetime object representing the creation date/time """ created_at: str = self.git_project.created_at if created_at[-1] in ('z', 'Z'): created_at = created_at[:-1] + '+00:00' return datetime.fromisoformat(created_at) def get_users(self) -> List[GitlabUser]: """ :return: List of User objects representing the project's members (not including inherited members) """ gitlab_users: Iterable[User] = self.git_project.members.list(all=True) users: List[GitlabUser] = [] for user in gitlab_users: users.append(GitlabUser(user)) return users def get_all_users(self) -> List[GitlabUser]: """ :return: List of User objects representing all of the project's members (including inherited members) """ gitlab_users: Iterable[User] = self.git_project.members.all(all=True) users: List[GitlabUser] = [] for user in gitlab_users: users.append(GitlabUser(user)) return users def add_user_as_maintainer(self, user: GitlabUser) -> None: self.git_project.members.create({'user_id': user.get_user_id(), 'access_level': MAINTAINER_ACCESS}) def get_milestones(self) -> List[GitlabMilestone]: """ :return: List of Milestone objects representing the project's milestones """ gitlab_milestones: Iterable[ProjectMilestone] = self.git_project.milestones.list() milestones: List[GitlabMilestone] = [] for milestone in gitlab_milestones: milestones.append(GitlabMilestone(milestone)) return milestones def get_issues(self) -> List[GitlabIssue]: """ :return: List of Issue objects representing project's issues, sorted by creation date """ gitlab_issues: Iterable[Issue] = self.git_project.issues.list(order_by='created_at', sort='asc', all=True) issues: List[GitlabIssue] = [] for issue in gitlab_issues: issues.append(GitlabIssue(issue)) return issues def create_issue(self, title: str, description: str) -> GitlabIssue: gitlab_issue: Issue = self.git_project.issues.create({'title': title, 'description': description}) return GitlabIssue(gitlab_issue) # noinspection PyShadowingNames def get_commits(self, branch_name: str = '', after_date: str = '1970-01-01', before_date: str = '9999-12-31') -> \ List[GitlabCommit]: """ :param branch_name: the branch to retrieve commits from; if an empty string (default) then retrieves commits from all branches <-- NO, RETRIEVES FROM DEFAULT BRANCH; WILL NEED TO FIX THAT #TODO :param after_date: the earliest date of any retrieved commit; if '1970-01-01' (Unix epoch) then treated as having no earliest-bound :param before_date: the latest date of any retrieved commit; if '9999-12-31' (Y10K problem) then treated a having no latest-bound :return: List of Commit objects representing the project's commits that meet the specified constraints """ filters: Dict[str, str] = {} if branch_name != '': filters['ref_name'] = branch_name if after_date != '1970-01-01': filters['since'] = after_date if before_date != '9999-12-31': filters['until'] = before_date if len(filters) == 0: gitlab_commits: Iterable[ProjectCommit] = self.git_project.commits.list(all=True) else: gitlab_commits = self.git_project.commits.list(all=True, query_parameters=filters) commits: List[GitlabCommit] = [] for commit in gitlab_commits: commits.append(GitlabCommit(commit)) return commits def get_branch_names(self) -> List[str]: branches = self.git_project.branches.list() branch_names: List[str] = list(map(lambda b: b.name, branches)) return branch_names def get_labels(self) -> Set[str]: """ :return: set of label names """ return set(map(lambda label: label.name, self.git_project.labels.list())) def __repr__(self) -> str: return self.get_name_with_namespace() # other git_project fields: # description # default_branch # tag_list # http_url_to_repo https URL to clone repository # avatar_url # star_count # forks_count # last_activity_at # namespace namespace's JSON object # _links JSON object with api/v4 links to self, issues, merge_requests, # repo_branches, labels, events, members # empty_repo # archived # repository_access_level, # resolve_outdated_diff_discussions # container_registry_enabled # issues_enabled # merge_requests_enabled # jobs_enabled # snippets_enabled # issues_access_level # wiki_access_level # builds_access_level # snippets_access_level # shared_runners_enabled # lfs_enabled # import_status # import_error # open_issues_count # runners_token # ci_default_git_depth # public_jobs # build_git_strategy # build_timeout # auto_cancel_pending_pipelines # build_coverage_regex # ci_config_path # shared_with_groups # only_allow_merge_if_pipeline_succeeds # request_access_enabled # only_allow_merge_if_all_discussions_are_resolved # printing_merge_request_link_enabled # merge_method # auto_devops_enabled # auto_devops_deploy_strategy # permissions if __name__ == '__main__': namespace = 'csce_361/sandbox' test_projects = GitlabProject.get_projects_by_group(namespace) print('All projects in sandbox:') for test_project in test_projects: print(test_project) print('Selecting last project. Here are the commits:') test_project = test_projects[-1] commits = test_project.get_commits() diff = commits[-5].get_diffs() print(diff) print(commits[0].get_author()) for commit in commits: print(commit.get_message()) print(f'is a merge? {commit.is_merge()}') print(f'size: {commit.get_diff_size()}') print() """ print('Selecting second project. Here are the members:') test_project = test_projects[1] members = test_project.get_users() for member in members: print(member) print('Here are ALL the members:') members = test_project.get_all_users() for member in members: print(member) print('Here are the issues:') test_issues = test_project.get_issues() for test_issue in test_issues: creation = test_issue.get_created_at() print(f'{test_issue}\tcreated at {creation}.') test_projects = GitlabProject.get_projects_by_keyword('csce361-homework') number_of_projects = len(test_projects) print(f'retrieved {number_of_projects} projects matching \'csce361-homework\'') start_date = datetime(2019, 8, 1, tzinfo=timezone.utc) test_projects = list(filter(lambda p: p.get_created_at() > start_date, test_projects)) new_number_of_projects = len(test_projects) print(f'after culling, there are {new_number_of_projects} projects that were created in/after August 2019') print(f'including {test_projects[0]} created by {test_projects[0].get_creator()} at ' f'{test_projects[0].get_created_at()}.') """