# -*- coding: utf-8 -*-

#

# Copyright (C) 2013-2017 Gauvain Pocentek <gauvain@pocentek.net>

#

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU Lesser General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

# GNU Lesser General Public License for more details.

#

# You should have received a copy of the GNU Lesser General Public License

# along with this program. If not, see <http://www.gnu.org/licenses/>.

from __future__ import print_function

from __future__ import absolute_import

import base64

import json

import warnings

import six

from six.moves import urllib

import gitlab

from gitlab.base import * # noqa

from gitlab.exceptions import * # noqa

from gitlab import utils

class SidekiqManager(object):

"""Manager for the Sidekiq methods.

This manager doesn't actually manage objects but provides helper fonction

for the sidekiq metrics API.

"""

def __init__(self, gl):

"""Constructs a Sidekiq manager.

Args:

gl (gitlab.Gitlab): Gitlab object referencing the GitLab server.

"""

self.gitlab = gl

def _simple_get(self, url, **kwargs):

r = self.gitlab._raw_get(url, **kwargs)

raise_error_from_response(r, GitlabGetError)

return r.json()

def queue_metrics(self, **kwargs):

"""Returns the registred queues information."""

return self._simple_get('/sidekiq/queue_metrics', **kwargs)

def process_metrics(self, **kwargs):

"""Returns the registred sidekiq workers."""

return self._simple_get('/sidekiq/process_metrics', **kwargs)

def job_stats(self, **kwargs):

"""Returns statistics about the jobs performed."""

return self._simple_get('/sidekiq/job_stats', **kwargs)

def compound_metrics(self, **kwargs):

"""Returns all available metrics and statistics."""

return self._simple_get('/sidekiq/compound_metrics', **kwargs)

class UserEmail(GitlabObject):

_url = '/users/%(user_id)s/emails'

canUpdate = False

shortPrintAttr = 'email'

requiredUrlAttrs = ['user_id']

requiredCreateAttrs = ['email']

class UserEmailManager(BaseManager):

obj_cls = UserEmail

class UserKey(GitlabObject):

_url = '/users/%(user_id)s/keys'

canGet = 'from_list'

canUpdate = False

requiredUrlAttrs = ['user_id']

requiredCreateAttrs = ['title', 'key']

class UserKeyManager(BaseManager):

obj_cls = UserKey

class UserProject(GitlabObject):

_url = '/projects/user/%(user_id)s'

_constructorTypes = {'owner': 'User', 'namespace': 'Group'}

canUpdate = False

canDelete = False

canList = False

canGet = False

requiredUrlAttrs = ['user_id']

requiredCreateAttrs = ['name']

optionalCreateAttrs = ['default_branch', 'issues_enabled', 'wall_enabled',

'merge_requests_enabled', 'wiki_enabled',

'snippets_enabled', 'public', 'visibility_level',

'description', 'builds_enabled', 'public_builds',

'import_url', 'only_allow_merge_if_build_succeeds']

class UserProjectManager(BaseManager):

obj_cls = UserProject

class User(GitlabObject):

_url = '/users'

shortPrintAttr = 'username'

requiredCreateAttrs = ['email', 'username', 'name']

optionalCreateAttrs = ['password', 'reset_password', 'skype', 'linkedin',

'twitter', 'projects_limit', 'extern_uid',

'provider', 'bio', 'admin', 'can_create_group',

'website_url', 'confirm', 'external',

'organization', 'location']

requiredUpdateAttrs = ['email', 'username', 'name']

optionalUpdateAttrs = ['password', 'skype', 'linkedin', 'twitter',

'projects_limit', 'extern_uid', 'provider', 'bio',

'admin', 'can_create_group', 'website_url',

'confirm', 'external', 'organization', 'location']

managers = (

('emails', 'UserEmailManager', [('user_id', 'id')]),

('keys', 'UserKeyManager', [('user_id', 'id')]),

('projects', 'UserProjectManager', [('user_id', 'id')]),

)

def _data_for_gitlab(self, extra_parameters={}, update=False,

as_json=True):

if hasattr(self, 'confirm'):

self.confirm = str(self.confirm).lower()

return super(User, self)._data_for_gitlab(extra_parameters)

def block(self, **kwargs):

"""Blocks the user."""

url = '/users/%s/block' % self.id

