.. _api:

===

API

===

Starting from RhodeCode version 1.2 a simple API was implemented.

There's a single schema for calling all api methods. API is implemented

with JSON protocol both ways. An url to send API request to RhodeCode is

<your_server>/_admin/api

API ACCESS FOR WEB VIEWS

++++++++++++++++++++++++

API access can also be turned on for each web view in RhodeCode that is 

decorated with `@LoginRequired` decorator. To enable API access simple change 

the standard login decorator to `@LoginRequired(api_access=True)`. 

After this change, a rhodecode view can be accessed without login by adding a 

GET parameter `?api_key=<api_key>` to url. By default this is only

enabled on RSS/ATOM feed views.

API ACCESS

++++++++++

All clients are required to send JSON-RPC spec JSON data::

    {   

        "id:"<id>",

        "api_key":"<api_key>",

        "method":"<method_name>",

        "args":{"<arg_key>":"<arg_val>"}

    }

Example call for autopulling remotes repos using curl::

    curl https://server.com/_admin/api -X POST -H 'content-type:text/plain' --data-binary '{"id":1,"api_key":"xe7cdb2v278e4evbdf5vs04v832v0efvcbcve4a3","method":"pull","args":{"repo":"CPython"}}'

Simply provide

 - *id* A value of any type, which is used to match the response with the request that it is replying to.

 - *api_key* for access and permission validation.

 - *method* is name of method to call

 - *args* is an key:value list of arguments to pass to method

.. note::

    api_key can be found in your user account page

RhodeCode API will return always a JSON-RPC response::

    {   

        "id":<id>, # matching id sent by request

        "result": "<result>"|null, # JSON formatted result, null if any errors

        "error": "null"|<error_message> # JSON formatted error (if any)

    }

All responses from API will be `HTTP/1.0 200 OK`, if there's an error while

calling api *error* key from response will contain failure description

and result will be null.

API CLIENT

++++++++++

From version 1.4 RhodeCode adds a script that allows to easily

communicate with API. After installing RhodeCode a `rhodecode-api` script

will be available.

To get started quickly simply run::

  rhodecode-api _create_config --apikey=<youapikey> --apihost=<rhodecode host>

This will create a file named .config in the directory you executed it storing

json config file with credentials. You can skip this step and always provide

both of the arguments to be able to communicate with server

after that simply run any api command for example get_repo::

 rhodecode-api get_repo

 calling {"api_key": "<apikey>", "id": 75, "args": {}, "method": "get_repo"} to http://127.0.0.1:5000

 rhodecode said:

 {'error': 'Missing non optional `repoid` arg in JSON DATA',

  'id': 75,

  'result': None}

Ups looks like we forgot to add an argument

Let's try again now giving the repoid as parameters::

    rhodecode-api get_repo repoid:rhodecode   

    calling {"api_key": "<apikey>", "id": 39, "args": {"repoid": "rhodecode"}, "method": "get_repo"} to http://127.0.0.1:5000

    rhodecode said:

    {'error': None,

     'id': 39,

     'result': <json data...>}

API METHODS

+++++++++++

pull

----

Pulls given repo from remote location. Can be used to automatically keep

remote repos up to date. This command can be executed only using api_key

belonging to user with admin rights

INPUT::

    id : <id_for_response>

    api_key : "<api_key>"

    method :  "pull"

    args :    {

                "repoid" : "<reponame or repo_id>"

              }

OUTPUT::

    id : <id_given_in_input>

    result : "Pulled from `<reponame>`"

    error :  null

get_user

--------

Get's an user by username or user_id, Returns empty result if user is not found.

This command can be executed only using api_key belonging to user with admin 

rights.

INPUT::

    id : <id_for_response>

    api_key : "<api_key>"

    method :  "get_user"

    args :    { 

                "userid" : "<username or user_id>"

              }

