PoC that implements asynchronous WebSockets and Http

Pipfile

@@ -7,6 +7,8 @@ name = "pypi"
 requests = "2.25.1"
 websocket-client= "0.57.0"
 urllib3 = "1.26.4"
+websockets = "~=8.1"
+aiohttp = "3.7.4.post0"
 
 [dev-packages]
 pre-commit = "2.10.1"

livechat/customer/__init__.py

@@ -1,3 +1,5 @@
 #pylint: disable=C0114
+from livechat.customer.rtm.async_client import AsyncCustomerRTM
 from livechat.customer.rtm.client import CustomerRTM
+from livechat.customer.web.async_client import AsyncCustomerWeb
 from livechat.customer.web.client import CustomerWeb

livechat/customer/rtm/async_client.py

@@ -0,0 +1,159 @@
+''' Async Customer RTM client implementation. '''
+
+# pylint: disable=W0613,W0622,C0103,R0913,R0903,W0107
+
+import random
+from abc import ABCMeta
+
+from livechat.utils.async_ws_client import AsyncWebsocketClient
+from livechat.utils.helpers import prepare_payload
+
+
+class AsyncCustomerRTM:
+    ''' Main class that gets specific client. '''
+    @staticmethod
+    def get_client(license_id: int = None,
+                   version: str = '3.3',
+                   base_url: str = 'api.livechatinc.com'):
+        ''' Returns client for specific Customer RTM version.
+
+            Args:
+                license_id (int): License ID.
+                version (str): API's version. Defaults to `3.3`.
+                base_url (str): API's base url. Defaults to `api.livechatinc.com`.
+
+            Returns:
+                CustomerRTMInterface: API client object for specified version.
+
+            Raises:
+                ValueError: If the specified version does not exist.
+        '''
+        client = {
+            '3.3': CustomerRTM33(license_id, version, base_url),
+            '3.4': CustomerRTM34(license_id, version, base_url)
+        }.get(version)
+        if not client:
+            raise ValueError('Provided version does not exist.')
+        return client
+
+
+class CustomerRTMInterface(metaclass=ABCMeta):
+    ''' CustomerRTM interface class. '''
+    def __init__(self, license_id, version, url):
+        if not license_id or not isinstance(license_id, int):
+            raise ValueError(
+                'Pipe was not opened. Something`s wrong with your `license_id`.'
+            )
+        self.ws = AsyncWebsocketClient(
+            url=
+            f'wss://{url}/v{version}/customer/rtm/ws?license_id={license_id}')
+
+    async def open_connection(self, origin: dict = None) -> None:
+        ''' Opens WebSocket connection.
+
+            Args:
+                origin (dict): Specifies origin while creating websocket connection.
+        '''
+        if origin:
+            await self.ws.open(origin=origin)
+        else:
+            await self.ws.open()
+
+    async def close_connection(self) -> None:
+        ''' Closes WebSocket connection. '''
+        await self.ws.close()
+
+    async def receive_message(self) -> dict:
+        ''' Receives message from WebSocket. '''
+        return await self.ws.receive()
+
+    async def login(self,
+                    request_id: str = str(random.randint(1, 9999999999)),
+                    token: str = None,
+                    payload: dict = None) -> dict:
+        ''' Logs in customer.
+
+            Args:
+                request_id (str): unique id of the request.
+                    If not provided, a random id will be generated.
+                token (str): OAuth token from the Customer's account.
+                payload (dict): Custom payload to be used as request's data.
+                        It overrides all other parameters provided for the method.
+
+            Returns:
+                dict: request that will be sent out.
+        '''
+        if payload is None:
+            payload = prepare_payload(locals())
+        request = {
+            'action': 'login',
+            'payload': payload,
+            'request_id': request_id
+        }
+        await self.ws.send(request)
+        return request
+
+    async def start_chat(self,
+                         request_id: str = str(random.randint(1, 9999999999)),
+                         chat: dict = None,
+                         active: bool = None,
+                         continuous: bool = None,
+                         payload: dict = None) -> dict:
+        ''' Starts a chat.
+
+            Args:
+                request_id (str): unique id of the request.
+                    If not provided, a random id will be generated.
+                chat (dict): Chat object.
+                active (bool): When set to False, creates an inactive thread; default: True.
+                continuous (bool): Starts chat as continuous (online group is not required); default: False.
+                payload (dict): Custom payload to be used as request's data.
+                        It overrides all other parameters provided for the method.
+
+            Returns:
+                dict: request that will be sent out.
+        '''
+        if payload is None:
+            payload = prepare_payload(locals())
+        request = {
+            'action': 'start_chat',
+            'payload': payload,
+            'request_id': request_id
+        }
+        await self.ws.send(request)
+        return request
+
+    async def deactivate_chat(self,
+                              request_id: str = str(
+                                  random.randint(1, 9999999999)),
+                              id: str = None,
+                              payload: dict = None) -> dict:
+        ''' Deactivates a chat by closing the currently open thread.
+
+            Args:
+                request_id (str): unique id of the request.
+                    If not provided, a random id will be generated.
+                id (str): Chat ID to deactivate.
+                payload (dict): Custom payload to be used as request's data.
+                    It overrides all other parameters provided for the method.
+
+            Returns:
+                dict: request that will be sent out.
+        '''
+        if payload is None:
+            payload = prepare_payload(locals())
+        request = {
+            'action': 'deactivate_chat',
+            'payload': payload,
+            'request_id': request_id
+        }
+        await self.ws.send(request)
+        return request
+
+
+class CustomerRTM33(CustomerRTMInterface):
+    ''' Customer RTM version 3.3 class. '''
+
+
+class CustomerRTM34(CustomerRTMInterface):
+    ''' Customer RTM version 3.4 class. '''