r = self.gitlab._raw_put(url, **kwargs)

raise_error_from_response(r, GitlabBlockError)

self.state = 'blocked'

def unblock(self, **kwargs):

"""Unblocks the user."""

url = '/users/%s/unblock' % self.id

r = self.gitlab._raw_put(url, **kwargs)

raise_error_from_response(r, GitlabUnblockError)

self.state = 'active'

def __eq__(self, other):

if type(other) is type(self):

selfdict = self.as_dict()

otherdict = other.as_dict()

selfdict.pop('password', None)

otherdict.pop('password', None)

return selfdict == otherdict

return False

class UserManager(BaseManager):

obj_cls = User

def search(self, query, **kwargs):

"""Search users.

Args:

query (str): The query string to send to GitLab for the search.

all (bool): If True, return all the items, without pagination

**kwargs: Additional arguments to send to GitLab.

Returns:

list(User): A list of matching users.

Raises:

GitlabConnectionError: If the server cannot be reached.

GitlabListError: If the server fails to perform the request.

"""

url = self.obj_cls._url + '?search=' + query

return self.gitlab._raw_list(url, self.obj_cls, **kwargs)

def get_by_username(self, username, **kwargs):

"""Get a user by its username.

Args:

username (str): The name of the user.

**kwargs: Additional arguments to send to GitLab.

Returns:

User: The matching user.

Raises:

GitlabConnectionError: If the server cannot be reached.

GitlabGetError: If the server fails to perform the request.

"""

url = self.obj_cls._url + '?username=' + username

results = self.gitlab._raw_list(url, self.obj_cls, **kwargs)

assert len(results) in (0, 1)

try:

return results[0]

except IndexError:

raise GitlabGetError('no such user: ' + username)

class CurrentUserEmail(GitlabObject):

_url = '/user/emails'

canUpdate = False

shortPrintAttr = 'email'

requiredCreateAttrs = ['email']

class CurrentUserEmailManager(BaseManager):

obj_cls = CurrentUserEmail

class CurrentUserKey(GitlabObject):

_url = '/user/keys'

canUpdate = False

shortPrintAttr = 'title'

requiredCreateAttrs = ['title', 'key']

class CurrentUserKeyManager(BaseManager):

obj_cls = CurrentUserKey

class CurrentUser(GitlabObject):

_url = '/user'

canList = False

canCreate = False

canUpdate = False

canDelete = False

shortPrintAttr = 'username'

managers = (

('emails', 'CurrentUserEmailManager', [('user_id', 'id')]),

('keys', 'CurrentUserKeyManager', [('user_id', 'id')]),

)

class ApplicationSettings(GitlabObject):

_url = '/application/settings'

_id_in_update_url = False

getRequiresId = False

optionalUpdateAttrs = ['after_sign_out_path',

'container_registry_token_expire_delay',

'default_branch_protection',

'default_project_visibility',

'default_projects_limit',

'default_snippet_visibility',

'domain_blacklist',

'domain_blacklist_enabled',

'domain_whitelist',

'enabled_git_access_protocol',

'gravatar_enabled',

'home_page_url',

'max_attachment_size',

'repository_storage',

'restricted_signup_domains',

'restricted_visibility_levels',

'session_expire_delay',

'sign_in_text',

'signin_enabled',

'signup_enabled',

'twitter_sharing_enabled',

'user_oauth_applications']

canList = False

canCreate = False

canDelete = False

def _data_for_gitlab(self, extra_parameters={}, update=False,

as_json=True):

data = (super(ApplicationSettings, self)

._data_for_gitlab(extra_parameters, update=update,

as_json=False))

if not self.domain_whitelist:

data.pop('domain_whitelist', None)

return json.dumps(data)

class ApplicationSettingsManager(BaseManager):

obj_cls = ApplicationSettings

class BroadcastMessage(GitlabObject):

_url = '/broadcast_messages'

requiredCreateAttrs = ['message']

optionalCreateAttrs = ['starts_at', 'ends_at', 'color', 'font']

requiredUpdateAttrs = []

optionalUpdateAttrs = ['message', 'starts_at', 'ends_at', 'color', 'font']

class BroadcastMessageManager(BaseManager):

obj_cls = BroadcastMessage

class Key(GitlabObject):

_url = '/deploy_keys'

