feat: unit test for client module

.github/workflows/publish-to-pypi.yml

@@ -0,0 +1,35 @@
+name: Publish to PyPI
+
+on: workflow_dispatch
+
+jobs:
+  authorize:
+    name: Authorize
+    runs-on: ubuntu-latest
+    steps:
+      - name: ${{ github.actor }} permission check to do a release
+        uses: "lannonbr/repo-permission-check-action@2.0.2"
+        with:
+          permission: "write"
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+
+  build-n-publish:
+    name: Build and publish to PyPI
+    runs-on: ubuntu-latest
+    needs: [authorize]
+    steps:
+      - uses: actions/checkout@v3
+      - name: Set up Python 3.6
+        uses: actions/setup-python@v3
+        with:
+          python-version: 3.6
+      - name: Install dependencies
+        run: python -m pip install build setuptools wheel twine
+      - name: Build a binary wheel and a source tarball
+        run: python -m build --sdist --wheel --outdir dist/ .
+      - name: Publish distribution PyPI
+        uses: pypa/gh-action-pypi-publish@release/v1
+        with:
+          user: __token__
+          password: ${{ secrets.PYPI_API_TOKEN }}

.github/workflows/publish-to-test-pypi.yml

@@ -19,18 +19,18 @@ jobs:
     runs-on: ubuntu-latest
     needs: [authorize]
     steps:
-    - uses: actions/checkout@v3
-    - name: Set up Python 3.6
-      uses: actions/setup-python@v3
-      with:
-        python-version: 3.6
-    - name: Install dependencies
-      run: python -m pip install build setuptools wheel twine
-    - name: Build a binary wheel and a source tarball
-      run: python -m build --sdist --wheel --outdir dist/ .
-    - name: Publish distribution Test PyPI
-      uses: pypa/gh-action-pypi-publish@release/v1
-      with:
-        user: __token__
-        password: ${{ secrets.TEST_PYPI_API_TOKEN }}
-        repository_url: https://test.pypi.org/legacy/
+      - uses: actions/checkout@v3
+      - name: Set up Python 3.6
+        uses: actions/setup-python@v3
+        with:
+          python-version: 3.6
+      - name: Install dependencies
+        run: python -m pip install build setuptools wheel twine
+      - name: Build a binary wheel and a source tarball
+        run: python -m build --sdist --wheel --outdir dist/ .
+      - name: Publish distribution Test PyPI
+        uses: pypa/gh-action-pypi-publish@release/v1
+        with:
+          user: __token__
+          password: ${{ secrets.TEST_PYPI_API_TOKEN }}
+          repository_url: https://test.pypi.org/legacy/

.github/workflows/test.yml

@@ -0,0 +1,15 @@
+name: Unit Test
+
+on: [pull_request]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Set up Python 3.6
+        uses: actions/setup-python@v3
+        with:
+          python-version: 3.6
+      - name: Unit Test
+        run: python -m unittest discover -s ./src -p 'test_*.py'

README.md

@@ -4,7 +4,7 @@
 The official Amplitude backend Python SDK for server-side instrumentation.
 
 ## Installation and Quick Start
