feat: Added support for events batching

Author: sakshamg1304

requirements.txt

@@ -1,4 +1,3 @@
-aiohttp>=3.8.4
 murmurhash==1.0.10
 requests>=2.27.1
 jsonschema>=3.2.0

setup.py

@@ -121,7 +121,7 @@ def run(self):
 
 setup(
     name="vwo-fme-python-sdk",
-    version="1.8.0",
+    version="1.9.0",
     description="VWO Feature Management and Experimentation SDK for Python",
     long_description=long_description,
     long_description_content_type="text/markdown",

vwo/api/set_attribute_api.py

@@ -51,5 +51,13 @@ def create_and_send_impression_for_attribute(
             ip_address=context.get_ip_address(),
         )
 
-        # Send the constructed properties and payload as a POST request
-        send_post_api_request(properties, payload)
+        from vwo.vwo_client import VWOClient
+        vwo_instance = VWOClient.get_instance()
+
+        # Check if batch events are enabled
+        if vwo_instance.batch_event_queue is not None:
+            # Enqueue the event to the batch queue
+            vwo_instance.batch_event_queue.enqueue(payload)
+        else:
+            # Send the event immediately if batch events are not enabled
+            send_post_api_request(properties, payload)

vwo/api/track_api.py

@@ -93,5 +93,13 @@ def create_and_send_impression_for_track(
             ip_address=context.get_ip_address(),
         )
 
-        # Send the constructed properties and payload as a POST request
-        send_post_api_request(properties, payload)
+        from vwo.vwo_client import VWOClient
+        vwo_instance = VWOClient.get_instance()
+
+        # Check if batch events are enabled
+        if vwo_instance.batch_event_queue is not None:
+            # Enqueue the event to the batch queue
+            vwo_instance.batch_event_queue.enqueue(payload)
+        else:
+            # Send the event immediately if batch events are not enabled
+            send_post_api_request(properties, payload)

vwo/constants/Constants.py

@@ -17,7 +17,7 @@ class Constants:
     # TODO: read from setup.py
     package_file = {
         "name": "vwo-fme-python-sdk",
-        "version": "1.8.0"
+        "version": "1.9.0"
     }
 
     # Constants
@@ -56,3 +56,6 @@ class Constants:
     RANDOM_ALGO = 1
 
     VWO_META_MEG_KEY = "_vwo_meta_meg_"
+
+    THREAD_POOL_MAX_WORKERS = 5
+    SHOULD_USE_THREADING = True

vwo/enums/url_enum.py

@@ -20,7 +20,7 @@ class UrlEnum(Enum):
     BASE_URL = "dev.visualwebsiteoptimizer.com"
     SETTINGS_URL = "/server-side/settings"
     # WEBHOOK_SETTINGS_URL = '/server-side/pull'
-    # BATCH_EVENTS = '/server-side/batch-events'
-    EVENTS = "/events/t"
-    ATTRIBUTE_CHECK = "/check-attribute"
-    GET_USER_DATA = "/get-user-details"
+    EVENTS = '/events/t'
+    ATTRIBUTE_CHECK = '/check-attribute'
+    GET_USER_DATA = '/get-user-details'
+    BATCH_EVENTS = '/server-side/batch-events-v2'

vwo/models/batch_event_data.py

@@ -0,0 +1,68 @@
+# Copyright 2024 Wingify Software Pvt. Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+class BatchEventData:
+    def __init__(self, events_per_request: int = 100, request_time_interval: int = 600, flush_callback=None):
+        """
+        Initialize the BatchEventData model with optional parameters.
+        
+        :param events_per_request: The number of events to send per batch (default is 100).
+        :param request_time_interval: The time interval (in seconds) between batch requests (default is 600 seconds).
+        """
+        self.events_per_request = events_per_request  # Default value: 100
+        self.request_time_interval = request_time_interval  # Default value: 600
+        self.flush_callback = flush_callback
+
+    def get_events_per_request(self) -> int:
+        """
+        Get the number of events to send per batch.
+        """
+        return self.events_per_request
+
+    def set_events_per_request(self, events_per_request: int) -> None:
+        """
+        Set the number of events to send per batch.
+        
+        :param events_per_request: The number of events to send per batch.
+        """
+        self.events_per_request = events_per_request
+
+    def get_request_time_interval(self) -> int:
+        """
+        Get the time interval (in seconds) between batch requests.
+        """
+        return self.request_time_interval
+
+    def set_request_time_interval(self, request_time_interval: int) -> None:
+        """
+        Set the time interval (in seconds) between batch requests.
+        
+        :param request_time_interval: The time interval between batch requests in seconds.
+        """
+        self.request_time_interval = request_time_interval
+    
+    def get_flush_callback(self):
+        """
+        Get the flush callback function.
+        """
+        return self.flush_callback
+
+    def set_flush_callback(self, flush_callback: None):
+        """
+        Set the flush callback function.
+        
+        :param flush_callback: The callback function to be set.
+        """
+        self.flush_callback = flush_callback

vwo/models/vwo_options_model.py

@@ -15,7 +15,7 @@
 
 from typing import Dict
 
-
+from vwo.models.batch_event_data import BatchEventData
 class VWOOptionsModel:
     def __init__(self, options: Dict):
         """
@@ -35,6 +35,17 @@ def __init__(self, options: Dict):
         self.network = options.get("network", None)
         self.vwoBuilder = options.get("vwoBuilder", None)
 
+        # Initialize BatchEventData
+        batch_event_data = options.get('batch_event_data', None)
+        if batch_event_data:
+            self.batchEventData = BatchEventData(
+                events_per_request=batch_event_data.get('events_per_request'),
+                request_time_interval=batch_event_data.get('request_time_interval'),
+                flush_callback=batch_event_data.get('flush_callback', None)
+            )
+        else:
+            self.batchEventData = None
+
     def get_account_id(self) -> str:
         """
         Get the account ID.
@@ -122,3 +133,11 @@ def get_vwo_builder(self):
         :return: The VWO builder instance.
         """
         return self.vwoBuilder
+    
+    def get_batch_event_data(self):
+        """
+        Get the batch event data.
+        
+        :return: The BatchEventData instance.
+        """
+        return self.batchEventData

vwo/packages/network_layer/manager/network_manager.py

@@ -18,17 +18,25 @@
 from ..models.response_model import ResponseModel
 from ..handlers.request_handler import RequestHandler
 from ..client.network_client import NetworkClient
-import aiohttp
 from ...logger.core.log_manager import LogManager
 from ....utils.log_message_util import error_messages
+from concurrent.futures import ThreadPoolExecutor
+from typing import Callable, Dict, Any
+from ....constants.Constants import Constants
 
 
 class NetworkManager:
     _instance = None
 
-    def __init__(self):
+    def __init__(self, threading: Dict[str, Any] = None):
         self.client = None
         self.config = None
+        self.should_use_threading = threading.get(
+            "enabled", Constants.SHOULD_USE_THREADING
+        )
+        self.thread_pool_max_workers = threading.get(
+            "max_workers", Constants.THREAD_POOL_MAX_WORKERS
+        )
 
     def set_config(self, config: GlobalRequestModel):
         self.config = config
@@ -41,9 +49,9 @@ def attach_client(self):
         self.config = GlobalRequestModel()
 
     @classmethod
-    def get_instance(cls) -> "NetworkManager":
+    def get_instance(cls, threading: Dict[str, Any] = None) -> "NetworkManager":
         if cls._instance is None:
-            cls._instance = cls()
+            cls._instance = cls(threading)
         return cls._instance
 
     def create_request(self, request: RequestModel) -> RequestModel:
@@ -65,21 +73,7 @@ def post(self, request: RequestModel) -> ResponseModel:
             return response
         return self.client.post(request_model)
 
-    async def post_async(self, request: RequestModel):
-        try:
-            options = request.get_options()
-            async with aiohttp.ClientSession() as session:
-                async with session.post(
-                    options["url"],
-                    json=options.get("json"),
-                    headers=options.get("headers"),
-                    timeout=options.get("timeout"),
-                ) as response:
-                    return response.status
-        except Exception as err:
-            LogManager.get_instance().error(
-                error_messages.get("NETWORK_CALL_FAILED").format(
-                    method="POST",
-                    err=err,
-                )
-            )
+        # Generic function to handle background task execution with ThreadPoolExecutor
+    def execute_in_background(self, func: Callable):
+        executor = ThreadPoolExecutor(max_workers=self.thread_pool_max_workers)
+        executor.submit(func)

vwo/resources/debug-messages.json

@@ -10,5 +10,6 @@
 
     "IMPRESSION_FOR_TRACK_USER": "Impression built for vwo_variationShown(VWO standard event for tracking user) event haivng Account ID:{accountId}, User ID:{userId}, and experiment ID:{campaignId}",
     "IMPRESSION_FOR_TRACK_GOAL": "Impression built for event:{eventName} event having Account ID:{accountId}, and user ID:{userId}",
-    "IMPRESSION_FOR_SYNC_VISITOR_PROP": "Impression built for {eventName}(VWO internal event) event for Account ID:{accountId}, and user ID:{userId}"
+    "IMPRESSION_FOR_SYNC_VISITOR_PROP": "Impression built for {eventName}(VWO internal event) event for Account ID:{accountId}, and user ID:{userId}",
+    "BATCHING_INITIALIZED": "BATCHING_INITIALIZED"
   }