canGet = 'from_list'

canCreate = False

canUpdate = False

canDelete = False

def __init__(self, *args, **kwargs):

warnings.warn("`Key` is deprecated, use `DeployKey` instead",

DeprecationWarning)

super(Key, self).__init__(*args, **kwargs)

class KeyManager(BaseManager):

obj_cls = Key

class DeployKey(GitlabObject):

_url = '/deploy_keys'

canGet = 'from_list'

canCreate = False

canUpdate = False

canDelete = False

class DeployKeyManager(BaseManager):

obj_cls = DeployKey

class NotificationSettings(GitlabObject):

_url = '/notification_settings'

_id_in_update_url = False

getRequiresId = False

optionalUpdateAttrs = ['level',

'notification_email',

'new_note',

'new_issue',

'reopen_issue',

'close_issue',

'reassign_issue',

'new_merge_request',

'reopen_merge_request',

'close_merge_request',

'reassign_merge_request',

'merge_merge_request']

canList = False

canCreate = False

canDelete = False

class NotificationSettingsManager(BaseManager):

obj_cls = NotificationSettings

class Gitignore(GitlabObject):

_url = '/templates/gitignores'

canDelete = False

canUpdate = False

canCreate = False

idAttr = 'name'

class GitignoreManager(BaseManager):

obj_cls = Gitignore

class Gitlabciyml(GitlabObject):

_url = '/templates/gitlab_ci_ymls'

canDelete = False

canUpdate = False

canCreate = False

idAttr = 'name'

class GitlabciymlManager(BaseManager):

obj_cls = Gitlabciyml

class GroupIssue(GitlabObject):

_url = '/groups/%(group_id)s/issues'

canGet = 'from_list'

canCreate = False

canUpdate = False

canDelete = False

requiredUrlAttrs = ['group_id']

optionalListAttrs = ['state', 'labels', 'milestone', 'order_by', 'sort']

class GroupIssueManager(BaseManager):

obj_cls = GroupIssue

class GroupMember(GitlabObject):

_url = '/groups/%(group_id)s/members'

canGet = 'from_list'

requiredUrlAttrs = ['group_id']

requiredCreateAttrs = ['access_level', 'user_id']

optionalCreateAttrs = ['expires_at']

requiredUpdateAttrs = ['access_level']

optionalCreateAttrs = ['expires_at']

shortPrintAttr = 'username'

def _update(self, **kwargs):

self.user_id = self.id

super(GroupMember, self)._update(**kwargs)

class GroupMemberManager(BaseManager):

obj_cls = GroupMember

class GroupNotificationSettings(NotificationSettings):

_url = '/groups/%(group_id)s/notification_settings'

requiredUrlAttrs = ['group_id']

class GroupNotificationSettingsManager(BaseManager):

obj_cls = GroupNotificationSettings

class GroupAccessRequest(GitlabObject):

_url = '/groups/%(group_id)s/access_requests'

canGet = 'from_list'

canUpdate = False

def approve(self, access_level=gitlab.DEVELOPER_ACCESS, **kwargs):

"""Approve an access request.

Args:

access_level (int): The access level for the user.

Raises:

GitlabConnectionError: If the server cannot be reached.

GitlabUpdateError: If the server fails to perform the request.

"""

url = ('/groups/%(group_id)s/access_requests/%(id)s/approve' %

{'group_id': self.group_id, 'id': self.id})

data = {'access_level': access_level}

r = self.gitlab._raw_put(url, data=data, **kwargs)

raise_error_from_response(r, GitlabUpdateError, 201)

self._set_from_dict(r.json())

class GroupAccessRequestManager(BaseManager):

obj_cls = GroupAccessRequest

class Hook(GitlabObject):

_url = '/hooks'

canUpdate = False

requiredCreateAttrs = ['url']

shortPrintAttr = 'url'

class HookManager(BaseManager):

obj_cls = Hook

class Issue(GitlabObject):

_url = '/issues'

_constructorTypes = {'author': 'User', 'assignee': 'User',

'milestone': 'ProjectMilestone'}

canGet = 'from_list'

canDelete = False

canUpdate = False

canCreate = False

shortPrintAttr = 'title'

optionalListAttrs = ['state', 'labels', 'order_by', 'sort']

class IssueManager(BaseManager):

obj_cls = Issue