-Please visit the [Developer Center](https://developers.amplitude.com/docs/python) for instructions on installing and using our the SDK.
+Please visit the [Developer Center](https://developers.amplitude.com/docs/python-beta) for instructions on installing and using our the SDK.
 
 ## Changelog
 View the [releases here](https://github.com/amplitude/Amplitude-Python/releases).

examples/trackevent.py

@@ -17,3 +17,4 @@ def callback_fun(event, code, message):
 }
 client.track(event)
 client.flush()
+client.shutdown()

src/amplitude/client.py

@@ -32,17 +32,18 @@ class Amplitude:
         flush(): Flush all event waiting to be sent in the buffer
         add(plugin): Add the plugin object to client instance.
         remove(plugin): Remove the plugin object from client instance
+        shutdown(): Shutdown the client instance
     """
 
-    def __init__(self, api_key: str, configuration=Config()):
+    def __init__(self, api_key: str, configuration: Config = Config()):
         """The constructor for the Amplitude class
 
         Args:
             api_key (str): The api key of amplitude project. Must be set properly before tracking events.
             configuration (amplitude.config.Config, optional): The configuration of client instance. A new instance
                 with default config value will be used by default.
         """
-        self.configuration = configuration
+        self.configuration: Config = configuration
         self.configuration.api_key = api_key
         self.__timeline = Timeline()
         self.__timeline.setup(self)
@@ -158,3 +159,8 @@ def remove(self, plugin: Plugin):
         """
         self.__timeline.remove(plugin)
         return self
+
+    def shutdown(self):
+        """Shutdown the client instance, not accepting new events, flush all events in buffer"""
+        self.configuration.opt_out = True
+        self.__timeline.shutdown()

src/amplitude/constants.py

@@ -1,7 +1,7 @@
 from enum import Enum
 
 SDK_LIBRARY = "amplitude-python"
-SDK_VERSION = "0.0.2"
+SDK_VERSION = "0.0.3"
 
 EU_ZONE = "EU"
 DEFAULT_ZONE = "US"

src/amplitude/plugin.py

@@ -71,6 +71,9 @@ def execute(self, event: BaseEvent) -> None:
         event = self.timeline.process(event)
         super().execute(event)
 
+    def shutdown(self):
+        self.timeline.shutdown()
+
 
 class AmplitudeDestinationPlugin(DestinationPlugin):
 
@@ -97,6 +100,10 @@ def execute(self, event: BaseEvent) -> None:
     def flush(self):
         self.workers.flush()
 
+    def shutdown(self):
+        self.timeline.shutdown()
+        self.workers.stop()
+
 
 class ContextPlugin(Plugin):
 

src/amplitude/processor.py

@@ -8,8 +8,9 @@ def __init__(self, worker):
         self.configuration = worker.configuration
         self.storage = worker.storage
 
-    def setup(self, configuration):
+    def setup(self, configuration, storage):
         self.configuration = configuration
+        self.storage = storage
 
     def process_response(self, res, events):
         if res.status == HttpStatus.SUCCESS:
@@ -70,4 +71,4 @@ def callback(self, events, code, message):
                     self.configuration.callback(event, code, message)
                 event.callback(code, message)
             except Exception:
-                self.configuration.logger.error(f"Error callback for event {event}")
+                self.configuration.logger.exception(f"Error callback for event {event}")

src/amplitude/storage.py

@@ -72,7 +72,8 @@ def pull(self, batch_size: int) -> List[BaseEvent]:
             result = self.ready_queue[:batch_size]
             self.ready_queue = self.ready_queue[batch_size:]
             index = 0
-            while index < len(self.buffer_data) and index < (batch_size - len(result)) and current_time >= self.buffer_data[index][0]:
+            while index < len(self.buffer_data) and index < (batch_size - len(result)) and \
+                    current_time >= self.buffer_data[index][0]:
                 event = self.buffer_data[index][1]
                 result.append(event)
                 index += 1

src/amplitude/timeline.py

@@ -44,7 +44,7 @@ def flush(self):
             try:
                 destination.flush()
             except Exception:
-                self.logger.error(f"Error for flush events")
+                self.logger.exception("Error for flush events")
 
     def process(self, event):
         if self.configuration.opt_out:
@@ -67,7 +67,11 @@ def apply_plugins(self, plugin_type, event):
                     else:
                         result = plugin.execute(result)
                 except InvalidEventError:
-                    self.logger.error(f"Invalid event body {event}")
+                    self.logger.exception(f"Invalid event body {event}")
                 except Exception:
-                    self.logger.error(f"Error for apply {plugin_type.name} plugin for event {event}")
+                    self.logger.exception(f"Error for apply {plugin_type.name} plugin for event {event}")
         return result
+
+    def shutdown(self):
+        for destination in self.plugins[PluginType.DESTINATION]:
+            destination.shutdown()

src/amplitude/worker.py

@@ -2,8 +2,8 @@
 from concurrent.futures import ThreadPoolExecutor
 from threading import Thread
 
+from amplitude.exception import InvalidAPIKeyError
 from amplitude.http_client import HttpClient
-from amplitude import utils
 from amplitude.processor import ResponseProcessor
 
 
@@ -20,7 +20,7 @@ def __init__(self):
     def setup(self, configuration, storage):
         self.configuration = configuration
         self.storage = storage
-        self.response_processor.setup(configuration)
+        self.response_processor.setup(configuration, storage)
 
     def start(self):
         self.consumer.start()
@@ -33,13 +33,17 @@ def stop(self):
 
     def flush(self):
         events = self.storage.pull_all()
-        self.send(events)
+        if events:
+            self.send(events)
 
     def send(self, events):
         url = self.configuration.server_url
         payload = self.get_payload(events)
         res = HttpClient.post(url, payload)
-        self.response_processor.process_response(res, events)
+        try:
+            self.response_processor.process_response(res, events)
+        except InvalidAPIKeyError:
+            self.configuration.logger.error("Invalid API Key")
 
     def get_payload(self, events) -> bytes:
         payload_body = {
@@ -64,4 +68,3 @@ def buffer_consumer(self):
                     wait_time = self.storage.wait_time / 1000
                     if wait_time > 0:
                         self.storage.lock.wait(wait_time)
-