OUTPUT::

    id : <id_given_in_input>

    result: None if user does not exist or 

            {

                "user_id" :  "<user_id>",

                "username" : "<username>",

                "firstname": "<firstname>",

                "lastname" : "<lastname>",

                "email" :    "<email>",

                "emails":    "<list_of_all_additional_emails>",

                "active" :   "<bool>",

                "admin" :    "<bool>",

                "ldap_dn" :  "<ldap_dn>",

                "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

get_users

---------

Lists all existing users. This command can be executed only using api_key

belonging to user with admin rights.

INPUT::

    id : <id_for_response>

    api_key : "<api_key>"

    method :  "get_users"

    args :    { }

OUTPUT::

    id : <id_given_in_input>

    result: [

              {

                "user_id" :  "<user_id>",

                "username" : "<username>",

                "firstname": "<firstname>",

                "lastname" : "<lastname>",

                "email" :    "<email>",

                "emails":    "<list_of_all_additional_emails>",

                "active" :   "<bool>",

                "admin" :    "<bool>",

                "ldap_dn" :  "<ldap_dn>",

                "last_login": "<last_login>",

              },

    	      …

            ]

    error:  null

create_user

-----------

Creates new user. This command can 

be executed only using api_key belonging to user with admin rights.

INPUT::

    id : <id_for_response>

    api_key : "<api_key>"

    method :  "create_user"

    args :    {

                "username" :  "<username>",

                "email" :     "<useremail>",

                "password" :  "<password>",

                "firstname" : "<firstname> = Optional(None)",

                "lastname" :  "<lastname> = Optional(None)",

                "active" :    "<bool> = Optional(True)",

                "admin" :     "<bool> = Optional(False)",

                "ldap_dn" :   "<ldap_dn> = Optional(None)"

              }

OUTPUT::

    id : <id_given_in_input>

    result: {

              "msg" : "created new user `<username>`",

              "user": {

                "user_id" :  "<user_id>",

                "username" : "<username>",

                "firstname": "<firstname>",

                "lastname" : "<lastname>",

                "email" :    "<email>",

                "emails":    "<list_of_all_additional_emails>",

                "active" :   "<bool>",

                "admin" :    "<bool>",

                "ldap_dn" :  "<ldap_dn>",

                "last_login": "<last_login>",

              },

            }

    error:  null

update_user

-----------

updates given user if such user exists. This command can 

be executed only using api_key belonging to user with admin rights.

INPUT::

    id : <id_for_response>

    api_key : "<api_key>"

    method :  "update_user"

    args :    {

                "userid" : "<user_id or username>",

                "username" :  "<username> = Optional",

                "email" :     "<useremail> = Optional",

                "password" :  "<password> = Optional",

                "firstname" : "<firstname> = Optional",

                "lastname" :  "<lastname> = Optional",

                "active" :    "<bool> = Optional",

                "admin" :     "<bool> = Optional",

                "ldap_dn" :   "<ldap_dn> = Optional"

              }

OUTPUT::

    id : <id_given_in_input>

    result: {

              "msg" : "updated user ID:<userid> <username>",

              "user": {

                "user_id" :  "<user_id>",

                "username" : "<username>",

                "firstname": "<firstname>",

                "lastname" : "<lastname>",

                "email" :    "<email>",

                "emails":    "<list_of_all_additional_emails>",

                "active" :   "<bool>",

                "admin" :    "<bool>",

                "ldap_dn" :  "<ldap_dn>",

                "last_login": "<last_login>",

              },              

            }

    error:  null

delete_user

-----------

deletes givenuser if such user exists. This command can 

be executed only using api_key belonging to user with admin rights.

INPUT::

    id : <id_for_response>

    api_key : "<api_key>"

    method :  "delete_user"

    args :    {

                "userid" : "<user_id or username>",

              }

OUTPUT::

    id : <id_given_in_input>

    result: {

              "msg" : "deleted user ID:<userid> <username>",

              "user": null

            }

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

"""

    rhodecode.controllers.api

    ~~~~~~~~~~~~~~~~~~~~~~~~~

    API controller for RhodeCode

    :created_on: Aug 20, 2011

    :author: marcink

    :copyright: (C) 2011-2012 Marcin Kuzminski <marcin@python-works.com>

    :license: GPLv3, see COPYING for more details.

"""

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

# modify it under the terms of the GNU General Public License

# as published by the Free Software Foundation; version 2

# of the License or (at your opinion) any later version of the license.

#

# 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 General Public License for more details.

#

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

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,

# MA  02110-1301, USA.

import traceback

import logging

from rhodecode.controllers.api import JSONRPCController, JSONRPCError

from rhodecode.lib.auth import HasPermissionAllDecorator, \

    HasPermissionAnyDecorator, PasswordGenerator, AuthUser

from rhodecode.model.meta import Session

from rhodecode.model.scm import ScmModel

from rhodecode.model.repo import RepoModel

from rhodecode.model.user import UserModel

from rhodecode.model.users_group import UsersGroupModel

from rhodecode.model.permission import PermissionModel

log = logging.getLogger(__name__)

class Optional(object):

    """

    Defines an optional parameter::

        param = param.getval() if isinstance(param, Optional) else param

        param = param() if isinstance(param, Optional) else param

    is equivalent of::

        param = Optional.extract(param)

    """

    def __init__(self, type_):

        self.type_ = type_

    def __repr__(self):

        return '<Optional:%s>' % self.type_.__repr__()

    def __call__(self):

        return self.getval()

    def getval(self):

        """

        returns value from this Optional instance

        """

        return self.type_

    @classmethod

    def extract(cls, val):

        if isinstance(val, cls):

            return val.getval()

        return val

def get_user_or_error(userid):

    """

    Get user by id or name or return JsonRPCError if not found

    :param userid:

    """

    user = UserModel().get_user(userid)

    if user is None:

        raise JSONRPCError("user `%s` does not exist" % userid)

    return user

def get_repo_or_error(repoid):

    """

    Get repo by id or name or return JsonRPCError if not found

    :param userid:

    """

    repo = RepoModel().get_repo(repoid)

    if repo is None:

        raise JSONRPCError('repository `%s` does not exist' % (repoid))

    return repo

def get_users_group_or_error(usersgroupid):

    """

    Get users group by id or name or return JsonRPCError if not found

    :param userid:

    """

    users_group = UsersGroupModel().get_group(usersgroupid)

    if users_group is None:

        raise JSONRPCError('users group `%s` does not exist' % usersgroupid)

    return users_group

def get_perm_or_error(permid):

    """

    Get permission by id or name or return JsonRPCError if not found

    :param userid:

    """

    perm = PermissionModel().get_permission_by_name(permid)

    if perm is None:

        raise JSONRPCError('permission `%s` does not exist' % (permid))

    return perm

class ApiController(JSONRPCController):

    """

    API Controller

    Each method needs to have USER as argument this is then based on given

    API_KEY propagated as instance of user object

    Preferably this should be first argument also

    Each function should also **raise** JSONRPCError for any

    errors that happens

    """

    @HasPermissionAllDecorator('hg.admin')

    def pull(self, apiuser, repoid):

        """

        Dispatch pull action on given repo

        :param apiuser:

        :param repoid:

        """

        repo = get_repo_or_error(repoid)

        try:

            ScmModel().pull_changes(repo.repo_name,

                                    self.rhodecode_user.username)

            return 'Pulled from `%s`' % repo.repo_name

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError(

                'Unable to pull changes from `%s`' % repo.repo_name

            )

    @HasPermissionAllDecorator('hg.admin')

    def get_user(self, apiuser, userid):

        """"

        Get a user by username

        :param apiuser:

        :param userid:

        """

        user = get_user_or_error(userid)

        data = user.get_api_data()

        data['permissions'] = AuthUser(user_id=user.user_id).permissions

        return data

    @HasPermissionAllDecorator('hg.admin')

    def get_users(self, apiuser):

        """"

        Get all users

        :param apiuser:

        """

        result = []

        for user in UserModel().get_all():

            result.append(user.get_api_data())

        return result

    @HasPermissionAllDecorator('hg.admin')

    def create_user(self, apiuser, username, email, password,

                    firstname=Optional(None), lastname=Optional(None),

                    active=Optional(True), admin=Optional(False),

                    ldap_dn=Optional(None)):

        """

        Create new user

        :param apiuser:

        :param username:

        :param email:

        :param password:

        :param firstname:

        :param lastname:

        :param active:

        :param admin:

        :param ldap_dn:

        """

        if UserModel().get_by_username(username):

            raise JSONRPCError("user `%s` already exist" % username)

        if UserModel().get_by_email(email, case_insensitive=True):

            raise JSONRPCError("email `%s` already exist" % email)

        if ldap_dn:

            # generate temporary password if ldap_dn

            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),

                ldap_dn=Optional.extract(ldap_dn)

            )

            Session().commit()

            return dict(

                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)

    @HasPermissionAllDecorator('hg.admin')

    def update_user(self, apiuser, userid, username=Optional(None),

                    email=Optional(None), firstname=Optional(None),

                    lastname=Optional(None), active=Optional(None),

                    admin=Optional(None), ldap_dn=Optional(None),

                    password=Optional(None)):

        """

        Updates given user

        :param apiuser:

        :param userid:

        :param username:

        :param email:

        :param firstname:

        :param lastname:

        :param active:

        :param admin:

        :param ldap_dn:

        :param password:

        """

        user = get_user_or_error(userid)

        # call function and store only updated arguments

        updates = {}

        def store_update(attr, name):

            if not isinstance(attr, Optional):

                updates[name] = attr

        try:

            store_update(username, 'username')

            store_update(password, 'password')

            store_update(email, 'email')

            store_update(firstname, 'name')

            store_update(lastname, 'lastname')

            store_update(active, 'active')

            store_update(admin, 'admin')

            store_update(ldap_dn, 'ldap_dn')

            user = UserModel().update_user(user, **updates)

            Session().commit()

            return dict(

                msg='updated user ID:%s %s' % (user.user_id, user.username),

                user=user.get_api_data()

            )

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError('failed to update user `%s`' % userid)

    @HasPermissionAllDecorator('hg.admin')

    def delete_user(self, apiuser, userid):

        """"

        Deletes an user

        :param apiuser:

        :param userid:

        """

        user = get_user_or_error(userid)

        try:

            UserModel().delete(userid)

            Session().commit()

            return dict(

                msg='deleted user ID:%s %s' % (user.user_id, user.username),

                user=None

            )

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError('failed to delete ID:%s %s' % (user.user_id,

                                                              user.username))

    @HasPermissionAllDecorator('hg.admin')

    def get_users_group(self, apiuser, usersgroupid):

        """"

        Get users group by name or id

        :param apiuser:

        :param usersgroupid:

        """

        users_group = get_users_group_or_error(usersgroupid)

        data = users_group.get_api_data()

        members = []

        for user in users_group.members:

            user = user.user

            members.append(user.get_api_data())

        data['members'] = members

        return data

    @HasPermissionAllDecorator('hg.admin')

    def get_users_groups(self, apiuser):

        """"

        Get all users groups

        :param apiuser:

        """

        result = []

        for users_group in UsersGroupModel().get_all():

            result.append(users_group.get_api_data())

        return result

    @HasPermissionAllDecorator('hg.admin')

    def create_users_group(self, apiuser, group_name, active=Optional(True)):

        """

        Creates an new usergroup

        :param apiuser:

        :param group_name:

        :param active:

        """

        if UsersGroupModel().get_by_name(group_name):

            raise JSONRPCError("users group `%s` already exist" % group_name)