class License(GitlabObject):

_url = '/licenses'

canDelete = False

canUpdate = False

canCreate = False

idAttr = 'key'

optionalListAttrs = ['popular']

optionalGetAttrs = ['project', 'fullname']

class LicenseManager(BaseManager):

obj_cls = License

class Snippet(GitlabObject):

_url = '/snippets'

_constructorTypes = {'author': 'User'}

requiredCreateAttrs = ['title', 'file_name', 'content']

optionalCreateAttrs = ['lifetime', 'visibility_level']

optionalUpdateAttrs = ['title', 'file_name', 'content', 'visibility_level']

shortPrintAttr = 'title'

def raw(self, streamed=False, action=None, chunk_size=1024, **kwargs):

"""Return the raw content of a snippet.

Args:

streamed (bool): If True the data will be processed by chunks of

`chunk_size` and each chunk is passed to `action` for

treatment.

action (callable): Callable responsible of dealing with chunk of

data.

chunk_size (int): Size of each chunk.

Returns:

str: The snippet content.

Raises:

GitlabConnectionError: If the server cannot be reached.

GitlabGetError: If the server fails to perform the request.

"""

url = ("/snippets/%(snippet_id)s/raw" % {'snippet_id': self.id})

r = self.gitlab._raw_get(url, **kwargs)

raise_error_from_response(r, GitlabGetError)

return utils.response_content(r, streamed, action, chunk_size)

class SnippetManager(BaseManager):

obj_cls = Snippet

def public(self, **kwargs):

"""List all the public snippets.

Args:

all (bool): If True, return all the items, without pagination

**kwargs: Additional arguments to send to GitLab.

Returns:

list(gitlab.Gitlab.Snippet): The list of snippets.

"""

return self.gitlab._raw_list("/snippets/public", Snippet, **kwargs)

class Namespace(GitlabObject):

_url = '/namespaces'

canGet = 'from_list'

canUpdate = False

canDelete = False

canCreate = False

optionalListAttrs = ['search']

class NamespaceManager(BaseManager):

obj_cls = Namespace

class ProjectBoardList(GitlabObject):

_url = '/projects/%(project_id)s/boards/%(board_id)s/lists'

requiredUrlAttrs = ['project_id', 'board_id']

_constructorTypes = {'label': 'ProjectLabel'}

requiredCreateAttrs = ['label_id']

requiredUpdateAttrs = ['position']

class ProjectBoardListManager(BaseManager):

obj_cls = ProjectBoardList

class ProjectBoard(GitlabObject):

_url = '/projects/%(project_id)s/boards'

requiredUrlAttrs = ['project_id']

_constructorTypes = {'labels': 'ProjectBoardList'}

canGet = 'from_list'

canUpdate = False

canCreate = False

canDelete = False

managers = (

('lists', 'ProjectBoardListManager',

[('project_id', 'project_id'), ('board_id', 'id')]),

)

class ProjectBoardManager(BaseManager):

obj_cls = ProjectBoard

class ProjectBranch(GitlabObject):

_url = '/projects/%(project_id)s/repository/branches'

_constructorTypes = {'author': 'User', "committer": "User"}

idAttr = 'name'

canUpdate = False

requiredUrlAttrs = ['project_id']

requiredCreateAttrs = ['branch_name', 'ref']

def protect(self, protect=True, **kwargs):

"""Protects the branch."""

url = self._url % {'project_id': self.project_id}

action = 'protect' if protect else 'unprotect'

url = "%s/%s/%s" % (url, self.name, action)

r = self.gitlab._raw_put(url, data=None, content_type=None, **kwargs)

raise_error_from_response(r, GitlabProtectError)

if protect:

self.protected = protect

else:

del self.protected

def unprotect(self, **kwargs):

"""Unprotects the branch."""

self.protect(False, **kwargs)

class ProjectBranchManager(BaseManager):

obj_cls = ProjectBranch

class ProjectBuild(GitlabObject):

_url = '/projects/%(project_id)s/builds'

_constructorTypes = {'user': 'User',

'commit': 'ProjectCommit',

'runner': 'Runner'}

requiredUrlAttrs = ['project_id']

canDelete = False

canUpdate = False

canCreate = False

def cancel(self, **kwargs):

"""Cancel the build."""

