EjabberdAPIClient.get_instance() - Code Metrics - Inspection of "add privacy_set()" - dirkmoors/pyejabberd - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#22)

unknown

created 2016-10-06 09:09 UTC

EjabberdAPIClient.get_instance() C

↳ Parent: EjabberdAPIClient

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	46
rs	5.5
cc	7

# -*- 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 check_password(self, user, host, password):
        """
        Check if a password is correct.
        :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.CheckPassword, 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 stats(self, name):
        """
        Get statistical value

        :param name: registeredusers onlineusers onlineusersnode uptimeseconds
        :type name: str|unicode
        :rtype: int
        :return: statistical value
        """
        return self._call_api(definitions.Stats, name=name)

    def stats_host(self, name, host):
        """
        Get statistical value for this host

        :param name: registeredusers onlineusers onlineusersnode uptimeseconds
        :type name: str|unicode
        :param host: The XMPP_DOMAIN
        :type host: str|unicode
        :rtype: int
        :return: statistical value
        """
        return self._call_api(definitions.StatsHost, name=name, host=host)

    def send_message(self, type, _from, to, subject, body):
        """
        Send a message to a local or remote bare of full JID

        :param type: chat | headline | normal
        :type type: str|unicode
        :param _from: from JID
        :type _from: str|unicode
        :param to: to JID
        :type to: str|unicode
        :param subject: message subject
        :type subject: str|unicode
        :param body: message body
        :type body: str|unicode
        :rtype: int
        :return: result code
        """
        return self._call_api(definitions.SendMessage,
                              type=type, _from=_from, to=to,
                              subject=subject, body=body)

    def privacy_set(self, user, host, xmlquery):
        """
        Send a IQ set privacy stanza for a local account

        :param user: The Username
        :type user: str|unicode
        :param host: The XMPP_DOMAIN
        :type host: str|unicode
        :param xmlquery: xmlquery
        :type xmlquery: str|unicode
        :rtype: int
        :return: responce code
        """
        return self._call_api(definitions.PrivacySet,
                              user=user, host=host, xmlquery=xmlquery)

    def get_last(self, user, host):
        return self._call_api(definitions.GetLast,
                              user=user, host=host)

    def ban_account(self, user, host, reason):
        return self._call_api(definitions.BanAccount,
                              user=user, host=host, reason=reason)

    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
            # we need lstrip underscore for special argument: _from
            serialized_arguments[argument_descriptor.name.lstrip('_')] = \
                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 check_password(self, user, host, password):
200			"""
201			Check if a password is correct.
202			:param user: The username for the user we want to check the password for
203			:type user: str\|unicode
204			:param host: The XMPP_DOMAIN
205			:type host: str\|unicode
206			:param password: The password we want to check for the user
207			:type password: str\|unicode
208			:rtype: bool
209			:return: A boolean indicating if the given password matches the user's password
210			"""
211			return self._call_api(definitions.CheckPassword, user=user, host=host, password=password)
212
213			def set_nickname(self, user, host, nickname):
214			"""
215			Set nickname in a user's vCard
216			:param user: The username for the user we want to set the nickname to
217			:type user: str\|unicode
218			:param host: The XMPP_DOMAIN
219			:type host: str\|unicode
220			:param nickname: The nickname to assign to the user
221			:type nickname: str\|unicode
222			:rtype: bool
223			:return: A boolean indicating nickname was assigned successfully
224			"""
225			return self._call_api(definitions.SetNickname, user=user, host=host, nickname=nickname)
226
227			def connected_users(self):
228			"""
229			List all established sessions
230			:rtype: list
231			:return: a list of dictionaries containing user jids
232			"""
233			return self._call_api(definitions.ConnectedUsers)
234
235			def connected_users_info(self):
236			"""
237			List all established sessions and their information
238			:rtype: list
239			:return: a list of dictionaries containing user info
240			"""
241			return self._call_api(definitions.ConnectedUsersInfo)
242
243			def connected_users_number(self):
244			"""
245			Get the number of established sessions
246			:rtype: int
247			:return: number of established user sessions
248			"""
249			return self._call_api(definitions.ConnectedUsersNumber)
250
251			def user_sessions_info(self, user, host):
252			"""
253			Get information about all sessions of a user
254			:param user: The username for the user we want info for
255			:type user: str\|unicode
256			:param host: The XMPP_DOMAIN
257			:type host: str\|unicode
258			:rtype: list
259			:return: list of information of sessions for a user
260			"""
261			return self._call_api(definitions.UserSessionInfo, user=user, host=host)
262
263			def muc_online_rooms(self, host=None):
264			"""
265			List existing rooms ('global' to get all vhosts)
266			:param host: The XMPP_DOMAIN
267			:type host: str\|unicode
268			:rtype: Iterable
269			:return: A list of online rooms in the format 'name@service'
270			"""
271			host = host or 'global'
272			return self._call_api(definitions.MucOnlineRooms, host=host)
273
274			def create_room(self, name, service, host):
275			"""
276			Create a MUC room name@service in host
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 created successfully
285			"""
286			return self._call_api(definitions.CreateRoom, name=name, service=service, host=host)
287
288			def destroy_room(self, name, service, host):
289			"""
290			Destroy 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			:param host: The XMPP_DOMAIN
296			:type host: str\|unicode
297			:rtype: bool
298			:return: A boolean indicating whether the room has been destroyed successfully
299			"""
300			return self._call_api(definitions.DestroyRoom, name=name, service=service, host=host)
301
302			def get_room_options(self, name, service):
303			"""
304			Get options from a MUC room
305			:param name: The name for the room
306			:type name: str\|unicode
307			:param service: The MUC service name (e.g. "conference")
308			:type service: str\|unicode
309			:rtype: dict
310			:return: A dict containing the room options
311			"""
312			return self._call_api(definitions.GetRoomOptions, name=name, service=service)
313
314			def change_room_option(self, name, service, option, value):
315			"""
316			Change an option in a MUC room
317			:param name: The name for the room
318			:type name: str\|unicode
319			:param service: The MUC service name (e.g. "conference")
320			:type service: str\|unicode
321			:param option: The option to change
322			:type option: muc.enums.MUCRoomOption
323			:param value: The new value
324			:type value: str\|unicode\|int\|bool
325			:rtype: bool
326			:return: A boolean indicating whether the room option has been changed successfully
327			"""
328			return self._call_api(definitions.ChangeRoomOption, name=name, service=service, option=option, value=value)
329
330			def set_room_affiliation(self, name, service, jid, affiliation):
331			"""
332			Change an affiliation for a user in a MUC room
333			:param name:The name for the room
334			:type name: str\|unicode
335			:param service: The MUC service name (e.g. "conference")
336			:type service: str\|unicode
337			:param jid: The jabber id for the user you want to change the affiliation for
338			:type jid: str\|unicode
339			:param affiliation: The affiliation to the room
340			:type affiliation: muc.enums.Affiliation
341			:rtype: list
342			:return: A list containing dictionaries containing affiliation info
343			"""
344			return self._call_api(definitions.SetRoomAffiliation, name=name, service=service, jid=jid,
345			affiliation=affiliation)
346
347			def get_room_affiliations(self, name, service):
348			"""
349			Get the affiliations for a MUC room
350			:param name:The name for the room
351			:type name: str\|unicode
352			:param service: The MUC service name (e.g. "conference")
353			:type service: str\|unicode
354			:return:
355			"""
356			return self._call_api(definitions.GetRoomAffiliations, name=name, service=service)
357
358			def add_rosteritem(self, localuser, localserver, user, server, nick, group, subs):
359			"""
360			Add an item to a user's roster
361
362			:param localuser: The username of user we are going to add a contact to
363			:type localuser: str\|unicode
364			:param localserver: The XMPP_DOMAIN
365			:type localserver: str\|unicode
366			:param user: The contact we are going to add to the user's roster
367			:type user: str\|unicode
368			:param server: The XMPP_DOMAIN
369			:type server: str\|unicode
370			:param nick: Nickname of the contact
371			:type nick: str\|unicode
372			:param group: To what contact group the contact goes to
373			:type group: str\|unicode
374			:param subs: The type of subscription
375			:type subs: str\|unicode
376			:return:
377			"""
378			return self._call_api(definitions.AddRosterItem,
379			localuser=localuser, localserver=localserver,
380			user=user, server=server,
381			nick=nick, group=group, subs=subs)
382
383			def delete_rosteritem(self, localuser, localserver, user, server):
384			"""
385			Delete an item from a user's roster
386
387			:param localuser: The username of user we are going to delete a contact from
388			:type localuser: str\|unicode
389			:param localserver: The XMPP_DOMAIN
390			:type localserver: str\|unicode
391			:param user: The contact we are going to delete from the user's roster
392			:type user: str\|unicode
393			:param server: The XMPP_DOMAIN
394			:type server: str\|unicode
395			:return:
396			"""
397			return self._call_api(definitions.DeleteRosterItem, localuser=localuser, localserver=localserver, user=user, server=server)
398
399			def get_roster(self, user, host):
400			"""
401			Get roster of a user
402
403			:param user: The username of the user we want contact information for
404			:type user: str\|unicode
405			:param host: The XMPP_DOMAIN
406			:type host: str\|unicode
407			:rtype: Iterable
408			:return: A list of user's contacts
409			"""
410			return self._call_api(definitions.GetRoster, user=user, host=host)
411
412			def stats(self, name):
413			"""
414			Get statistical value
415
416			:param name: registeredusers onlineusers onlineusersnode uptimeseconds
417			:type name: str\|unicode
418			:rtype: int
419			:return: statistical value
420			"""
421			return self._call_api(definitions.Stats, name=name)
422
423			def stats_host(self, name, host):
424			"""
425			Get statistical value for this host
426
427			:param name: registeredusers onlineusers onlineusersnode uptimeseconds
428			:type name: str\|unicode
429			:param host: The XMPP_DOMAIN
430			:type host: str\|unicode
431			:rtype: int
432			:return: statistical value
433			"""
434			return self._call_api(definitions.StatsHost, name=name, host=host)
435
436			def send_message(self, type, _from, to, subject, body):
437			"""
438			Send a message to a local or remote bare of full JID
439
440			:param type: chat \| headline \| normal
441			:type type: str\|unicode
442			:param _from: from JID
443			:type _from: str\|unicode
444			:param to: to JID
445			:type to: str\|unicode
446			:param subject: message subject
447			:type subject: str\|unicode
448			:param body: message body
449			:type body: str\|unicode
450			:rtype: int
451			:return: result code
452			"""
453			return self._call_api(definitions.SendMessage,
454			type=type, _from=_from, to=to,
455			subject=subject, body=body)
456
457			def privacy_set(self, user, host, xmlquery):
458			"""
459			Send a IQ set privacy stanza for a local account
460
461			:param user: The Username
462			:type user: str\|unicode
463			:param host: The XMPP_DOMAIN
464			:type host: str\|unicode
465			:param xmlquery: xmlquery
466			:type xmlquery: str\|unicode
467			:rtype: int
468			:return: responce code
469			"""
470			return self._call_api(definitions.PrivacySet,
471			user=user, host=host, xmlquery=xmlquery)
472
473			def get_last(self, user, host):
474			return self._call_api(definitions.GetLast,
475			user=user, host=host)
476
477			def ban_account(self, user, host, reason):
478			return self._call_api(definitions.BanAccount,
479			user=user, host=host, reason=reason)
480
481			def _validate_and_serialize_arguments(self, api, arguments):
482			"""
483			Internal method to validate and serialize arguments
484			:param api: An instance of an API class
485			:param arguments: A dictionary of arguments that will be passed to the method
486			:type arguments: dict
487			:rtype: dict
488			:return: The serialized arguments
489			"""
490			serialized_arguments = {}
491
492			for i in range(len(api.arguments)):
493			argument_descriptor = api.arguments[i]
494			assert isinstance(argument_descriptor, APIArgument)
495
496			# Validate argument presence
497			argument_name = str(argument_descriptor.name)
498			if argument_descriptor.required and argument_name not in arguments:
499			raise IllegalArgumentError('Missing required argument "%s"' % argument_name)
500
501			# Serializer argument value
502			# we need lstrip underscore for special argument: _from
503			serialized_arguments[argument_descriptor.name.lstrip('_')] = \
504			argument_descriptor.serializer_class().to_api(arguments.get(argument_name))
505
506			return serialized_arguments
507
508			def _report_method_call(self, method, arguments):
509			"""
510			Internal method to print info about a method call
511			:param method: The name oft hem ethod to call
512			:type method: str\|unicode
513			:param arguments: A dictionary of arguments that will be passed to the method
514			:type: arguments: dict
515			:return:
516			"""
517			if self.verbose:
518			print('===> %s(%s)' % (method, ', '.join(['%s=%s' % (key, value) for (key, value) in arguments.items()])))
519
520			def _call_api(self, api_class, **kwargs):
521			"""
522			Internal method used to perform api calls
523			:param api_class:
524			:type api_class: py:class:API
525			:param kwargs:
526			:type kwargs: dict
527			:rtype: object
528			:return: Returns return value of the XMLRPC Method call
529			"""
530			# Validate api_class
531			assert issubclass(api_class, API)
532
533			# Create api instance
534			api = api_class()
535
536			# Copy arguments
537			arguments = copy.copy(kwargs)
538
539			# Transform arguments
540			arguments = api.transform_arguments(**arguments)
541
542			# Validate and serialize arguments
543			arguments = self._validate_and_serialize_arguments(api, arguments)
544
545			# Retrieve method
546			method = getattr(self.proxy, str(api.method))
547
548			# Print method call with arguments
549			self._report_method_call(api.method, arguments)
550
551			# Perform call
552			if not api.authenticate:
553			response = method(arguments)
554			else:
555			response = method(self.auth, arguments)
556
557			# Validate response
558			api.validate_response(api, arguments, response)
559
560			# Transform response
561			result = api.transform_response(api, arguments, response)
562
563			return result
564

dirkmoors / pyejabberd

GitHub Access Token became invalid

Pull Request — master (#22)

EjabberdAPIClient.get_instance() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like