Skip to content
Snippets Groups Projects
Select Git revision
  • aec77cab80d582cd26986f6b3d4c34d808ac4cd2
  • master default protected
2 results

gitlab_classes.py

Blame
  • gitlab_classes.py 16.08 KiB
    from datetime import datetime
    from datetime import timezone
    import gitlab
    from config import Config
    
    
    class GitlabSession:
        __instance = None
    
        @staticmethod
        def get_session():
            if GitlabSession.__instance is None:
                GitlabSession.__instance = gitlab.Gitlab(Config.gitlab_url, private_token=Config.gitlab_api_key)
            return GitlabSession.__instance
    
    
    class GitlabUser:
        def __init__(self, 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):
            return self.git_user.id
    
        def get_name(self):
            return self.git_user.name
    
        def get_username(self):
            return self.git_user.username
    
        def get_site(self):
            return self.git_user.web_url
    
        def __repr__(self):
            username = self.get_username()
            return f'@{username}'
    
        def __eq__(self, other):
            if isinstance(other, GitlabUser):
                return self.get_username() == other.get_username()
            else:
                return False
    
        def __ne__(self, other):
            return not self.__eq__(other)
    
    
    class GitlabIssue:
        def __init__(self, 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):
            """
            :return: universally-unique identifier
            """
            return self.git_issue.id
    
        def get_project_issue_id(self):
            """
            :return: project-unique identifier
            """
            return self.git_issue.iid
    
        def get_title(self):
            """
            :return: issue's title
            """
            return self.git_issue.title
    
        def get_description(self):
            """
            :return: issue's description
            """
            return self.git_issue.description
    
        def get_state(self):
            """
            :return: opened or closed
            """
            return self.git_issue.state
    
        def get_created_at(self):
            """
            :return: an "aware" datetime object representing the creation date/time
            """
            created_at = 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):
            """
            :return: an "aware" datetime object representing the last date/time the issue was updated
            """
            updated_at = 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):
            """
            :return: an "aware" datetime object representing the last date/time the issue was closed, or None if the issue
            is open
            """
            closed_at = 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):
            """
            :return: list of labels
            """
            return self.git_issue.labels.list(all=True)
    
        def get_page(self):
            """
            :return: HTTPS URL to issue's page
            """
            return self.git_issue.web_url
    
        def __repr__(self):
            issue_number = self.get_project_issue_id()
            title = self.get_title()
            return f'{issue_number}. {title}'
    
        # other git_issue fields:
        # project_id
        # title
        # description
        # closed_by             user
        # milestone
        # assignees             list of users
        # author                user
        # assignee              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:
        def __init__(self, commit):
            super().__init__()
            self.gitlab_commit = commit
    
        def get_author(self):
            return {'name': self.gitlab_commit.author_name, 'email': self.gitlab_commit.author_email}
    
        def get_timestamp(self):
            return self.gitlab_commit.created_at
    
        def get_message(self):
            return self.gitlab_commit.message
    
        def is_merge(self):
            return len(self.gitlab_commit.parent_ids) > 1
    
        def get_diffs(self):
            diffs = []
            gitlab_diffs = self.gitlab_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
    
        def get_diff_size(self):
            insertions = 0
            deletions = 0
            if not self.is_merge():
                for diff in self.get_diffs():
                    insertions += diff['+']
                    deletions += diff['-']
            return max(insertions, deletions)
    
        # 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 GitlabProject:
        def __init__(self, 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):
            """
            :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 = 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 = []
            for project in gitlab_projects:
                projects.append(GitlabProject(project))
            return projects
    
        @staticmethod
        def get_projects_by_keyword(search_term):
            gitlab_projects = GitlabSession.get_session().projects.list(search=search_term, all=True)
            projects = []
            for project in gitlab_projects:
                projects.append(GitlabProject(project))
            return projects
    
        @staticmethod
        def create_project(project_name):
            return GitlabProject(GitlabSession.get_session().projects.create({'name': project_name}))
    
        @staticmethod
        def create_project_in_group(group_name, project_name):
            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):
            return self.git_project.id
    
        def get_name(self):
            """
            :return: project name without namespace
            """
            return self.git_project.name
    
        def get_name_with_namespace(self):
            """
            :return: project name with namespace, spaces around slashes
            """
            return self.git_project.name_with_namespace
    
        def get_path(self):
            """
            :return: path without namespace (may differ from name if name has spaces)
            """
            return self.git_project.path
    
        def get_path_with_namespace(self):
            """
            :return: path with namespace, no spaces around slashes
            """
            return self.git_project.path_with_namespace
    
        def get_cloning_url(self):
            """
            :return: SSH URL to clone repository
            """
            return self.git_project.ssh_url_to_repo
    
        def get_site(self):
            """
            :return: HTTPS URL to git site
            """
            return self.git_project.web_url
    
        def get_readme_url(self):
            """
            :return: HTTPS URL to README.md
            """
            return self.git_project.readme_url
    
        def get_visibility(self):
            """
            :return: 'private', etc.
            """
            return self.git_project.visibility
    
        def get_creator(self):
            """
            :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):
            """
            :return: an "aware" datetime object representing the creation date/time
            """
            created_at = 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):
            """
            :return: List of User objects representing the project's members (not including inherited members)
            """
            gitlab_users = self.git_project.members.list(all=True)
            users = []
            for user in gitlab_users:
                users.append(GitlabUser(user))
            return users
    
        def get_all_users(self):
            """
            :return: List of User objects representing all of the project's members (including inherited members)
            """
            gitlab_users = self.git_project.members.all(all=True)
            users = []
            for user in gitlab_users:
                users.append(GitlabUser(user))
            return users
    
        def add_user_as_maintainer(self, user):
            self.git_project.members.create({'user_id': user.get_user_id(), 'access_level': gitlab.MAINTAINER_ACCESS})
    
        def get_issues(self):
            """
            :return: List of Issue objects representing project's issues, sorted by creation date
            """
            gitlab_issues = self.git_project.issues.list(order_by='created_at', sort='asc', all=True)
            issues = []
            for issue in gitlab_issues:
                issues.append(GitlabIssue(issue))
            return issues
    
        def create_issue(self, title, description):
            gitlab_issue = self.git_project.issues.create({'title': title, 'description': description})
            return GitlabIssue(gitlab_issue)
    
        def get_commits(self, branch_name = '', after_date ='1970-01-01', before_date ='9999-12-31'):
            """
            :param branch_name: the branch to retrieve commits from; if an empty string (default) then retrieves commits
            from all branches
            :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 = {}
            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 = self.git_project.commits.list(all=True)
            else:
                gitlab_commits = self.git_project.commits.list(all=True, query_parameters=filters)
            commits = []
            for commit in gitlab_commits:
                commits.append(GitlabCommit(commit))
            return commits
    
        def __repr__(self):
            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
        # resolve_outdated_diff_discussions
        # container_registry_enabled
        # issues_enabled
        # merge_requests_enabled
        # jobs_enabled
        # snippets_enabled
        # issues_access_level
        # repository_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()}.')
        """