url = '/projects/%s/builds/%s/cancel' % (self.project_id, self.id)

r = self.gitlab._raw_post(url)

raise_error_from_response(r, GitlabBuildCancelError, 201)

def retry(self, **kwargs):

"""Retry the build."""

url = '/projects/%s/builds/%s/retry' % (self.project_id, self.id)

r = self.gitlab._raw_post(url)

raise_error_from_response(r, GitlabBuildRetryError, 201)

def play(self, **kwargs):

"""Trigger a build explicitly."""

url = '/projects/%s/builds/%s/play' % (self.project_id, self.id)

r = self.gitlab._raw_post(url)

raise_error_from_response(r, GitlabBuildPlayError)

def erase(self, **kwargs):

"""Erase the build (remove build artifacts and trace)."""

url = '/projects/%s/builds/%s/erase' % (self.project_id, self.id)

r = self.gitlab._raw_post(url)

raise_error_from_response(r, GitlabBuildEraseError, 201)

def keep_artifacts(self, **kwargs):

"""Prevent artifacts from being delete when expiration is set.

Raises:

GitlabConnectionError: If the server cannot be reached.

GitlabCreateError: If the request failed.

"""

url = ('/projects/%s/builds/%s/artifacts/keep' %

(self.project_id, self.id))

r = self.gitlab._raw_post(url)

raise_error_from_response(r, GitlabGetError, 200)

def artifacts(self, streamed=False, action=None, chunk_size=1024,

**kwargs):

"""Get the build artifacts.

Args:

streamed (bool): If True the data will be processed by chunks of

`chunk_size` and each chunk is passed to `action` for

treatment.

action (callable): Callable responsible of dealing with chunk of

data.

chunk_size (int): Size of each chunk.

Returns:

str: The artifacts if `streamed` is False, None otherwise.

Raises:

GitlabConnectionError: If the server cannot be reached.

GitlabGetError: If the artifacts are not available.

"""

url = '/projects/%s/builds/%s/artifacts' % (self.project_id, self.id)

r = self.gitlab._raw_get(url, streamed=streamed, **kwargs)

raise_error_from_response(r, GitlabGetError, 200)

return utils.response_content(r, streamed, action, chunk_size)

def trace(self, streamed=False, action=None, chunk_size=1024, **kwargs):

"""Get the build trace.

Args:

streamed (bool): If True the data will be processed by chunks of

`chunk_size` and each chunk is passed to `action` for

treatment.

action (callable): Callable responsible of dealing with chunk of

data.

chunk_size (int): Size of each chunk.

Returns:

str: The trace.

Raises:

GitlabConnectionError: If the server cannot be reached.

GitlabGetError: If the trace is not available.

"""

url = '/projects/%s/builds/%s/trace' % (self.project_id, self.id)

r = self.gitlab._raw_get(url, streamed=streamed, **kwargs)

raise_error_from_response(r, GitlabGetError, 200)

return utils.response_content(r, streamed, action, chunk_size)

class ProjectBuildManager(BaseManager):

obj_cls = ProjectBuild

class ProjectCommitStatus(GitlabObject):

_url = '/projects/%(project_id)s/repository/commits/%(commit_id)s/statuses'

_create_url = '/projects/%(project_id)s/statuses/%(commit_id)s'

canUpdate = False

canDelete = False

requiredUrlAttrs = ['project_id', 'commit_id']

optionalGetAttrs = ['ref_name', 'stage', 'name', 'all']

requiredCreateAttrs = ['state']

optionalCreateAttrs = ['description', 'name', 'context', 'ref',

'target_url']

class ProjectCommitStatusManager(BaseManager):

obj_cls = ProjectCommitStatus

class ProjectCommitComment(GitlabObject):

_url = '/projects/%(project_id)s/repository/commits/%(commit_id)s/comments'

canUpdate = False

canGet = False

canDelete = False

requiredUrlAttrs = ['project_id', 'commit_id']

requiredCreateAttrs = ['note']

optionalCreateAttrs = ['path', 'line', 'line_type']

class ProjectCommitCommentManager(BaseManager):

obj_cls = ProjectCommitComment

class ProjectCommit(GitlabObject):

_url = '/projects/%(project_id)s/repository/commits'

canDelete = False

canUpdate = False

requiredUrlAttrs = ['project_id']