livechat/customer/web/async_client.py

@@ -0,0 +1,156 @@
+''' Async Customer Web client implementation. '''
+
+# pylint: disable=W0613,R0913,W0622,C0103
+
+from abc import ABCMeta
+
+import aiohttp
+
+from livechat.utils.helpers import prepare_payload
+
+
+# pylint: disable=R0903
+class AsyncCustomerWeb:
+    ''' Allows retrieval of client for specific Customer Web
+        API version. '''
+    @staticmethod
+    def get_client(license_id: int,
+                   access_token: str,
+                   version: str = '3.3',
+                   base_url: str = 'api.livechatinc.com'):
+        ''' Returns client for specific API version.
+
+            Args:
+                token (str): Full token with type (Bearer/Basic) that will be
+                                used as `Authorization` header in requests to API.
+                version (str): API's version. Defaults to `3.3`.
+                base_url (str): API's base url. Defaults to `api.livechatinc.com`.
+
+            Returns:
+                API client object for specified version based on
+                `CustomerWebApiInterface`.
+
+            Raises:
+                ValueError: If the specified version does not exist.
+        '''
+        client = {
+            '3.3': CustomerWeb33(license_id, access_token, version, base_url),
+            '3.4': CustomerWeb34(license_id, access_token, version, base_url)
+        }.get(version)
+        if not client:
+            raise ValueError('Provided version does not exist.')
+        return client
+
+
+class CustomerWebInterface(metaclass=ABCMeta):
+    ''' Main class containing API methods. '''
+    def __init__(self, license_id: int, access_token: str, version: str,
+                 base_url: str):
+        self.api_url = f'https://{base_url}/v{version}/customer/action'
+        self.session = aiohttp.ClientSession()
+        self.session.headers.update({'Authorization': access_token})
+        self.license_id = str(license_id)
+
+    def modify_header(self, header: dict) -> None:
+        ''' Modifies provided header in session object.
+
+            Args:
+                header (dict): Header which needs to be modified.
+        '''
+        self.session.headers.update(header)
+
+    def remove_header(self, key: str) -> None:
+        ''' Removes provided header from session object.
+
+            Args:
+                key (str): Key which needs to be removed from the header.
+        '''
+        if key in self.session.headers:
+            del self.session.headers[key]
+
+    def get_headers(self) -> dict:
+        ''' Returns current header values in session object.
+
+            Returns:
+                dict: Response which presents current header values in session object.
+        '''
+        return dict(self.session.headers)
+
+    async def close_session(self) -> None:
+        ''' Closes active session. '''
+        if not self.session.closed:
+            await self.session.close()
+
+    async def start_chat(self,
+                         chat: dict = None,
+                         active: bool = None,
+                         continuous: bool = None,
+                         payload: dict = None) -> aiohttp.ClientResponse:
+        ''' Starts a chat.
+
+            Args:
+                chat (dict): Dict containing chat properties, access and thread.
+                active (bool): When set to False, creates an inactive thread; default: True.
+                continuous (bool): Starts chat as continuous (online group is not required); default: False.
+                payload (dict): Custom payload to be used as request's data.
+                                It overrides all other parameters provided for the method.
+
+            Returns:
+                aiohttp.ClientResponse: The Response object from `aiohttp` library,
+                    which contains a server’s response to an HTTP request.
+        '''
+        if payload is None:
+            payload = prepare_payload(locals())
+        return await self.session.post(
+            f'{self.api_url}/start_chat?license_id={self.license_id}',
+            json=payload)
+
+    async def get_chat(self,
+                       chat_id: str = None,
+                       thread_id: str = None,
+                       payload: dict = None) -> aiohttp.ClientResponse:
+        ''' Returns a thread that the current Customer has access to in a given chat.
+
+            Args:
+                chat_id (str): ID of the chat for which thread is to be returned.
+                thread_id (str): ID of the thread to show. Default: the latest thread (if exists)
+                payload (dict): Custom payload to be used as request's data.
+                                It overrides all other parameters provided for the method.
+            Returns:
+                aiohttp.ClientResponse: The Response object from `aiohttp` library,
+                    which contains a server’s response to an HTTP request.
+        '''
+        if payload is None:
+            payload = prepare_payload(locals())
+        return await self.session.post(
+            f'{self.api_url}/get_chat?license_id={self.license_id}',
+            json=payload)
+
+    async def deactivate_chat(self,
+                              id: str = None,
+                              payload: dict = None) -> aiohttp.ClientResponse:
+        ''' Deactivates a chat by closing the currently open thread.
+            Sending messages to this thread will no longer be possible.
+
+            Args:
+                id (str): ID of chat to be deactivated.
+                payload (dict): Custom payload to be used as request's data.
+                                It overrides all other parameters provided for the method.
+
+            Returns:
+                aiohttp.ClientResponse: The Response object from `aiohttp` library,
+                    which contains a server’s response to an HTTP request.
+        '''
+        if payload is None:
+            payload = prepare_payload(locals())
+        return await self.session.post(
+            f'{self.api_url}/deactivate_chat?license_id={self.license_id}',
+            json=payload)
+
+
+class CustomerWeb33(CustomerWebInterface):
+    ''' Customer API version 3.3 class. '''
+
+
+class CustomerWeb34(CustomerWebInterface):
+    ''' Customer API version 3.4 class. '''

