src.pyejabberd.EjabberdAPIClient.__init__() - Code Metrics - dirkmoors/pyejabberd - Measure and Improve Code Quality continuously with Scrutinizer

src.pyejabberd.EjabberdAPIClient.init() B
last analyzed 2016-03-09 14:09 UTC

↳ Parent: src.pyejabberd.EjabberdAPIClient

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	1
dl	0
loc	26
rs	8.8571

# -*- coding: utf-8 -*-
from __future__ import unicode_literals, print_function
import copy

from builtins import range

from .compat import urlparse
from .compat import xmlrpc_client

from . import contract, definitions, defaults
from .core.definitions import API, APIArgument
from .core.errors import IllegalArgumentError


class EjabberdAPIClient(contract.EjabberdAPIContract):
    """
    Python Client for the Ejabberd XML-RPC API
    """
    def __init__(self, host, port, username, password, user_domain, protocol=None, verbose=False):
        """
        Constructor
        :param host:
        :type host: str|unicode
        :param port:
        :type port: int
        :param username:
        :type username: str|unicode
        :param password:
        :type password: str|unicode
        :param user_domain:
        :type user_domain: str|unicode
        :param protocol: http or https
        :type protocol: str|unicode
        :param verbose:
        :type verbose: bool
        """
        self.host = host
        self.port = port
        self.username = username
        self.password = password
        self.user_domain = user_domain
        self.protocol = protocol or defaults.XMLRPC_API_PROTOCOL
        self.verbose = verbose
        self._proxy = None

    @staticmethod
    def get_instance(service_url, verbose=False):
        """
        Returns a EjabberdAPIClient instance based on a '12factor app' compliant service_url

        :param service_url: A connection string in the format:
            <http|https>://<username>:<password>@<host>(:port)/user_domain
        :type service_url: str|unicode
        :param verbose:
        :type verbose: bool
        :return: EjabberdAPIClient instance
        """
        fmt_error = \
            'from_string expects service_url like https://USERNAME:PASSWORD@HOST:PORT/DOMAIN'

        o = urlparse(service_url)

        protocol = o.scheme
        assert protocol in ('http', 'https'), fmt_error

        netloc_parts = o.netloc.split('@')
        assert len(netloc_parts) == 2, fmt_error

        auth, server = netloc_parts

        auth_parts = auth.split(':')
        assert len(auth_parts) == 2, fmt_error

        username, password = auth_parts

        server_parts = server.split(':')
        assert len(server_parts) <= 2, fmt_error

        if len(server_parts) == 2:
            host, port = server_parts
            port = int(port)
        else:
            host, port = server_parts[0], defaults.XMLRPC_API_PORT

        path_parts = o.path.lstrip('/').split('/')
        assert len(path_parts) == 1, fmt_error

        user_domain = path_parts[0]

        return EjabberdAPIClient(host, port, username, password, user_domain, protocol=protocol,
                                 verbose=verbose)

    @property
    def service_url(self):
        """
        Returns the FQDN to the Ejabberd server's XML-RPC endpoint
        :return:
        """
        return "%s://%s:%s/" % (self.protocol, self.host, self.port)

    @property
    def proxy(self):
        """
        Returns the proxy object that is used to perform the calls to the XML-RPC endpoint
        :rtype: :py:class:xmlrpclib.ServerProxy
        :return the proxy object that is used to perform the calls to the XML-RPC endpoint
        """
        if self._proxy is None:
            self._proxy = xmlrpc_client.ServerProxy(self.service_url, verbose=(1 if self.verbose else 0))
        return self._proxy

    @property
    def auth(self):
        """
        Returns a dictionary containing the basic authorization info
        :rtype: dict
        :return: a dictionary containing the basic authorization info
        """
        return {
            'user': self.username,
            'server': self.user_domain,
            'password': self.password
        }

    def echo(self, sentence):
        """
        Echo's the input back
        :param sentence:
        :type sentence: str|unicode
        :rtype: str|unicode
        :return: The echoed response, which should be the same as the input
        """
        return self._call_api(definitions.Echo, sentence=sentence)

    def registered_users(self, host):
        """
        List all registered users in the xmpp_host
        :param host: The XMPP_DOMAIN
        :type host: str|unicode
        :rtype: Iterable
        :return: A list of registered users in the xmpp_host
        """
        return self._call_api(definitions.RegisteredUsers, host=host)

    def register(self, user, host, password):
        """
        Registers a user to the ejabberd server
        :param user: The username for the new user
        :type user: str|unicode
        :param host: The XMPP_DOMAIN
        :type host: str|unicode
        :param password: The password for the new user
        :type password: str|unicode
        :rtype: bool
        :return: A boolean indicating if the registration has succeeded
        """
        return self._call_api(definitions.Register, user=user, host=host, password=password)

    def unregister(self, user, host):
        """
        UnRegisters a user from the ejabberd server
        :param user: The username for the new user
        :type user: str|unicode
        :param host: The XMPP_DOMAIN
        :type host: str|unicode
        :rtype: bool
        :return: A boolean indicating if the unregistration has succeeded
        """
        return self._call_api(definitions.UnRegister, user=user, host=host)

    def change_password(self, user, host, newpass):
        """
        Change the password for a given user
        :param user: The username for the user we want to change the password for
        :type user: str|unicode
        :param host: The XMPP_DOMAIN
        :type host: str|unicode
        :param newpass: The new password
        :type newpass: str|unicode
        :rtype: bool
        :return: A boolean indicating if the password change has succeeded
        """
        return self._call_api(definitions.ChangePassword, user=user, host=host, newpass=newpass)

    def check_password_hash(self, user, host, password):
        """
        Checks whether a password is correct for a given user. The used hash-method is fixed to sha1.
        :param user: The username for the user we want to check the password for
        :type user: str|unicode
        :param host: The XMPP_DOMAIN
        :type host: str|unicode
        :param password: The password we want to check for the user
        :type password: str|unicode
        :rtype: bool
        :return: A boolean indicating if the given password matches the user's password
        """
        return self._call_api(definitions.CheckPasswordHash, user=user, host=host, password=password)

    def set_nickname(self, user, host, nickname):
        """
        Set nickname in a user's vCard
        :param user: The username for the user we want to set the nickname to
        :type user: str|unicode
        :param host: The XMPP_DOMAIN
        :type host: str|unicode
        :param nickname: The nickname to assign to the user
        :type nickname: str|unicode
        :rtype: bool
        :return: A boolean indicating nickname was assigned successfully
        """
        return self._call_api(definitions.SetNickname, user=user, host=host, nickname=nickname)

    def connected_users(self):
        """
        List all established sessions
        :rtype: list
        :return: a list of dictionaries containing user jids
        """
        return self._call_api(definitions.ConnectedUsers)

    def connected_users_info(self):
        """
        List all established sessions and their information
        :rtype: list
        :return: a list of dictionaries containing user info
        """
        return self._call_api(definitions.ConnectedUsersInfo)

    def connected_users_number(self):
        """
        Get the number of established sessions
        :rtype: int
        :return: number of established user sessions
        """
        return self._call_api(definitions.ConnectedUsersNumber)

    def user_sessions_info(self, user, host):
        """
        Get information about all sessions of a user
        :param user: The username for the user we want info for
        :type user: str|unicode
        :param host: The XMPP_DOMAIN
        :type host: str|unicode
        :rtype: list
        :return: list of information of sessions for a user
        """
        return self._call_api(definitions.UserSessionInfo, user=user, host=host)

    def muc_online_rooms(self, host=None):
        """
        List existing rooms ('global' to get all vhosts)
        :param host: The XMPP_DOMAIN
        :type host: str|unicode
        :rtype: Iterable
        :return: A list of online rooms in the format 'name@service'
        """
        host = host or 'global'
        return self._call_api(definitions.MucOnlineRooms, host=host)

    def create_room(self, name, service, host):
        """
        Create a MUC room name@service in host
        :param name: The name for the room
        :type name: str|unicode
        :param service: The MUC service name (e.g. "conference")
        :type service: str|unicode
        :param host: The XMPP_DOMAIN
        :type host: str|unicode
        :rtype: bool
        :return: A boolean indicating whether the room has been created successfully
        """
        return self._call_api(definitions.CreateRoom, name=name, service=service, host=host)

    def destroy_room(self, name, service, host):
        """
        Destroy a MUC room
        :param name: The name for the room
        :type name: str|unicode
        :param service: The MUC service name (e.g. "conference")
        :type service: str|unicode
        :param host: The XMPP_DOMAIN
        :type host: str|unicode
        :rtype: bool
        :return: A boolean indicating whether the room has been destroyed successfully
        """
        return self._call_api(definitions.DestroyRoom, name=name, service=service, host=host)

    def get_room_options(self, name, service):
        """
        Get options from a MUC room
        :param name: The name for the room
        :type name: str|unicode
        :param service: The MUC service name (e.g. "conference")
        :type service: str|unicode
        :rtype: dict
        :return: A dict containing the room options
        """
        return self._call_api(definitions.GetRoomOptions, name=name, service=service)

    def change_room_option(self, name, service, option, value):
        """
        Change an option in a MUC room
        :param name: The name for the room
        :type name: str|unicode
        :param service: The MUC service name (e.g. "conference")
        :type service: str|unicode
        :param option: The option to change
        :type option: muc.enums.MUCRoomOption
        :param value: The new value
        :type value: str|unicode|int|bool
        :rtype: bool
        :return: A boolean indicating whether the room option has been changed successfully
        """
        return self._call_api(definitions.ChangeRoomOption, name=name, service=service, option=option, value=value)

    def set_room_affiliation(self, name, service, jid, affiliation):
        """
        Change an affiliation for a user in a MUC room
        :param name:The name for the room
        :type name: str|unicode
        :param service: The MUC service name (e.g. "conference")
        :type service: str|unicode
        :param jid: The jabber id for the user you want to change the affiliation for
        :type jid: str|unicode
        :param affiliation: The affiliation to the room
        :type affiliation: muc.enums.Affiliation
        :rtype: list
        :return: A list containing dictionaries containing affiliation info
        """
        return self._call_api(definitions.SetRoomAffiliation, name=name, service=service, jid=jid,
                              affiliation=affiliation)

    def get_room_affiliations(self, name, service):
        """
        Get the affiliations for a MUC room
        :param name:The name for the room
        :type name: str|unicode
        :param service: The MUC service name (e.g. "conference")
        :type service: str|unicode
        :return:
        """
        return self._call_api(definitions.GetRoomAffiliations, name=name, service=service)

    def add_rosteritem(self, localuser, localserver, user, server, nick, group, subs):
        """
        Add an item to a user's roster

        :param localuser: The username of user we are going to add a contact to
        :type localuser: str|unicode
        :param localserver: The XMPP_DOMAIN
        :type localserver: str|unicode
        :param user: The contact we are going to add to the user's roster
        :type user: str|unicode
        :param server: The XMPP_DOMAIN
        :type server: str|unicode
        :param nick: Nickname of the contact
        :type nick: str|unicode
        :param group: To what contact group the contact goes to
        :type group: str|unicode
        :param subs: The type of subscription
        :type subs: str|unicode
        :return:
        """
        return self._call_api(definitions.AddRosterItem,
                              localuser=localuser, localserver=localserver,
                              user=user, server=server,
                              nick=nick, group=group, subs=subs)

    def delete_rosteritem(self, localuser, localserver, user, server):
        """
        Delete an item from a user's roster

        :param localuser: The username of user we are going to delete a contact from
        :type localuser: str|unicode
        :param localserver: The XMPP_DOMAIN
        :type localserver: str|unicode
        :param user: The contact we are going to delete from the user's roster
        :type user: str|unicode
        :param server: The XMPP_DOMAIN
        :type server: str|unicode
        :return:
        """
        return self._call_api(definitions.DeleteRosterItem, localuser=localuser, localserver=localserver, user=user, server=server)

    def get_roster(self, user, host):
        """
        Get roster of a user

        :param user: The username of the user we want contact information for
        :type user: str|unicode
        :param host: The XMPP_DOMAIN
        :type host: str|unicode
        :rtype: Iterable
        :return: A list of user's contacts
        """
        return self._call_api(definitions.GetRoster, user=user, host=host)

    def _validate_and_serialize_arguments(self, api, arguments):
        """
        Internal method to validate and serialize arguments
        :param api: An instance of an API class
        :param arguments: A dictionary of arguments that will be passed to the method
        :type arguments: dict
        :rtype: dict
        :return: The serialized arguments
        """
        serialized_arguments = {}

        for i in range(len(api.arguments)):
            argument_descriptor = api.arguments[i]
            assert isinstance(argument_descriptor, APIArgument)

            # Validate argument presence
            argument_name = str(argument_descriptor.name)
            if argument_descriptor.required and argument_name not in arguments:
                raise IllegalArgumentError('Missing required argument "%s"' % argument_name)

            # Serializer argument value
            serialized_arguments[argument_descriptor.name] = \
                argument_descriptor.serializer_class().to_api(arguments.get(argument_name))

        return serialized_arguments

    def _report_method_call(self, method, arguments):
        """
        Internal method to print info about a method call
        :param method: The name oft hem ethod to call
        :type method: str|unicode
        :param arguments: A dictionary of arguments that will be passed to the method
        :type: arguments: dict
        :return:
        """
        if self.verbose:
            print('===> %s(%s)' % (method, ', '.join(['%s=%s' % (key, value) for (key, value) in arguments.items()])))

    def _call_api(self, api_class, **kwargs):
        """
        Internal method used to perform api calls
        :param api_class:
        :type api_class: py:class:API
        :param kwargs:
        :type kwargs: dict
        :rtype: object
        :return: Returns return value of the XMLRPC Method call
        """
        # Validate api_class
        assert issubclass(api_class, API)

        # Create api instance
        api = api_class()

        # Copy arguments
        arguments = copy.copy(kwargs)

        # Transform arguments
        arguments = api.transform_arguments(**arguments)

        # Validate and serialize arguments
        arguments = self._validate_and_serialize_arguments(api, arguments)

        # Retrieve method
        method = getattr(self.proxy, str(api.method))

        # Print method call with arguments
        self._report_method_call(api.method, arguments)

        # Perform call
        if not api.authenticate:
            response = method(arguments)
        else:
            response = method(self.auth, arguments)

        # Validate response
        api.validate_response(api, arguments, response)

        # Transform response
        result = api.transform_response(api, arguments, response)

        return result