requiredCreateAttrs = ['branch_name', 'commit_message', 'actions']

optionalCreateAttrs = ['author_email', 'author_name']

shortPrintAttr = 'title'

managers = (

('comments', 'ProjectCommitCommentManager',

[('project_id', 'project_id'), ('commit_id', 'id')]),

('statuses', 'ProjectCommitStatusManager',

[('project_id', 'project_id'), ('commit_id', 'id')]),

)

def diff(self, **kwargs):

"""Generate the commit diff."""

url = ('/projects/%(project_id)s/repository/commits/%(commit_id)s/diff'

% {'project_id': self.project_id, 'commit_id': self.id})

r = self.gitlab._raw_get(url, **kwargs)

raise_error_from_response(r, GitlabGetError)

return r.json()

def blob(self, filepath, streamed=False, action=None, chunk_size=1024,

**kwargs):

"""Generate the content of a file for this commit.

Args:

filepath (str): Path of the file to request.

streamed (bool): If True the data will be processed by chunks of

`chunk_size` and each chunk is passed to `action` for

treatment.

action (callable): Callable responsible of dealing with chunk of

data.

chunk_size (int): Size of each chunk.

Returns:

str: The content of the file

Raises:

GitlabConnectionError: If the server cannot be reached.

GitlabGetError: If the server fails to perform the request.

"""

url = ('/projects/%(project_id)s/repository/blobs/%(commit_id)s' %

{'project_id': self.project_id, 'commit_id': self.id})

url += '?filepath=%s' % filepath

r = self.gitlab._raw_get(url, streamed=streamed, **kwargs)

raise_error_from_response(r, GitlabGetError)

return utils.response_content(r, streamed, action, chunk_size)

def builds(self, **kwargs):

"""List the build for this commit.

Returns:

list(ProjectBuild): A list of builds.

Raises:

GitlabConnectionError: If the server cannot be reached.

GitlabListError: If the server fails to perform the request.

"""

url = '/projects/%s/repository/commits/%s/builds' % (self.project_id,

self.id)

return self.gitlab._raw_list(url, ProjectBuild, **kwargs)

def cherry_pick(self, branch, **kwargs):

"""Cherry-pick a commit into a branch.

Args:

branch (str): Name of target branch.

Raises:

GitlabCherryPickError: If the cherry pick could not be applied.

"""

url = ('/projects/%s/repository/commits/%s/cherry_pick' %

(self.project_id, self.id))

r = self.gitlab._raw_post(url, data={'project_id': self.project_id,

'branch': branch}, **kwargs)

errors = {400: GitlabCherryPickError}

raise_error_from_response(r, errors, expected_code=201)

class ProjectCommitManager(BaseManager):

obj_cls = ProjectCommit

class ProjectEnvironment(GitlabObject):

_url = '/projects/%(project_id)s/environments'

canGet = 'from_list'

requiredUrlAttrs = ['project_id']

requiredCreateAttrs = ['name']

optionalCreateAttrs = ['external_url']

optionalUpdateAttrs = ['name', 'external_url']

class ProjectEnvironmentManager(BaseManager):

obj_cls = ProjectEnvironment

class ProjectKey(GitlabObject):

_url = '/projects/%(project_id)s/keys'

canUpdate = False

requiredUrlAttrs = ['project_id']

requiredCreateAttrs = ['title', 'key']

class ProjectKeyManager(BaseManager):

obj_cls = ProjectKey

def enable(self, key_id):

"""Enable a deploy key for a project."""

url = '/projects/%s/deploy_keys/%s/enable' % (self.parent.id, key_id)

r = self.gitlab._raw_post(url)

raise_error_from_response(r, GitlabProjectDeployKeyError, 201)

def disable(self, key_id):

"""Disable a deploy key for a project."""

url = '/projects/%s/deploy_keys/%s/disable' % (self.parent.id, key_id)

r = self.gitlab._raw_delete(url)

raise_error_from_response(r, GitlabProjectDeployKeyError, 200)

class ProjectEvent(GitlabObject):

_url = '/projects/%(project_id)s/events'

canGet = 'from_list'

canDelete = False

canUpdate = False

canCreate = False

requiredUrlAttrs = ['project_id']

shortPrintAttr = 'target_title'

class ProjectEventManager(BaseManager):

obj_cls = ProjectEvent