from __future__ import with_statement

import random

import mock

from rhodecode.tests import *

from rhodecode.lib.compat import json

from rhodecode.lib.auth import AuthUser

from rhodecode.model.user import UserModel

from rhodecode.model.users_group import UsersGroupModel

from rhodecode.model.repo import RepoModel

from rhodecode.model.meta import Session

API_URL = '/_admin/api'

def _build_data(apikey, method, **kw):

    """

    Builds API data with given random ID

    :param random_id:

    :type random_id:

    """

    random_id = random.randrange(1, 9999)

    return random_id, json.dumps({

        "id": random_id,

        "api_key": apikey,

        "method": method,

        "args": kw

    })

jsonify = lambda obj: json.loads(json.dumps(obj))

def crash(*args, **kwargs):

    raise Exception('Total Crash !')

TEST_USERS_GROUP = 'test_users_group'

def make_users_group(name=TEST_USERS_GROUP):

    gr = UsersGroupModel().create(name=name)

    UsersGroupModel().add_user_to_group(users_group=gr,

                                        user=TEST_USER_ADMIN_LOGIN)

    Session().commit()

    return gr

def destroy_users_group(name=TEST_USERS_GROUP):

    UsersGroupModel().delete(users_group=name, force=True)

    Session().commit()

def create_repo(repo_name, repo_type):

    # create new repo

    form_data = dict(repo_name=repo_name,

                     repo_name_full=repo_name,

                     fork_name=None,

                     description='description %s' % repo_name,

                     repo_group=None,

                     private=False,

                     repo_type=repo_type,

                     clone_uri=None,

                     landing_rev='tip')

    cur_user = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)

    r = RepoModel().create(form_data, cur_user)

    Session().commit()

    return r

