Added async option

Author: amar-jaiswal-haptik

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2020 interakt.ai
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

setup.py

@@ -23,7 +23,8 @@
     readme = f.read()
 
 install_requires = [
-    "requests>=2.20,<3.0"
+    "requests>=2.20,<3.0",
+    "backoff==1.10.0"
 ]
 
 tests_require = [
@@ -36,16 +37,14 @@
 
 setup(
     name='interakt-track-python',
-    packages=['track'],
+    packages=['track', 'track.tests'],
     version=VERSION,
-    # TODO : Add repository url
-    url='',
+    url='https://github.com/interakt/track-python',
+    download_url=f'https://github.com/interakt/track-python/archive/v{VERSION}.tar.gz',
     author="Amar Jaiswal",
     author_email="amar.j@cawstudios.com",
     maintainer="interakt.ai",
-    # Chose a license from here: https://help.github.com/articles/licensing-a-repository
-    # TODO : Add License
-    license='',
+    license='MIT License',
     description='The easy way to integrate track apis for interakt',
     keywords=['INTERAKT', 'KIWI'],
     install_requires=install_requires,
@@ -57,12 +56,8 @@
         'Development Status :: 3 - Alpha',
         'Intended Audience :: Developers',
         'Topic :: Software Development :: Build Tools',
-        # TODO: Add license
-        # 'License :: OSI Approved :: MIT License',   # Again, pick a license
+        'License :: OSI Approved :: MIT License',
         "Operating System :: OS Independent",
-        'Programming Language :: Python :: 3',
-        'Programming Language :: Python :: 3.4',
-        'Programming Language :: Python :: 3.5',
         'Programming Language :: Python :: 3.6',
         'Programming Language :: Python :: 3.7',
         'Programming Language :: Python :: 3.8',

track/__init__.py

@@ -6,27 +6,55 @@
 write_key = None
 default_client = None
 host = None
-sync_mode = True
+sync_mode = False
 debug = False
 timeout = 10
+max_retries = 3
+on_error = None
+max_queue_size = 10000
 
 
 def identify(user_id=None, country_code='+91', phone_number=None, traits={}):
     """Send an identify call for customer"""
-    return _proxy('identify', user_id=user_id, country_code=country_code, phone_number=phone_number, traits=traits)
+    return _proxy('identify', user_id=user_id, country_code=country_code,
+                  phone_number=phone_number, traits=traits)
 
 
 def event(user_id=None, event=None, traits={}):
     """Send an event track call."""
     return _proxy('event', user_id=user_id, event=event, traits=traits)
 
 
+def flush():
+    """Tell the client to flush."""
+    _proxy('flush')
+
+
+def join():
+    """Block program until the client clears the queue"""
+    _proxy('join')
+
+
+def shutdown():
+    """Flush all messages and cleanly shutdown the client"""
+    _proxy('flush')
+    _proxy('join')
+
+
 def _proxy(method, *args, **kwargs):
     """Create an analytics client if one doesn't exist and send to it."""
     global default_client
     if not default_client:
         default_client = Client(
-            write_key, host=host, debug=debug, sync_mode=sync_mode, timeout=timeout)
+            write_key=write_key,
+            host=host,
+            debug=debug,
+            sync_mode=sync_mode,
+            timeout=timeout,
+            max_queue_size=max_queue_size,
+            on_error=on_error,
+            max_retries=max_retries
+        )
 
     fn = getattr(default_client, method)
     return fn(*args, **kwargs)

track/client.py

@@ -1,25 +1,46 @@
-from track.utils import require, stringify
-from track.request import post
-from track.const import ApiPaths
 import logging
 import numbers
+
+from track.const import ApiPaths
+from track.consumer import Consumer
+from track.request import post
+from track.utils import require, stringify
+
+try:
+    import queue
+except ImportError:
+    import Queue as queue
+
 ID_TYPES = (numbers.Number, str)
 
 
 class Client(object):
     logger = logging.getLogger('interakt')
 
-    def __init__(self, write_key=None, host=None, debug=False, sync_mode=True, timeout=10):
+    def __init__(self, write_key=None, host=None, debug=False,
+                 sync_mode=True, timeout=10, max_queue_size=10000, on_error=None, max_retries=3):
         """Create a new interakt client"""
         require('write_key', write_key, str)
+
+        self.queue = queue.Queue(maxsize=max_queue_size)
         self.write_key = write_key
         self.debug = debug
         self.host = host
         self.sync_mode = sync_mode
         self.timeout = timeout
+        self.on_error = on_error
         if debug:
             self.logger.setLevel(logging.DEBUG)
 
+        if sync_mode:
+            self.consumer = None
+        else:
+            self.consumer = Consumer(
+                queue=self.queue, write_key=write_key,
+                host=host, on_error=on_error, retries=max_retries,
+                timeout=timeout
+            )
+
     def identify(self, user_id=None, country_code='+91', phone_number=None, traits={}):
         """Tie a user to their actions and record traits about them."""
         if not user_id and not phone_number:
@@ -29,26 +50,64 @@ def identify(self, user_id=None, country_code='+91', phone_number=None, traits={
         if phone_number:
             require('phone_number', phone_number, str)
         require('traits', traits, dict)
-        msg = {
+        body = {
             'userId': stringify(val=user_id),
             'countryCode': country_code,
             'phoneNumber': phone_number,
             'traits': traits
         }
-        return self.__send_request(path=ApiPaths.Identify.value, msg=msg)
+        return self.__queue_request(path=ApiPaths.Identify.value, body=body)
 
     def event(self, user_id=None, event=None, traits={}):
         """To record user events"""
         traits = traits or {}
         require('user_id', user_id, ID_TYPES)
         require('traits', traits, dict)
         require('event', event, str)
-        msg = {
+        body = {
             'userId': stringify(val=user_id),
             'event': event,
             'traits': traits
         }
-        return self.__send_request(path=ApiPaths.Event.value, msg=msg)
+        return self.__queue_request(path=ApiPaths.Event.value, body=body)
+
+    def flush(self):
+        """Forces a flush from the internal queue to the server"""
+        queue = self.queue
+        size = queue.qsize()
+        queue.join()
+        # Note that this message may not be precise, because of threading.
+        self.logger.debug('successfully flushed about %s items.', size)
 
-    def __send_request(self, path, msg):
-        return post(self.write_key, host=self.host, path=path, body=msg, timeout=self.timeout)
+    def join(self):
+        """Ends the consumer thread once the queue is empty.
+        Blocks execution until finished
+        """
+        self.consumer.pause()
+        try:
+            self.consumer.join()
+        except RuntimeError:
+            # consumer thread has not started
+            pass
+
+    def shutdown(self):
+        """Flush all messages and cleanly shutdown the client"""
+        self.flush()
+        self.join()
+
+    def __queue_request(self, path, body):
+        # Directly call api in sync mode and return response
+        if self.sync_mode:
+            return post(self.write_key, host=self.host, path=path, body=body, timeout=self.timeout)
+
+        queue_msg = {
+            'path': path,
+            'body': body
+        }
+        try:
+            self.queue.put(queue_msg, block=False)
+            self.logger.debug(f'Enqueued msg for {path}')
+            return True, queue_msg
+        except queue.Full:
+            self.logger.warning('track-python queue is full')
+            return False, queue_msg

track/consumer.py

@@ -0,0 +1,84 @@
+import logging
+from threading import Thread
+from track.request import post, APIError
+import backoff
+
+try:
+    from queue import Empty
+except ImportError:
+    from Queue import Empty
+
+
+class Consumer(Thread):
+    """Consumes the messages from the client's queue."""
+    logger = logging.getLogger('interakt')
+
+    def __init__(self, queue, write_key, host=None,
+                 on_error=None, retries=3, timeout=10):
+        Thread.__init__(self)
+        self.running = True
+        self.queue = queue
+        self.write_key = write_key
+        self.host = host
+        self.on_error = on_error
+        self.retries = retries
+        self.timeout = timeout
+
+    def run(self):
+        """Runs the consumer."""
+        self.logger.debug('consumer is running...')
+        while self.running:
+            self.upload()
+
+        self.logger.debug('consumer exited.')
+
+    def pause(self):
+        """Pause the consumer."""
+        self.running = False
+
+    def upload(self):
+        queue = self.queue
+        queue_msg = None
+        try:
+            queue_msg = queue.get(block=True)
+        except Empty:
+            self.logger.debug("queue is empty now")
+
+        if not queue_msg:
+            self.logger.debug("Nothing left in queue exiting")
+            return False
+        try:
+            self.request(queue_msg=queue_msg)
+            success = True
+        except Exception as e:
+            self.logger.error(f"Error uploading: {e}")
+            success = False
+            if self.on_error:
+                self.on_error(e, queue_msg)
+        finally:
+            self.queue.task_done()
+        return success
+
+    def request(self, queue_msg):
+        """Attempt to upload the queue_msg and retry before raising an error """
+
+        def fatal_exception(exc):
+            if isinstance(exc, APIError):
+                # retry on server errors and client errors
+                # with 429 status code (rate limited),
+                # don't retry on other client errors
+                return (400 <= exc.status < 500) and exc.status != 429
+            else:
+                # retry on all other errors (eg. network)
+                return False
+
+        @backoff.on_exception(
+            backoff.expo,
+            Exception,
+            max_tries=self.retries + 1,
+            giveup=fatal_exception)
+        def send_request():
+            post(write_key=self.write_key, host=self.host,
+                 path=queue_msg['path'], body=queue_msg['body'], timeout=self.timeout)
+
+        send_request()

track/request.py

@@ -22,6 +22,26 @@ def post(write_key, host=None, path=None, body=None, timeout=10):
     logger.debug(f'Making request: {body}')
     response = _session.post(url=url, headers=headers,
                              json=body, timeout=timeout)
-    payload = response.json()
-    logger.debug(f'Received response: {payload}')
-    return response
+    if response.status_code == 200:
+        logger.debug("Data uploaded successfully")
+        return response
+
+    try:
+        payload = response.json()
+        logger.debug(f'Received response: {payload}')
+        raise APIError(payload.get("result"),
+                       response.status_code, payload.get("message"))
+    except ValueError:
+        raise APIError('Unknown', response.status_code, response.text)
+
+
+class APIError(Exception):
+
+    def __init__(self, status, code, message):
+        self.message = message
+        self.status = status
+        self.code = code
+
+    def __str__(self):
+        msg = "[Segment] {0}: {1} ({2})"
+        return msg.format(self.code, self.message, self.status)

track/tests/__init__.py

@@ -1 +1 @@
-VERSION = '0.0.4'
+VERSION = '0.0.4-alpha'