class ProjectFork(GitlabObject):

_url = '/projects/fork/%(project_id)s'

canUpdate = False

canDelete = False

canList = False

canGet = False

requiredUrlAttrs = ['project_id']

optionalCreateAttrs = ['namespace']

class ProjectForkManager(BaseManager):

obj_cls = ProjectFork

class ProjectHook(GitlabObject):

_url = '/projects/%(project_id)s/hooks'

requiredUrlAttrs = ['project_id']

requiredCreateAttrs = ['url']

optionalCreateAttrs = ['push_events', 'issues_events', 'note_events',

'merge_requests_events', 'tag_push_events',

'build_events', 'enable_ssl_verification', 'token',

'pipeline_events']

shortPrintAttr = 'url'

class ProjectHookManager(BaseManager):

obj_cls = ProjectHook

class ProjectIssueNote(GitlabObject):

_url = '/projects/%(project_id)s/issues/%(issue_id)s/notes'

_constructorTypes = {'author': 'User'}

canDelete = False

requiredUrlAttrs = ['project_id', 'issue_id']

requiredCreateAttrs = ['body']

optionalCreateAttrs = ['created_at']

# file attachment settings (see #56)

description_attr = "body"

project_id_attr = "project_id"

class ProjectIssueNoteManager(BaseManager):

obj_cls = ProjectIssueNote

class ProjectIssue(GitlabObject):

_url = '/projects/%(project_id)s/issues/'

_constructorTypes = {'author': 'User', 'assignee': 'User',

'milestone': 'ProjectMilestone'}

optionalListAttrs = ['state', 'labels', 'milestone', 'iid', 'order_by',

'sort']

requiredUrlAttrs = ['project_id']

requiredCreateAttrs = ['title']

optionalCreateAttrs = ['description', 'assignee_id', 'milestone_id',

'labels', 'created_at', 'due_date']

optionalUpdateAttrs = ['title', 'description', 'assignee_id',

'milestone_id', 'labels', 'created_at',

'updated_at', 'state_event', 'due_date']

shortPrintAttr = 'title'

managers = (

('notes', 'ProjectIssueNoteManager',

[('project_id', 'project_id'), ('issue_id', 'id')]),

)

# file attachment settings (see #56)

description_attr = "description"

project_id_attr = "project_id"

def subscribe(self, **kwargs):

"""Subscribe to an issue.

Raises:

GitlabConnectionError: If the server cannot be reached.

GitlabSubscribeError: If the subscription cannot be done

"""

url = ('/projects/%(project_id)s/issues/%(issue_id)s/subscription' %

{'project_id': self.project_id, 'issue_id': self.id})

r = self.gitlab._raw_post(url, **kwargs)

raise_error_from_response(r, GitlabSubscribeError)

self._set_from_dict(r.json())

def unsubscribe(self, **kwargs):

"""Unsubscribe an issue.

Raises:

GitlabConnectionError: If the server cannot be reached.

GitlabUnsubscribeError: If the unsubscription cannot be done

"""

url = ('/projects/%(project_id)s/issues/%(issue_id)s/subscription' %

{'project_id': self.project_id, 'issue_id': self.id})

r = self.gitlab._raw_delete(url, **kwargs)

raise_error_from_response(r, GitlabUnsubscribeError)

self._set_from_dict(r.json())

def move(self, to_project_id, **kwargs):

"""Move the issue to another project.

Raises:

GitlabConnectionError: If the server cannot be reached.

"""

url = ('/projects/%(project_id)s/issues/%(issue_id)s/move' %

{'project_id': self.project_id, 'issue_id': self.id})

data = {'to_project_id': to_project_id}

data.update(**kwargs)

r = self.gitlab._raw_post(url, data=data)

raise_error_from_response(r, GitlabUpdateError, 201)

self._set_from_dict(r.json())

def todo(self, **kwargs):

"""Create a todo for the issue.

Raises:

GitlabConnectionError: If the server cannot be reached.

"""

url = ('/projects/%(project_id)s/issues/%(issue_id)s/todo' %

{'project_id': self.project_id, 'issue_id': self.id})

r = self.gitlab._raw_post(url, **kwargs)

raise_error_from_response(r, GitlabTodoError, [201, 304])

def time_stats(self, **kwargs):

"""Get time stats for the issue.