Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from gitlabform.gitlab.core import GitLabCore
class GitLabCommits(GitLabCore):
def get_commit(self, project_and_group_name, sha):
return self._make_requests_to_api("projects/%s/repository/commits/%s", (project_and_group_name, sha))
def get_ahead_and_behind(self, project_and_group_name, protected_branch, feature_branch):
ahead = 0
behind = 0
response = self._make_requests_to_api("projects/%s/repository/compare?from=%s&to=%s",
(project_and_group_name, protected_branch, feature_branch))
if len(response) > 0:
ahead = len(response['commits'])
response = self._make_requests_to_api("projects/%s/repository/compare?from=%s&to=%s",
(project_and_group_name, feature_branch, protected_branch))
if len(response) > 0:
from gitlabform.gitlab.core import GitLabCore
class GitLabPipelines(GitLabCore):
def get_pipelines(self, project_and_group_name, branch):
pipelines = self._make_requests_to_api("projects/%s/pipelines?ref=%s", (project_and_group_name, branch),
paginated=True)
return pipelines
def get_pipeline(self, project_and_group_name, pipeline_id):
pipeline = self._make_requests_to_api("/projects/%s/pipelines/%s", (project_and_group_name, pipeline_id))
return pipeline
def retry_pipeline(self, project_and_group_name, pipeline_id):
pipeline = self._make_requests_to_api("projects/%s/pipelines/%s/retry",
(project_and_group_name, pipeline_id),
method='POST', expected_codes=[200, 201])
return pipeline
from gitlabform.gitlab.core import GitLabCore
class GitLabServices(GitLabCore):
def get_service(self, project_and_group_name, service):
return self._make_requests_to_api("projects/%s/services/%s", (project_and_group_name, service))
def set_service(self, project_and_group_name, service, data):
self._make_requests_to_api("projects/%s/services/%s", (project_and_group_name, service),
'PUT', data, expected_codes=[200, 201])
def delete_service(self, project_and_group_name, service):
self._make_requests_to_api("projects/%s/services/%s", (project_and_group_name, service),
'DELETE', expected_codes=[200, 204])
from gitlabform.gitlab.core import GitLabCore
class GitLabBranches(GitLabCore):
def protect_branch(self, project_and_group_name, branch, developers_can_push, developers_can_merge):
data = {
"id": project_and_group_name,
"branch": branch,
"developers_can_push": developers_can_push,
"developers_can_merge": developers_can_merge,
}
return self._make_requests_to_api("projects/%s/repository/branches/%s/protect",
(project_and_group_name, branch),
method='PUT', data=data,
expected_codes=[200, 201])
def unprotect_branch(self, project_and_group_name, branch):
data = {
"id": project_and_group_name,
import base64
from gitlabform.gitlab.core import GitLabCore
class GitLabRepositories(GitLabCore):
def get_commits_with_string_in_compare_results(self, project_and_group_name, c_from, c_to, with_string):
compare_results = self.compare(project_and_group_name, c_from, c_to)
commits = compare_results['commits']
return [commit for commit in commits if with_string in commit['title']]
def compare(self, project_and_group_name, c_from, c_to):
pid = self._get_project_id(project_and_group_name)
return self._make_requests_to_api("projects/%s/repository/compare?from=%s&to=%s", (pid, c_from, c_to))
def get_file(self, project_and_group_name, branch, path):
pid = self._get_project_id(project_and_group_name)
result = self._make_requests_to_api("projects/%s/repository/files?ref=%s&file_path=%s", (pid, branch, path))
return base64.b64decode(result['content']).decode("utf-8")
def set_file(self, project_and_group_name, branch, path, content, commit_message):
from gitlabform.gitlab.core import GitLabCore, NotFoundException
class GitLabGroups(GitLabCore):
def get_groups(self):
"""
:return: sorted list of groups
"""
result = self._make_requests_to_api("groups?all_available=true", paginated=True)
return sorted(map(lambda x: x['full_path'], result))
def get_projects(self, group):
"""
:param group: group name
:return: sorted list of strings "group/project_name". Note that only projects from "group" namespace are
returned, so if "group" (= members of this group) is also a member of some projects, they won't be
returned here.
"""
try:
from gitlabform.gitlab.core import GitLabCore
class GitLabTags(GitLabCore):
def get_tags(self, project_and_group_name):
return self._make_requests_to_api("projects/%s/repository/tags", project_and_group_name)
def create_tag(self, project_and_group_name, tag_name, ref, message=None):
data = {
'tag_name': tag_name,
'ref': ref,
'message': message,
}
return self._make_requests_to_api("projects/%s/repository/tags", project_and_group_name, method='POST', data=data,
expected_codes=201)
import json
from gitlabform.gitlab.core import GitLabCore, NotFoundException
class GitLabProjects(GitLabCore):
def get_all_projects(self):
"""
:param group: group name
:return: sorted list of ALL projects you have access to, strings "group/project_name"
"""
try:
result = self._make_requests_to_api("projects?order_by=name&sort=asc", paginated=True)
return sorted(map(lambda x: x['path_with_namespace'], result))
except NotFoundException:
return []
def post_deploy_key(self, project_and_group_name, deploy_key):
# deploy_key has to be like this:
# {
# 'title': title,
from gitlabform.gitlab.core import GitLabCore
class GitLabMembers(GitLabCore):
def add_member_to_project(self, project_and_group_name, user, access_level, expires_at):
data = {
"user_id": self._get_user_id(user),
"expires_at": expires_at
}
if access_level is not None:
data['access_level'] = access_level
return self._make_requests_to_api("projects/%s/members", project_and_group_name, method='POST',
data=data, expected_codes=201)
def remove_member_from_project(self, project_and_group_name, user):
return self._make_requests_to_api("projects/%s/members/%s", (project_and_group_name, self._get_user_id(user)),
method='DELETE', expected_codes=204)
from gitlabform.gitlab.core import GitLabCore
class GitLabMergeRequests(GitLabCore):
def create_mr(self, project_and_group_name, source_branch, target_branch, title, description=None):
data = {
"id": project_and_group_name,
"source_branch": source_branch,
"target_branch": target_branch,
"title": title,
"description": description,
}
return self._make_requests_to_api("projects/%s/merge_requests", project_and_group_name,
method='POST', data=data, expected_codes=201)
def accept_mr(self, project_and_group_name, mr_iid):
return self._make_requests_to_api("projects/%s/merge_requests/%s/merge", (project_and_group_name, mr_iid),
method='PUT')