Select Git revision
gitlab_classes.py
-
Christopher Bohn authored- Retrieves students' gitlab usernames from the 03 Setup assignment submissions and validates the usernames (by retrieving data about gitlab accounts). - Right now it creates the cloning script - TODO: create the CSV file to track assignment partners
Christopher Bohn authored- Retrieves students' gitlab usernames from the 03 Setup assignment submissions and validates the usernames (by retrieving data about gitlab accounts). - Right now it creates the cloning script - TODO: create the CSV file to track assignment partners
gitlab_classes.py 17.96 KiB
from datetime import datetime
from typing import ClassVar, Dict, Iterable, List, Optional, Set, Union
from gitlab import Gitlab, MAINTAINER_ACCESS
from gitlab.v4.objects import Issue, Project, ProjectCommit, 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 __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
def get_state(self) -> str:
"""
:return: opened or closed
"""
return self.git_issue.state
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 __repr__(self) -> str:
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:
gitlab_commit: ProjectCommit
# noinspection PyShadowingNames
def __init__(self, commit: ProjectCommit):
super().__init__()
self.gitlab_commit = commit
def get_author(self) -> Dict[str, str]:
return {'name': self.gitlab_commit.author_name, 'email': self.gitlab_commit.author_email}
def get_timestamp(self) -> str: # TODO: is a string (vs a datetime) really what we want?
return self.gitlab_commit.created_at
def get_message(self) -> str:
return self.gitlab_commit.message
def is_merge(self) -> bool:
return len(self.gitlab_commit.parent_ids) > 1
# 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.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
# 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)
# 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:
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', etc.
"""
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_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_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
# 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()}.')
"""