def create_fork(fork_name, fork_type, fork_of):

    fork = RepoModel(Session())._get_repo(fork_of)

    r = create_repo(fork_name, fork_type)

    r.fork = fork

    Session().add(r)

    Session().commit()

    return r

def destroy_repo(repo_name):

    RepoModel().delete(repo_name)

    Session().commit()

class BaseTestApi(object):

    REPO = None

    REPO_TYPE = None

    @classmethod

    def setUpClass(self):

        self.usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)

        self.apikey = self.usr.api_key

        self.TEST_USER = UserModel().create_or_update(

            username='test-api',

            password='test',

            email='test@api.rhodecode.org',

            firstname='first',

            lastname='last'

        )

        Session().commit()

        self.TEST_USER_LOGIN = self.TEST_USER.username

    @classmethod

    def teardownClass(self):

        pass

    def setUp(self):

        self.maxDiff = None

        make_users_group()

    def tearDown(self):

        destroy_users_group()

    def _compare_ok(self, id_, expected, given):

        expected = jsonify({

            'id': id_,

            'error': None,

            'result': expected

        })

        given = json.loads(given)

        self.assertEqual(expected, given)

    def _compare_error(self, id_, expected, given):

        expected = jsonify({

            'id': id_,

            'error': expected,

            'result': None

        })

        given = json.loads(given)

        self.assertEqual(expected, given)

#    def test_Optional(self):

#        from rhodecode.controllers.api.api import Optional

#        option1 = Optional(None)

#        self.assertEqual('<Optional:%s>' % None, repr(option1))

#

#        self.assertEqual(1, Optional.extract(Optional(1)))

#        self.assertEqual('trololo', Optional.extract('trololo'))

    def test_api_wrong_key(self):

        id_, params = _build_data('trololo', 'get_user')

        response = self.app.post(API_URL, content_type='application/json',