1			# -- coding: utf-8 --
2			from __future__ import unicode_literals, print_function
3			import copy
4
5			from builtins import range
6
7			from .compat import urlparse
8			from .compat import xmlrpc_client
9
10			from . import contract, definitions, defaults
11			from .core.definitions import API, APIArgument
12			from .core.errors import IllegalArgumentError
13
14
15			class EjabberdAPIClient(contract.EjabberdAPIContract):
16			"""
17			Python Client for the Ejabberd XML-RPC API
18			"""
19			def __init__(self, host, port, username, password, user_domain, protocol=None, verbose=False):
20			"""
21			Constructor
22			:param host:
23			:type host: str\|unicode
24			:param port:
25			:type port: int
26			:param username:
27			:type username: str\|unicode
28			:param password:
29			:type password: str\|unicode
30			:param user_domain:
31			:type user_domain: str\|unicode
32			:param protocol: http or https
33			:type protocol: str\|unicode
34			:param verbose:
35			:type verbose: bool
36			"""
37			self.host = host
38			self.port = port
39			self.username = username
40			self.password = password
41			self.user_domain = user_domain
42			self.protocol = protocol or defaults.XMLRPC_API_PROTOCOL
43			self.verbose = verbose
44			self._proxy = None
45
46			@staticmethod
47			def get_instance(service_url, verbose=False):
48			"""
49			Returns a EjabberdAPIClient instance based on a '12factor app' compliant service_url
50
51			:param service_url: A connection string in the format:
52			<http\|https>://<username>:<password>@<host>(:port)/user_domain
53			:type service_url: str\|unicode
54			:param verbose:
55			:type verbose: bool
56			:return: EjabberdAPIClient instance
57			"""
58			fmt_error = \
59			'from_string expects service_url like https://USERNAME:PASSWORD@HOST:PORT/DOMAIN'
60
61			o = urlparse(service_url)
62
63			protocol = o.scheme
64			assert protocol in ('http', 'https'), fmt_error
65
66			netloc_parts = o.netloc.split('@')
67			assert len(netloc_parts) == 2, fmt_error
68
69			auth, server = netloc_parts
70
71			auth_parts = auth.split(':')
72			assert len(auth_parts) == 2, fmt_error
73
74			username, password = auth_parts
75
76			server_parts = server.split(':')
77			assert len(server_parts) <= 2, fmt_error
78
79			if len(server_parts) == 2:
80			host, port = server_parts
81			port = int(port)
82			else:
83			host, port = server_parts[0], defaults.XMLRPC_API_PORT
84
85			path_parts = o.path.lstrip('/').split('/')
86			assert len(path_parts) == 1, fmt_error
87
88			user_domain = path_parts[0]
89
90			return EjabberdAPIClient(host, port, username, password, user_domain, protocol=protocol,
91			verbose=verbose)
92
93			@property
94			def service_url(self):
95			"""
96			Returns the FQDN to the Ejabberd server's XML-RPC endpoint
97			:return:
98			"""
99			return "%s://%s:%s/" % (self.protocol, self.host, self.port)
100
101			@property
102			def proxy(self):
103			"""
104			Returns the proxy object that is used to perform the calls to the XML-RPC endpoint
105			:rtype: :py:class:xmlrpclib.ServerProxy
106			:return the proxy object that is used to perform the calls to the XML-RPC endpoint
107			"""
108			if self._proxy is None:
109			self._proxy = xmlrpc_client.ServerProxy(self.service_url, verbose=(1 if self.verbose else 0))
110			return self._proxy
111
112			@property
113			def auth(self):
114			"""
115			Returns a dictionary containing the basic authorization info
116			:rtype: dict
117			:return: a dictionary containing the basic authorization info
118			"""
119			return {
120			'user': self.username,
121			'server': self.user_domain,
122			'password': self.password
123			}
124
125			def echo(self, sentence):
126			"""
127			Echo's the input back
128			:param sentence:
129			:type sentence: str\|unicode
130			:rtype: str\|unicode
131			:return: The echoed response, which should be the same as the input
132			"""
133			return self._call_api(definitions.Echo, sentence=sentence)
134
135			def registered_users(self, host):
136			"""
137			List all registered users in the xmpp_host
138			:param host: The XMPP_DOMAIN
139			:type host: str\|unicode
140			:rtype: Iterable
141			:return: A list of registered users in the xmpp_host
142			"""
143			return self._call_api(definitions.RegisteredUsers, host=host)
144
145			def register(self, user, host, password):
146			"""
147			Registers a user to the ejabberd server
148			:param user: The username for the new user
149			:type user: str\|unicode
150			:param host: The XMPP_DOMAIN
151			:type host: str\|unicode
152			:param password: The password for the new user
153			:type password: str\|unicode
154			:rtype: bool
155			:return: A boolean indicating if the registration has succeeded
156			"""
157			return self._call_api(definitions.Register, user=user, host=host, password=password)
158
159			def unregister(self, user, host):
160			"""
161			UnRegisters a user from the ejabberd server
162			:param user: The username for the new user
163			:type user: str\|unicode
164			:param host: The XMPP_DOMAIN
165			:type host: str\|unicode
166			:rtype: bool
167			:return: A boolean indicating if the unregistration has succeeded
168			"""
169			return self._call_api(definitions.UnRegister, user=user, host=host)
170
171			def change_password(self, user, host, newpass):
172			"""
173			Change the password for a given user
174			:param user: The username for the user we want to change the password for
175			:type user: str\|unicode
176			:param host: The XMPP_DOMAIN
177			:type host: str\|unicode
178			:param newpass: The new password
179			:type newpass: str\|unicode
180			:rtype: bool
181			:return: A boolean indicating if the password change has succeeded
182			"""
183			return self._call_api(definitions.ChangePassword, user=user, host=host, newpass=newpass)
184
185			def check_password_hash(self, user, host, password):
186			"""
187			Checks whether a password is correct for a given user. The used hash-method is fixed to sha1.
188			:param user: The username for the user we want to check the password for
189			:type user: str\|unicode
190			:param host: The XMPP_DOMAIN
191			:type host: str\|unicode
192			:param password: The password we want to check for the user
193			:type password: str\|unicode
194			:rtype: bool
195			:return: A boolean indicating if the given password matches the user's password
196			"""
197			return self._call_api(definitions.CheckPasswordHash, user=user, host=host, password=password)
198
199			def set_nickname(self, user, host, nickname):
200			"""
201			Set nickname in a user's vCard
202			:param user: The username for the user we want to set the nickname to
203			:type user: str\|unicode
204			:param host: The XMPP_DOMAIN
205			:type host: str\|unicode
206			:param nickname: The nickname to assign to the user
207			:type nickname: str\|unicode
208			:rtype: bool
209			:return: A boolean indicating nickname was assigned successfully
210			"""
211			return self._call_api(definitions.SetNickname, user=user, host=host, nickname=nickname)
212
213			def connected_users(self):
214			"""
215			List all established sessions
216			:rtype: list
217			:return: a list of dictionaries containing user jids
218			"""
219			return self._call_api(definitions.ConnectedUsers)
220
221			def connected_users_info(self):
222			"""
223			List all established sessions and their information
224			:rtype: list
225			:return: a list of dictionaries containing user info
226			"""
227			return self._call_api(definitions.ConnectedUsersInfo)
228
229			def connected_users_number(self):
230			"""
231			Get the number of established sessions
232			:rtype: int
233			:return: number of established user sessions
234			"""
235			return self._call_api(definitions.ConnectedUsersNumber)
236
237			def user_sessions_info(self, user, host):
238			"""
239			Get information about all sessions of a user
240			:param user: The username for the user we want info for
241			:type user: str\|unicode
242			:param host: The XMPP_DOMAIN
243			:type host: str\|unicode
244			:rtype: list
245			:return: list of information of sessions for a user
246			"""
247			return self._call_api(definitions.UserSessionInfo, user=user, host=host)
248
249			def muc_online_rooms(self, host=None):
250			"""
251			List existing rooms ('global' to get all vhosts)
252			:param host: The XMPP_DOMAIN
253			:type host: str\|unicode
254			:rtype: Iterable
255			:return: A list of online rooms in the format 'name@service'
256			"""
257			host = host or 'global'
258			return self._call_api(definitions.MucOnlineRooms, host=host)
259
260			def create_room(self, name, service, host):
261			"""
262			Create a MUC room name@service in host
263			:param name: The name for the room
264			:type name: str\|unicode
265			:param service: The MUC service name (e.g. "conference")
266			:type service: str\|unicode
267			:param host: The XMPP_DOMAIN
268			:type host: str\|unicode
269			:rtype: bool
270			:return: A boolean indicating whether the room has been created successfully
271			"""
272			return self._call_api(definitions.CreateRoom, name=name, service=service, host=host)
273
274			def destroy_room(self, name, service, host):
275			"""
276			Destroy a MUC room
277			:param name: The name for the room
278			:type name: str\|unicode
279			:param service: The MUC service name (e.g. "conference")
280			:type service: str\|unicode
281			:param host: The XMPP_DOMAIN
282			:type host: str\|unicode
283			:rtype: bool
284			:return: A boolean indicating whether the room has been destroyed successfully
285			"""
286			return self._call_api(definitions.DestroyRoom, name=name, service=service, host=host)
287
288			def get_room_options(self, name, service):
289			"""
290			Get options from a MUC room
291			:param name: The name for the room
292			:type name: str\|unicode
293			:param service: The MUC service name (e.g. "conference")
294			:type service: str\|unicode
295			:rtype: dict
296			:return: A dict containing the room options
297			"""
298			return self._call_api(definitions.GetRoomOptions, name=name, service=service)
299
300			def change_room_option(self, name, service, option, value):
301			"""
302			Change an option in a MUC room
303			:param name: The name for the room
304			:type name: str\|unicode
305			:param service: The MUC service name (e.g. "conference")
306			:type service: str\|unicode
307			:param option: The option to change
308			:type option: muc.enums.MUCRoomOption
309			:param value: The new value
310			:type value: str\|unicode\|int\|bool
311			:rtype: bool
312			:return: A boolean indicating whether the room option has been changed successfully
313			"""
314			return self._call_api(definitions.ChangeRoomOption, name=name, service=service, option=option, value=value)
315
316			def set_room_affiliation(self, name, service, jid, affiliation):
317			"""
318			Change an affiliation for a user in a MUC room
319			:param name:The name for the room
320			:type name: str\|unicode
321			:param service: The MUC service name (e.g. "conference")
322			:type service: str\|unicode
323			:param jid: The jabber id for the user you want to change the affiliation for
324			:type jid: str\|unicode
325			:param affiliation: The affiliation to the room
326			:type affiliation: muc.enums.Affiliation
327			:rtype: list
328			:return: A list containing dictionaries containing affiliation info
329			"""
330			return self._call_api(definitions.SetRoomAffiliation, name=name, service=service, jid=jid,
331			affiliation=affiliation)
332
333			def get_room_affiliations(self, name, service):
334			"""
335			Get the affiliations for a MUC room
336			:param name:The name for the room
337			:type name: str\|unicode
338			:param service: The MUC service name (e.g. "conference")
339			:type service: str\|unicode
340			:return:
341			"""
342			return self._call_api(definitions.GetRoomAffiliations, name=name, service=service)
343
344			def add_rosteritem(self, localuser, localserver, user, server, nick, group, subs):
345			"""
346			Add an item to a user's roster
347
348			:param localuser: The username of user we are going to add a contact to
349			:type localuser: str\|unicode
350			:param localserver: The XMPP_DOMAIN
351			:type localserver: str\|unicode
352			:param user: The contact we are going to add to the user's roster
353			:type user: str\|unicode
354			:param server: The XMPP_DOMAIN
355			:type server: str\|unicode
356			:param nick: Nickname of the contact
357			:type nick: str\|unicode
358			:param group: To what contact group the contact goes to
359			:type group: str\|unicode
360			:param subs: The type of subscription
361			:type subs: str\|unicode
362			:return:
363			"""
364			return self._call_api(definitions.AddRosterItem,
365			localuser=localuser, localserver=localserver,
366			user=user, server=server,
367			nick=nick, group=group, subs=subs)
368
369			def delete_rosteritem(self, localuser, localserver, user, server):
370			"""
371			Delete an item from a user's roster
372
373			:param localuser: The username of user we are going to delete a contact from
374			:type localuser: str\|unicode
375			:param localserver: The XMPP_DOMAIN
376			:type localserver: str\|unicode
377			:param user: The contact we are going to delete from the user's roster
378			:type user: str\|unicode
379			:param server: The XMPP_DOMAIN
380			:type server: str\|unicode
381			:return:
382			"""
383			return self._call_api(definitions.DeleteRosterItem, localuser=localuser, localserver=localserver, user=user, server=server)
384
385			def get_roster(self, user, host):
386			"""
387			Get roster of a user
388
389			:param user: The username of the user we want contact information for
390			:type user: str\|unicode
391			:param host: The XMPP_DOMAIN
392			:type host: str\|unicode
393			:rtype: Iterable
394			:return: A list of user's contacts
395			"""
396			return self._call_api(definitions.GetRoster, user=user, host=host)
397
398			def _validate_and_serialize_arguments(self, api, arguments):
399			"""
400			Internal method to validate and serialize arguments
401			:param api: An instance of an API class
402			:param arguments: A dictionary of arguments that will be passed to the method
403			:type arguments: dict
404			:rtype: dict
405			:return: The serialized arguments
406			"""
407			serialized_arguments = {}
408
409			for i in range(len(api.arguments)):
410			argument_descriptor = api.arguments[i]
411			assert isinstance(argument_descriptor, APIArgument)
412
413			# Validate argument presence
414			argument_name = str(argument_descriptor.name)
415			if argument_descriptor.required and argument_name not in arguments:
416			raise IllegalArgumentError('Missing required argument "%s"' % argument_name)
417
418			# Serializer argument value
419			serialized_arguments[argument_descriptor.name] = \
420			argument_descriptor.serializer_class().to_api(arguments.get(argument_name))
421
422			return serialized_arguments
423
424			def _report_method_call(self, method, arguments):
425			"""
426			Internal method to print info about a method call
427			:param method: The name oft hem ethod to call
428			:type method: str\|unicode
429			:param arguments: A dictionary of arguments that will be passed to the method
430			:type: arguments: dict
431			:return:
432			"""
433			if self.verbose:
434			print('===> %s(%s)' % (method, ', '.join(['%s=%s' % (key, value) for (key, value) in arguments.items()])))
435
436			def _call_api(self, api_class, **kwargs):
437			"""
438			Internal method used to perform api calls
439			:param api_class:
440			:type api_class: py:class:API
441			:param kwargs:
442			:type kwargs: dict
443			:rtype: object
444			:return: Returns return value of the XMLRPC Method call
445			"""
446			# Validate api_class
447			assert issubclass(api_class, API)
448
449			# Create api instance
450			api = api_class()
451
452			# Copy arguments
453			arguments = copy.copy(kwargs)
454
455			# Transform arguments
456			arguments = api.transform_arguments(**arguments)
457
458			# Validate and serialize arguments
459			arguments = self._validate_and_serialize_arguments(api, arguments)
460
461			# Retrieve method
462			method = getattr(self.proxy, str(api.method))
463
464			# Print method call with arguments
465			self._report_method_call(api.method, arguments)
466
467			# Perform call
468			if not api.authenticate:
469			response = method(arguments)
470			else:
471			response = method(self.auth, arguments)
472
473			# Validate response
474			api.validate_response(api, arguments, response)
475
476			# Transform response
477			result = api.transform_response(api, arguments, response)
478
479			return result
480

dirkmoors / pyejabberd

GitHub Access Token became invalid

src.pyejabberd.EjabberdAPIClient.__init__() B last analyzed 2016-03-09 14:09 UTC

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like

src.pyejabberd.EjabberdAPIClient.init() B
last analyzed 2016-03-09 14:09 UTC