@@ -499,2131 +499,2131 @@ class ApiController(JSONRPCController):
}
"""
if isinstance(userid, Optional):
userid = apiuser.user_id
user = get_user_or_error(userid)
ips = UserIpMap.query().filter(UserIpMap.user == user).all()
return dict(
server_ip_addr=self.ip_addr,
user_ips=ips
)
# alias for old
show_ip = get_ip
@HasPermissionAnyDecorator('hg.admin')
def get_server_info(self, apiuser):
return server info, including Kallithea version and installed packages
:param apiuser: filled automatically from apikey
:type apiuser: AuthUser
OUTPUT::
id : <id_given_in_input>
result : {
'modules': [<module name>,...]
'py_version': <python version>,
'platform': <platform type>,
'kallithea_version': <kallithea version>
error : null
return Setting.get_server_info()
def get_user(self, apiuser, userid=Optional(OAttr('apiuser'))):
Gets a user by username or user_id, Returns empty result if user is
not found. If userid param is skipped it is set to id of user who is
calling this method. This command can be executed only using api_key
belonging to user with admin rights, or regular users that cannot
specify different userid than theirs
:param userid: user to get data for
:type userid: Optional(str or int)
result: None if user does not exist or
{
"user_id" : "<user_id>",
"api_key" : "<api_key>",
"api_keys": "[<list of all API keys including additional ones>]"
"username" : "<username>",
"firstname": "<firstname>",
"lastname" : "<lastname>",
"email" : "<email>",
"emails": "[<list of all emails including additional ones>]",
"ip_addresses": "[<ip_address_for_user>,...]",
"active" : "<bool: user active>",
"admin" : "<bool: user is admin>",
"extern_name" : "<extern_name>",
"extern_type" : "<extern type>
"last_login": "<last_login>",
"permissions": {
"global": ["hg.create.repository",
"repository.read",
"hg.register.manual_activate"],
"repositories": {"repo1": "repository.none"},
"repositories_groups": {"Group1": "group.read"}
},
error: null
if not HasPermissionAnyApi('hg.admin')(user=apiuser):
# make sure normal user does not pass someone else userid,
# he is not allowed to do that
if not isinstance(userid, Optional) and userid != apiuser.user_id:
raise JSONRPCError(
'userid is not the same as your user'
data = user.get_api_data()
data['permissions'] = AuthUser(user_id=user.user_id).permissions
return data
def get_users(self, apiuser):
Lists all existing users. This command can be executed only using api_key
belonging to user with admin rights.
result: [<user_object>, ...]
result = []
users_list = User.query().order_by(User.username) \
.filter(User.username != User.DEFAULT_USER) \
.all()
for user in users_list:
result.append(user.get_api_data())
return result
def create_user(self, apiuser, username, email, password=Optional(''),
firstname=Optional(''), lastname=Optional(''),
active=Optional(True), admin=Optional(False),
extern_name=Optional(EXTERN_TYPE_INTERNAL),
extern_type=Optional(EXTERN_TYPE_INTERNAL)):
Creates new user. Returns new user object. This command can
be executed only using api_key belonging to user with admin rights.
:param username: new username
:type username: str or int
:param email: email
:type email: str
:param password: password
:type password: Optional(str)
:param firstname: firstname
:type firstname: Optional(str)
:param lastname: lastname
:type lastname: Optional(str)
:param active: active
:type active: Optional(bool)
:param admin: admin
:type admin: Optional(bool)
:param extern_name: name of extern
:type extern_name: Optional(str)
:param extern_type: extern_type
:type extern_type: Optional(str)
result: {
"msg" : "created new user `<username>`",
"user": <user_obj>
ERROR OUTPUT::
result : null
error : {
"user `<username>` already exist"
or
"email `<email>` already exist"
"failed to create user `<username>`"
if User.get_by_username(username):
raise JSONRPCError("user `%s` already exist" % (username,))
if User.get_by_email(email):
raise JSONRPCError("email `%s` already exist" % (email,))
if Optional.extract(extern_name):
# generate temporary password if user is external
password = PasswordGenerator().gen_password(length=8)
try:
user = UserModel().create_or_update(
username=Optional.extract(username),
password=Optional.extract(password),
email=Optional.extract(email),
firstname=Optional.extract(firstname),
lastname=Optional.extract(lastname),
active=Optional.extract(active),
admin=Optional.extract(admin),
extern_type=Optional.extract(extern_type),
extern_name=Optional.extract(extern_name)
Session().commit()
msg='created new user `%s`' % username,
user=user.get_api_data()
except Exception:
log.error(traceback.format_exc())
raise JSONRPCError('failed to create user `%s`' % (username,))
def update_user(self, apiuser, userid, username=Optional(None),
email=Optional(None), password=Optional(None),
firstname=Optional(None), lastname=Optional(None),
active=Optional(None), admin=Optional(None),
extern_type=Optional(None), extern_name=Optional(None)):
updates given user if such user exists. This command can
:param userid: userid to update
:type userid: str or int
:param extern_name:
:param extern_type:
"msg" : "updated user ID:<userid> <username>",
"user": <user_object>,
"failed to update user `<username>`"
# only non optional arguments will be stored in updates
updates = {}
store_update(updates, username, 'username')
store_update(updates, password, 'password')
store_update(updates, email, 'email')
store_update(updates, firstname, 'name')
store_update(updates, lastname, 'lastname')
store_update(updates, active, 'active')
store_update(updates, admin, 'admin')
store_update(updates, extern_name, 'extern_name')
store_update(updates, extern_type, 'extern_type')
user = UserModel().update_user(user, **updates)
msg='updated user ID:%s %s' % (user.user_id, user.username),
except DefaultUserException:
raise JSONRPCError('editing default user is forbidden')
raise JSONRPCError('failed to update user `%s`' % (userid,))
def delete_user(self, apiuser, userid):
deletes given user if such user exists. This command can
:param userid: user to delete
"msg" : "deleted user ID:<userid> <username>",
"user": null
"failed to delete user ID:<userid> <username>"
UserModel().delete(userid)
msg='deleted user ID:%s %s' % (user.user_id, user.username),
user=None
raise JSONRPCError('failed to delete user ID:%s %s'
% (user.user_id, user.username))
# permission check inside
def get_user_group(self, apiuser, usergroupid):
Gets an existing user group. This command can be executed only using api_key
belonging to user with admin rights or user who has at least
read access to user group.
:param usergroupid: id of user_group to edit
:type usergroupid: str or int
result : None if group not exist
"users_group_id" : "<id>",
"group_name" : "<groupname>",
"active": "<bool>",
"members" : [<user_obj>,...]
user_group = get_user_group_or_error(usergroupid)
# check if we have at least read permission for this user group !
_perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
if not HasUserGroupPermissionAny(*_perms)(
user=apiuser, user_group_name=user_group.users_group_name):
raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
data = user_group.get_api_data()
def get_user_groups(self, apiuser):
Lists all existing user groups. This command can be executed only using
api_key belonging to user with admin rights or user who has at least
result : [<user_group_obj>,...]
extras = {'user': apiuser}
for user_group in UserGroupList(UserGroup.get_all(),
perm_set=_perms, extra_kwargs=extras):
result.append(user_group.get_api_data())
@HasPermissionAnyDecorator('hg.admin', 'hg.usergroup.create.true')
def create_user_group(self, apiuser, group_name, description=Optional(''),
owner=Optional(OAttr('apiuser')), active=Optional(True)):
Creates new user group. This command can be executed only using api_key
belonging to user with admin rights or an user who has create user group
permission
:param group_name: name of new user group
:type group_name: str
:param description: group description
:type description: str
:param owner: owner of group. If not passed apiuser is the owner
:type owner: Optional(str or int)
:param active: group is active
"msg": "created new user group `<groupname>`",
"user_group": <user_group_object>
"user group `<group name>` already exist"
"failed to create group `<group name>`"
if UserGroupModel().get_by_name(group_name):
raise JSONRPCError("user group `%s` already exist" % (group_name,))
if isinstance(owner, Optional):
owner = apiuser.user_id
owner = get_user_or_error(owner)
active = Optional.extract(active)
description = Optional.extract(description)
ug = UserGroupModel().create(name=group_name, description=description,
owner=owner, active=active)
msg='created new user group `%s`' % group_name,
user_group=ug.get_api_data()
raise JSONRPCError('failed to create group `%s`' % (group_name,))
def update_user_group(self, apiuser, usergroupid, group_name=Optional(''),
description=Optional(''), owner=Optional(None),
active=Optional(True)):
Updates given usergroup. This command can be executed only using api_key
belonging to user with admin rights or an admin of given user group
:param usergroupid: id of user group to update
:param owner: owner of group.
"msg": 'updated user group ID:<user group id> <user group name>',
"failed to update user group `<user group name>`"
# check if we have admin permission for this user group !
_perms = ('usergroup.admin',)
if not isinstance(owner, Optional):
store_update(updates, group_name, 'users_group_name')
store_update(updates, description, 'user_group_description')
store_update(updates, owner, 'user')
store_update(updates, active, 'users_group_active')
UserGroupModel().update(user_group, updates)
msg='updated user group ID:%s %s' % (user_group.users_group_id,
user_group.users_group_name),
user_group=user_group.get_api_data()
raise JSONRPCError('failed to update user group `%s`' % (usergroupid,))
def delete_user_group(self, apiuser, usergroupid):
Delete given user group by user group id or name.
This command can be executed only using api_key
:param usergroupid:
:type usergroupid: int
"msg": "deleted user group ID:<user_group_id> <user_group_name>"
"failed to delete user group ID:<user_group_id> <user_group_name>"
"RepoGroup assigned to <repo_groups_list>"
UserGroupModel().delete(user_group)
msg='deleted user group ID:%s %s' %
(user_group.users_group_id, user_group.users_group_name),
user_group=None
except UserGroupsAssignedException as e:
raise JSONRPCError(str(e))
raise JSONRPCError('failed to delete user group ID:%s %s' %
(user_group.users_group_id,
user_group.users_group_name)
def add_user_to_user_group(self, apiuser, usergroupid, userid):
Adds a user to a user group. If user exists in that group success will be
`false`. This command can be executed only using api_key
:param userid:
:type userid: int
"success": True|False # depends on if member is in group
"msg": "added member `<username>` to user group `<groupname>` |
User is already in that group"
"failed to add member to user group `<user_group_name>`"
ugm = UserGroupModel().add_user_to_group(user_group, user)
success = True if ugm != True else False
msg = 'added member `%s` to user group `%s`' % (
user.username, user_group.users_group_name
msg = msg if success else 'User is already in that group'
success=success,
msg=msg
'failed to add member to user group `%s`' % (
user_group.users_group_name,
def remove_user_from_user_group(self, apiuser, usergroupid, userid):
Removes a user from a user group. If user is not in given group success will
be `false`. This command can be executed only
using api_key belonging to user with admin rights or an admin of given user group
"success": True|False, # depends on if member is in group
"msg": "removed member <username> from user group <groupname> |
User wasn't in group"
success = UserGroupModel().remove_user_from_group(user_group, user)
msg = 'removed member `%s` from user group `%s`' % (
msg = msg if success else "User wasn't in group"
return dict(success=success, msg=msg)
'failed to remove member from user group `%s`' % (
def get_repo(self, apiuser, repoid):
Gets an existing repository by it's name or repository_id. Members will return
either users_group or user associated to that repository. This command can be
executed only using api_key belonging to user with admin
rights or regular user that have at least read access to repository.
:param repoid: repository name or repository id
:type repoid: str or int
"repo_id" : "<repo_id>",
"repo_name" : "<reponame>"
"repo_type" : "<repo_type>",
"clone_uri" : "<clone_uri>",
"enable_downloads": "<bool>",
"enable_locking": "<bool>",
"enable_statistics": "<bool>",
"private": "<bool>",
"created_on" : "<date_time_created>",
"description" : "<description>",
"landing_rev": "<landing_rev>",
"last_changeset": {
"author": "<full_author>",
"date": "<date_time_of_commit>",
"message": "<commit_message>",
"raw_id": "<raw_id>",
"revision": "<numeric_revision>",
"short_id": "<short_id>"
"owner": "<repo_owner>",
"fork_of": "<name_of_fork_parent>",
"members" : [
"name": "<username>",
"type" : "user",
"permission" : "repository.(read|write|admin)"
…
"name": "<usergroup name>",
"type" : "user_group",
"permission" : "usergroup.(read|write|admin)"
]
"followers": [<user_obj>, ...]
repo = get_repo_or_error(repoid)
# check if we have admin permission for this repo !
perms = ('repository.admin', 'repository.write', 'repository.read')
if not HasRepoPermissionAnyApi(*perms)(user=apiuser, repo_name=repo.repo_name):
raise JSONRPCError('repository `%s` does not exist' % (repoid,))
members = []
followers = []
for user in repo.repo_to_perm:
perm = user.permission.permission_name
user = user.user
user_data = {
'name': user.username,
'type': "user",
'permission': perm
members.append(user_data)
for user_group in repo.users_group_to_perm:
perm = user_group.permission.permission_name
user_group = user_group.users_group
user_group_data = {
'name': user_group.users_group_name,
'type': "user_group",
members.append(user_group_data)
for user in repo.followers:
followers.append(user.user.get_api_data())
data = repo.get_api_data()
data['members'] = members
data['followers'] = followers
def get_repos(self, apiuser):
Lists all existing repositories. This command can be executed only using
api_key belonging to user with admin rights or regular user that have
admin, write or read access to repository.
result: [
"private": : "<bool>",
"created_on" : "<datetimecreated>",
repos = RepoModel().get_all_user_repos(user=apiuser)
else:
repos = Repository.get_all()
for repo in repos:
result.append(repo.get_api_data())
def get_repo_nodes(self, apiuser, repoid, revision, root_path,
ret_type=Optional('all')):
returns a list of nodes and it's children in a flat list for a given path
at given revision. It's possible to specify ret_type to show only `files` or
`dirs`. This command can be executed only using api_key belonging to
user with admin rights or regular user that have at least read access to repository.
:param revision: revision for which listing should be done
:type revision: str
:param root_path: path from which start displaying
:type root_path: str
:param ret_type: return type 'all|files|dirs' nodes
:type ret_type: Optional(str)
"name" : "<name>"
"type" : "<type>",
ret_type = Optional.extract(ret_type)
_map = {}
_d, _f = ScmModel().get_nodes(repo, revision, root_path,
flat=False)
_map = {
'all': _d + _f,
'files': _f,
'dirs': _d,
return _map[ret_type]
except KeyError:
raise JSONRPCError('ret_type must be one of %s'
% (','.join(_map.keys())))
'failed to get repo: `%s` nodes' % repo.repo_name
@HasPermissionAnyDecorator('hg.admin', 'hg.create.repository')
def create_repo(self, apiuser, repo_name, owner=Optional(OAttr('apiuser')),
repo_type=Optional('hg'), description=Optional(''),
private=Optional(False), clone_uri=Optional(None),
landing_rev=Optional('rev:tip'),
enable_statistics=Optional(False),
enable_locking=Optional(False),
enable_downloads=Optional(False),
copy_permissions=Optional(False)):
Creates a repository. If repository name contains "/", all needed repository
groups will be created. For example "foo/bar/baz" will create groups
"foo", "bar" (with "foo" as parent), and create "baz" repository with
"bar" as group. This command can be executed only using api_key
belonging to user with admin rights or regular user that have create
repository permission. Regular users cannot specify owner parameter
:param repo_name: repository name
:type repo_name: str
:param owner: user_id or username
:type owner: Optional(str)
:param repo_type: 'hg' or 'git'
:type repo_type: Optional(str)
:param description: repository description
:type description: Optional(str)
:param private:
:type private: bool
:param clone_uri:
:type clone_uri: str
:param landing_rev: <rev_type>:<rev>
:type landing_rev: str
:param enable_locking:
:type enable_locking: bool
:param enable_downloads:
:type enable_downloads: bool
:param enable_statistics:
:type enable_statistics: bool
:param copy_permissions: Copy permission from group that repository is
being created.
:type copy_permissions: bool
"msg": "Created new repository `<reponame>`",
"success": true,
"task": "<celery task id or None if done sync>"
'failed to create repository `<repo_name>`
# forbid setting owner for non-admins
'Only Kallithea admin can specify `owner` param'
if RepoModel().get_by_repo_name(repo_name):
raise JSONRPCError("repo `%s` already exist" % repo_name)
defs = Setting.get_default_repo_settings(strip_prefix=True)
if isinstance(private, Optional):
private = defs.get('repo_private') or Optional.extract(private)
if isinstance(repo_type, Optional):
repo_type = defs.get('repo_type')
if isinstance(enable_statistics, Optional):
enable_statistics = defs.get('repo_enable_statistics')
if isinstance(enable_locking, Optional):
enable_locking = defs.get('repo_enable_locking')
if isinstance(enable_downloads, Optional):
enable_downloads = defs.get('repo_enable_downloads')
clone_uri = Optional.extract(clone_uri)
landing_rev = Optional.extract(landing_rev)
copy_permissions = Optional.extract(copy_permissions)
repo_name_cleaned = repo_name.split('/')[-1]
# create structure of groups and return the last group
repo_group = map_groups(repo_name)
data = dict(
repo_name=repo_name_cleaned,
repo_name_full=repo_name,
repo_type=repo_type,
repo_description=description,
owner=owner,
repo_private=private,
clone_uri=clone_uri,
repo_group=repo_group,
repo_landing_rev=landing_rev,
enable_statistics=enable_statistics,
enable_locking=enable_locking,
enable_downloads=enable_downloads,
repo_copy_permissions=copy_permissions,
task = RepoModel().create(form_data=data, cur_user=owner)
from celery.result import BaseAsyncResult
task_id = None
if isinstance(task, BaseAsyncResult):
task_id = task.task_id
# no commit, it's done in RepoModel, or async via celery
msg="Created new repository `%s`" % (repo_name,),
success=True, # cannot return the repo data here since fork
# can be done async
task=task_id
'failed to create repository `%s`' % (repo_name,))
def update_repo(self, apiuser, repoid, name=Optional(None),
owner=Optional(OAttr('apiuser')),
group=Optional(None),
description=Optional(''), private=Optional(False),
clone_uri=Optional(None), landing_rev=Optional('rev:tip'),
enable_downloads=Optional(False)):
Updates repo
:param name:
:param owner:
:param group:
:param description:
:param landing_rev:
if not HasRepoPermissionAnyApi('repository.admin')(user=apiuser,
repo_name=repo.repo_name):
if (name != repo.repo_name and
not HasPermissionAnyApi('hg.create.repository')(user=apiuser)
):
raise JSONRPCError('no permission to create (or move) repositories')
repo_group = group
if not isinstance(repo_group, Optional):
repo_group = get_repo_group_or_error(repo_group)
repo_group = repo_group.group_id
store_update(updates, name, 'repo_name')
store_update(updates, repo_group, 'repo_group')
store_update(updates, description, 'repo_description')
store_update(updates, private, 'repo_private')
store_update(updates, clone_uri, 'clone_uri')
store_update(updates, landing_rev, 'repo_landing_rev')
store_update(updates, enable_statistics, 'repo_enable_statistics')
store_update(updates, enable_locking, 'repo_enable_locking')
store_update(updates, enable_downloads, 'repo_enable_downloads')
RepoModel().update(repo, **updates)
msg='updated repo ID:%s %s' % (repo.repo_id, repo.repo_name),
repository=repo.get_api_data()
raise JSONRPCError('failed to update repo `%s`' % repoid)
@HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')
def fork_repo(self, apiuser, repoid, fork_name,
description=Optional(''), copy_permissions=Optional(False),
private=Optional(False), landing_rev=Optional('rev:tip')):
Creates a fork of given repo. In case of using celery this will
immediately return success message, while fork is going to be created
asynchronous. This command can be executed only using api_key belonging to
user with admin rights or regular user that have fork permission, and at least
read access to forking repository. Regular users cannot specify owner parameter.
:param fork_name:
:param copy_permissions:
INPUT::
id : <id_for_response>
api_key : "<api_key>"
args: {
"repoid" : "<reponame or repo_id>",
"fork_name": "<forkname>",
"owner": "<username or user_id = Optional(=apiuser)>",
"description": "<description>",
"copy_permissions": "<bool>",
"landing_rev": "<landing_rev>"
"msg": "Created fork of `<reponame>` as `<forkname>`",
repo_name = repo.repo_name
_repo = RepoModel().get_by_repo_name(fork_name)
if _repo:
type_ = 'fork' if _repo.fork else 'repo'
raise JSONRPCError("%s `%s` already exist" % (type_, fork_name))
if HasPermissionAnyApi('hg.admin')(user=apiuser):
pass
elif HasRepoPermissionAnyApi('repository.admin',
'repository.write',
'repository.read')(user=apiuser,
if not HasPermissionAnyApi('hg.create.repository')(user=apiuser):
raise JSONRPCError('no permission to create repositories')
group = map_groups(fork_name)
fork_base_name = fork_name.rsplit('/', 1)[-1]
form_data = dict(
repo_name=fork_base_name,
repo_name_full=fork_name,
repo_group=group,
repo_type=repo.repo_type,
description=Optional.extract(description),
private=Optional.extract(private),
copy_permissions=Optional.extract(copy_permissions),
landing_rev=Optional.extract(landing_rev),
update_after_clone=False,
fork_parent_id=repo.repo_id,
task = RepoModel().create_fork(form_data, cur_user=owner)
msg='Created fork of `%s` as `%s`' % (repo.repo_name,
fork_name),
'failed to fork repository `%s` as `%s`' % (repo_name,
fork_name)
def delete_repo(self, apiuser, repoid, forks=Optional('')):
Deletes a repository. This command can be executed only using api_key belonging
to user with admin rights or regular user that have admin access to repository.
When `forks` param is set it's possible to detach or delete forks of deleting
repository
:param forks: `detach` or `delete`, what do do with attached forks for repo
:type forks: Optional(str)
"msg": "Deleted repository `<reponame>`",
"success": true
handle_forks = Optional.extract(forks)
_forks_msg = ''
_forks = [f for f in repo.forks]
if handle_forks == 'detach':
_forks_msg = ' ' + 'Detached %s forks' % len(_forks)
elif handle_forks == 'delete':
_forks_msg = ' ' + 'Deleted %s forks' % len(_forks)
elif _forks:
'Cannot delete `%s` it still contains attached forks' %
(repo.repo_name,)
RepoModel().delete(repo, forks=forks)
msg='Deleted repository `%s`%s' % (repo.repo_name, _forks_msg),
success=True
'failed to delete repository `%s`' % (repo.repo_name,)
def grant_user_permission(self, apiuser, repoid, userid, perm):
Grant permission for user on given repository, or update existing one
if found. This command can be executed only using api_key belonging to user
with admin rights.
:param perm: (repository.(none|read|write|admin))
:type perm: str
"msg" : "Granted perm: `<perm>` for user: `<username>` in repo: `<reponame>`",
perm = get_perm_or_error(perm)
RepoModel().grant_user_permission(repo=repo, user=user, perm=perm)
msg='Granted perm: `%s` for user: `%s` in repo: `%s`' % (
perm.permission_name, user.username, repo.repo_name
),
'failed to edit permission for user: `%s` in repo: `%s`' % (
userid, repoid
def revoke_user_permission(self, apiuser, repoid, userid):
Revoke permission for user on given repository. This command can be executed
only using api_key belonging to user with admin rights.
"msg" : "Revoked perm for user: `<username>` in repo: `<reponame>`",
RepoModel().revoke_user_permission(repo=repo, user=user)
msg='Revoked perm for user: `%s` in repo: `%s`' % (
user.username, repo.repo_name
def grant_user_group_permission(self, apiuser, repoid, usergroupid, perm):
Grant permission for user group on given repository, or update
existing one if found. This command can be executed only using
api_key belonging to user with admin rights.
:param usergroupid: id of usergroup
"msg" : "Granted perm: `<perm>` for group: `<usersgroupname>` in repo: `<reponame>`",
"failed to edit permission for user group: `<usergroup>` in repo `<repo>`'
_perms = ('repository.admin',)
if not HasRepoPermissionAnyApi(*_perms)(
user=apiuser, repo_name=repo.repo_name):
RepoModel().grant_user_group_permission(
repo=repo, group_name=user_group, perm=perm)
msg='Granted perm: `%s` for user group: `%s` in '
'repo: `%s`' % (
perm.permission_name, user_group.users_group_name,
repo.repo_name
'failed to edit permission for user group: `%s` in '
usergroupid, repo.repo_name
def revoke_user_group_permission(self, apiuser, repoid, usergroupid):
Revoke permission for user group on given repository. This command can be
executed only using api_key belonging to user with admin rights.
"msg" : "Revoked perm for group: `<usersgroupname>` in repo: `<reponame>`",
RepoModel().revoke_user_group_permission(
repo=repo, group_name=user_group)
msg='Revoked perm for user group: `%s` in repo: `%s`' % (
user_group.users_group_name, repo.repo_name
def get_repo_group(self, apiuser, repogroupid):
Returns given repo group together with permissions, and repositories
inside the group
:param repogroupid: id/name of repository group
:type repogroupid: str or int
repo_group = get_repo_group_or_error(repogroupid)
for user in repo_group.repo_group_to_perm:
for user_group in repo_group.users_group_to_perm:
data = repo_group.get_api_data()
data["members"] = members
def get_repo_groups(self, apiuser):
Returns all repository groups
for repo_group in RepoGroupModel().get_all():
for repo_group in RepoGroup.get_all():
result.append(repo_group.get_api_data())
def create_repo_group(self, apiuser, group_name, description=Optional(''),
parent=Optional(None),
Creates a repository group. This command can be executed only using
:param group_name:
:type group_name:
:type description:
:type owner:
:param parent:
:type parent:
:type copy_permissions:
"msg": "created new repo group `<repo_group_name>`"
"repo_group": <repogroup_object>
failed to create repo group `<repogroupid>`
if RepoGroup.get_by_group_name(group_name):
raise JSONRPCError("repo group `%s` already exist" % (group_name,))
group_description = Optional.extract(description)
parent_group = Optional.extract(parent)
if not isinstance(parent, Optional):
parent_group = get_repo_group_or_error(parent_group)
repo_group = RepoGroupModel().create(
group_name=group_name,
group_description=group_description,
parent=parent_group,
copy_permissions=copy_permissions
msg='created new repo group `%s`' % group_name,
repo_group=repo_group.get_api_data()
raise JSONRPCError('failed to create repo group `%s`' % (group_name,))
def update_repo_group(self, apiuser, repogroupid, group_name=Optional(''),
description=Optional(''),
parent=Optional(None), enable_locking=Optional(False)):
store_update(updates, group_name, 'group_name')
store_update(updates, description, 'group_description')
store_update(updates, owner, 'owner')
store_update(updates, parent, 'parent_group')
store_update(updates, enable_locking, 'enable_locking')
repo_group = RepoGroupModel().update(repo_group, updates)
msg='updated repository group ID:%s %s' % (repo_group.group_id,
repo_group.group_name),
raise JSONRPCError('failed to update repository group `%s`'
% (repogroupid,))
def delete_repo_group(self, apiuser, repogroupid):
:param repogroupid: name or id of repository group
'msg': 'deleted repo group ID:<repogroupid> <repogroupname>
'repo_group': null
"failed to delete repo group ID:<repogroupid> <repogroupname>"
RepoGroupModel().delete(repo_group)
msg='deleted repo group ID:%s %s' %
(repo_group.group_id, repo_group.group_name),
repo_group=None
raise JSONRPCError('failed to delete repo group ID:%s %s' %
(repo_group.group_id, repo_group.group_name)
def grant_user_permission_to_repo_group(self, apiuser, repogroupid, userid,
perm, apply_to_children=Optional('none')):
Grant permission for user on given repository group, or update existing
one if found. This command can be executed only using api_key belonging
to user with admin rights, or user who has admin right to given repository
group.
:param perm: (group.(none|read|write|admin))
:param apply_to_children: 'none', 'repos', 'groups', 'all'
:type apply_to_children: str
"msg" : "Granted perm: `<perm>` (recursive:<apply_to_children>) for user: `<username>` in repo group: `<repo_group_name>`",
"failed to edit permission for user: `<userid>` in repo group: `<repo_group_name>`"
# check if we have admin permission for this repo group !
if not HasRepoGroupPermissionAnyApi('group.admin')(user=apiuser,
group_name=repo_group.group_name):
raise JSONRPCError('repository group `%s` does not exist' % (repogroupid,))
perm = get_perm_or_error(perm, prefix='group.')
apply_to_children = Optional.extract(apply_to_children)
RepoGroupModel().add_permission(repo_group=repo_group,
obj=user,
obj_type="user",
perm=perm,
recursive=apply_to_children)
msg='Granted perm: `%s` (recursive:%s) for user: `%s` in repo group: `%s`' % (
perm.permission_name, apply_to_children, user.username, repo_group.name
'failed to edit permission for user: `%s` in repo group: `%s`' % (
userid, repo_group.name))
def revoke_user_permission_from_repo_group(self, apiuser, repogroupid, userid,
apply_to_children=Optional('none')):
Revoke permission for user on given repository group. This command can
be executed only using api_key belonging to user with admin rights, or
user who has admin right to given repository group.
:type userid:
"msg" : "Revoked perm (recursive:<apply_to_children>) for user: `<username>` in repo group: `<repo_group_name>`",
RepoGroupModel().delete_permission(repo_group=repo_group,
msg='Revoked perm (recursive:%s) for user: `%s` in repo group: `%s`' % (
apply_to_children, user.username, repo_group.name
def grant_user_group_permission_to_repo_group(
self, apiuser, repogroupid, usergroupid, perm,
Grant permission for user group on given repository group, or update
api_key belonging to user with admin rights, or user who has admin
right to given repository group.
"msg" : "Granted perm: `<perm>` (recursive:<apply_to_children>) for user group: `<usersgroupname>` in repo group: `<repo_group_name>`",
"failed to edit permission for user group: `<usergroup>` in repo group: `<repo_group_name>`"
_perms = ('group.admin',)
if not HasRepoGroupPermissionAnyApi(*_perms)(
user=apiuser, group_name=repo_group.group_name):
'repository group `%s` does not exist' % (repogroupid,))
'user group `%s` does not exist' % (usergroupid,))
obj=user_group,
obj_type="user_group",
msg='Granted perm: `%s` (recursive:%s) for user group: `%s` in repo group: `%s`' % (
perm.permission_name, apply_to_children,
user_group.users_group_name, repo_group.name
'repo group: `%s`' % (
usergroupid, repo_group.name
def revoke_user_group_permission_from_repo_group(
self, apiuser, repogroupid, usergroupid,
executed only using api_key belonging to user with admin rights, or
"msg" : "Revoked perm (recursive:<apply_to_children>) for user group: `<usersgroupname>` in repo group: `<repo_group_name>`",
msg='Revoked perm (recursive:%s) for user group: `%s` in repo group: `%s`' % (
apply_to_children, user_group.users_group_name, repo_group.name
'failed to edit permission for user group: `%s` in repo group: `%s`' % (
def get_gist(self, apiuser, gistid):
Get given gist by id
:param gistid: id of private or public gist
:type gistid: str
gist = get_gist_or_error(gistid)
if gist.gist_owner != apiuser.user_id:
raise JSONRPCError('gist `%s` does not exist' % (gistid,))
return gist.get_api_data()
def get_gists(self, apiuser, userid=Optional(OAttr('apiuser'))):
Get all gists for given user. If userid is empty returned gists
are for user who called the api
:param userid: user to get gists for
user_id = apiuser.user_id
user_id = get_user_or_error(userid).user_id
gists = []
_gists = Gist().query() \
.filter(or_(Gist.gist_expires == -1, Gist.gist_expires >= time.time())) \
.filter(Gist.gist_owner == user_id) \
.order_by(Gist.created_on.desc())
for gist in _gists:
gists.append(gist.get_api_data())
return gists
def create_gist(self, apiuser, files, owner=Optional(OAttr('apiuser')),
gist_type=Optional(Gist.GIST_PUBLIC), lifetime=Optional(-1),
description=Optional('')):
Creates new Gist
:param files: files to be added to gist
{'filename': {'content':'...', 'lexer': null},
'filename2': {'content':'...', 'lexer': null}}
:type files: dict
:param owner: gist owner, defaults to api method caller
:param gist_type: type of gist 'public' or 'private'
:type gist_type: Optional(str)
:param lifetime: time in minutes of gist lifetime
:type lifetime: Optional(int)
:param description: gist description
"msg": "created new gist",
"gist": {}
"failed to create gist"
gist_type = Optional.extract(gist_type)
lifetime = Optional.extract(lifetime)
gist = GistModel().create(description=description,
gist_mapping=files,
gist_type=gist_type,
lifetime=lifetime)
msg='created new gist',
gist=gist.get_api_data()
raise JSONRPCError('failed to create gist')
# def update_gist(self, apiuser, gistid, files, owner=Optional(OAttr('apiuser')),
# gist_type=Optional(Gist.GIST_PUBLIC),
# gist_lifetime=Optional(-1), gist_description=Optional('')):
# gist = get_gist_or_error(gistid)
# updates = {}
def delete_gist(self, apiuser, gistid):
Deletes existing gist
:param gistid: id of gist to delete
"deleted gist ID: <gist_id>",
"gist": null
"failed to delete gist ID:<gist_id>"
GistModel().delete(gist)
msg='deleted gist ID:%s' % (gist.gist_access_id,),
gist=None
raise JSONRPCError('failed to delete gist ID:%s'
% (gist.gist_access_id,))
Status change: