Merge pull request #95 from trubrics/feature/flush_limit

Added hard limit to flush in SDK

Author: jeffkayne

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [1.8.2] - 2025-02-10
+### Fixed
+- Added missing flushing limit values, and setup automated flush if limit is reached
+
 
 ## [1.8.1] - 2025-02-07
 ### Fixed

README.md

@@ -79,7 +79,7 @@ client = Trubrics(
 ```
 
 - flush_interval: Time in seconds between automatic flushes (default: 10)
-- flush_at: Number of events that trigger a flush (default: 20)
+- flush_batch_size: Number of events that trigger a flush (default: 20)
 
 ## Development
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "trubrics"
-version = "1.8.1"
+version = "1.8.2"
 requires-python = ">=3.10"
 license = { text = "Apache 2.0" }
 authors = [

trubrics/config.py

@@ -0,0 +1,5 @@
+MIN_FLUSH_INTERVAL = 1
+MAX_FLUSH_BATCH_SIZE = 100
+DEFAULT_FLUSH_INTERVAL = 10
+DEFAULT_FLUSH_BATCH_SIZE = 20
+DEFAULT_FLUSH_PERIODIC_CHECK = 0.1

trubrics/logger.py

@@ -8,4 +8,4 @@
     )
     handler.setFormatter(formatter)
     trubrics_logger.addHandler(handler)
-    trubrics_logger.setLevel(logging.ERROR)
+    trubrics_logger.setLevel(logging.INFO)

trubrics/main.py

@@ -6,6 +6,13 @@
 
 import requests
 
+from trubrics.config import (
+    DEFAULT_FLUSH_BATCH_SIZE,
+    DEFAULT_FLUSH_INTERVAL,
+    DEFAULT_FLUSH_PERIODIC_CHECK,
+    MAX_FLUSH_BATCH_SIZE,
+    MIN_FLUSH_INTERVAL,
+)
 from trubrics.logger import trubrics_logger
 
 
@@ -14,22 +21,60 @@ def __init__(
         self,
         api_key: str,
         host: str = "https://app.trubrics.com/api/ingestion",
-        flush_interval: int = 10,
-        flush_at: int = 20,
+        flush_interval: float = DEFAULT_FLUSH_INTERVAL,
+        flush_batch_size: int = DEFAULT_FLUSH_BATCH_SIZE,
+        flush_periodic_check: float = DEFAULT_FLUSH_PERIODIC_CHECK,
         logger: logging.Logger = trubrics_logger,
     ):
+        f"""
+        Initialize the Trubrics client.
+        Args:
+            api_key (str): The API key for the Trubrics account.
+            host (str): The host URL for the Trubrics API.
+            flush_interval (int): The interval in seconds after which events should be flushed. Minimum possible value is {MIN_FLUSH_INTERVAL}.
+            flush_batch_size (int): The number of events to flush at a time. Max possible value is {MAX_FLUSH_BATCH_SIZE}.
+            flush_periodic_check (float): The interval in seconds between periodic flush checks.
+            logger (logging.Logger): The logger to use for logging.
+
+        """
         self.host = host
         self.api_key = api_key
         self.queue: list[dict] = []
-        self.flush_interval = flush_interval
-        self.flush_at = flush_at
         self.last_flush_time = datetime.now(timezone.utc)
         self.is_flushing = False
         self._lock = threading.Lock()
         self._stop_event = threading.Event()
+        self.logger = logger
+
+        self._init_flush_parameters(
+            flush_interval, flush_batch_size, flush_periodic_check
+        )
         self._thread = threading.Thread(target=self._periodic_flush, daemon=True)
         self._thread.start()
-        self.logger = logger
+
+    def _init_flush_parameters(
+        self, flush_interval: float, flush_batch_size: int, flush_periodic_check: float
+    ):
+        if flush_interval < MIN_FLUSH_INTERVAL:
+            self.logger.warning(
+                f"Flush interval {flush_interval} is too low. Setting it to minimum allowed value of {MIN_FLUSH_INTERVAL}."
+            )
+            flush_interval = MIN_FLUSH_INTERVAL
+        if flush_batch_size > MAX_FLUSH_BATCH_SIZE:
+            self.logger.warning(
+                f"Flush batch size {flush_batch_size} is too high. Setting to maximum allowed value of {MAX_FLUSH_BATCH_SIZE}."
+            )
+            flush_batch_size = MAX_FLUSH_BATCH_SIZE
+        if flush_periodic_check > flush_interval:
+            self.logger.warning(
+                f"Periodic flush check interval {flush_periodic_check} is higher than defined \
+                    flush interval period. Setting it to Flush interval period {flush_interval}."
+            )
+            flush_periodic_check = flush_interval
+
+        self.flush_interval = flush_interval
+        self.flush_batch_size = flush_batch_size
+        self.flush_periodic_check = flush_periodic_check
 
     def track(
         self,
@@ -47,10 +92,6 @@ def track(
             timestamp (datetime | None): The timestamp of event. If None, the current time in UTC is used. If not a datetime object, the event is ignored.
         """
 
-        if timestamp and not isinstance(timestamp, datetime):
-            self.logger.error("Timestamp must be a datetime object. Ignoring event.")
-            return
-
         event_dict = {
             "user_id": user_id,
             "event": event,
@@ -124,14 +165,19 @@ def flush(self):
                 self.queue.clear()
 
         if events:
-            success = self._post(events)
-            self.last_flush_time = datetime.now(timezone.utc)
+            for batch_id in range(0, len(events), self.flush_batch_size):
+                batch = events[batch_id : batch_id + self.flush_batch_size]
+                success = self._post(batch)
 
-            if not success:
-                self.logger.info(f"Retrying flush of {queue_len} events.")
-                time.sleep(5)
+                if not success:
+                    self.logger.warning(
+                        f"Retrying flush of batch {batch_id} of {len(batch)} events."
+                    )
+                    time.sleep(5)
+                    self._post(batch)
 
-                self._post(events)
+            self.last_flush_time = datetime.now(timezone.utc)
+            self.logger.info(f"Flush of {len(events)} events completed.")
 
         with self._lock:
             self.is_flushing = False
@@ -177,13 +223,14 @@ def _post(self, events: list[dict]):
 
     def _periodic_flush(self):
         while not self._stop_event.is_set():
-            time.sleep(1)
+            time.sleep(self.flush_periodic_check)
 
             queue_len = len(self.queue)
             now = datetime.now(timezone.utc)
             time_since_last_flush = (now - self.last_flush_time).total_seconds()
             if (
-                queue_len >= self.flush_at
+                queue_len >= self.flush_batch_size
                 or time_since_last_flush >= self.flush_interval
             ):
+                self.logger.debug(f"Time since last flush: {time_since_last_flush}")
                 self.flush()