livechat/utils/async_ws_client.py

@@ -0,0 +1,61 @@
+'''
+Async client for WebSocket connections.
+'''
+
+# pylint: disable=R1702
+
+import json
+import logging
+
+import websockets
+from websockets import WebSocketException
+
+
+class AsyncWebsocketClient:
+    ''' WebSocket asynchronous client based on websockets module. '''
+    def __init__(self, url: str):
+        self.url = url
+        self.websocket = None
+        logging.basicConfig()
+        self.logger = logging.getLogger()
+        self.logger.setLevel(logging.INFO)
+
+    async def open(self, ping_interval: int = 10, origin: dict = None) -> None:
+        ''' Open WebSocket connection.
+                Args:
+                    ping_interval(int): Specifies how often ping the server.
+                        Default: 10 seconds.
+                    origin (dict): Specifies origin while creating websocket connection.
+        '''
+        try:
+            if origin:
+                self.websocket = await websockets.connect(
+                    self.url, ping_interval=ping_interval, origin=origin)
+            else:
+                self.websocket = await websockets.connect(
+                    self.url, ping_interval=ping_interval)
+        except WebSocketException as exception:
+            self.logger.critical(f'WebSocket Exception: {exception}')
+
+    async def close(self) -> None:
+        ''' Close WebSocket connection. '''
+        if self.websocket is not None:
+            await self.websocket.close()
+            self.websocket = None
+
+    async def send(self, request: dict) -> None:
+        ''' Send request via WebSocket.
+                Args:
+                    request (dict): Dictionary which is being converted to payload.
+        '''
+        if self.websocket is not None:
+            await self.websocket.send(json.dumps(request))
+
+    async def receive(self) -> dict:
+        ''' Receive data from WebSocket.
+                Returns:
+                    dict: Dictionary containing response from WebSocket.
+        '''
+        if self.websocket is not None:
+            results = await self.websocket.recv()
+            return json.loads(results)

requirements.txt

@@ -1,3 +1,5 @@
+aiohttp==3.7.4.post0
 pytest==6.2.2
 requests==2.25.1
 websocket-client==0.57.0
+websockets==8.1