add error boundary (#72)

* add error boundary * make exception logging throw away * move mockserver setup to a class method * add test for keyboard interrupts * Address Feedback

Author: daniel-statsig

statsig/statsig_error_boundary.py

@@ -0,0 +1,54 @@
+from copyreg import constructor
+import traceback
+import requests
+from statsig.statsig_errors import StatsigNameError, StatsigRuntimeError, StatsigValueError
+
+from statsig.statsig_metadata import _StatsigMetadata
+
+
+class _StatsigErrorBoundary:
+ endpoint = "https://statsigapi.net/v1/sdk_exception"
+ _seen: set
+ _api_key: str
+
+ def __init__(self):
+ self._seen = set()
+
+ def set_api_key(self, api_key):
+ self._api_key = api_key
+
+ def capture(self, task, recover=None):
+ try:
+ return task()
+ except (StatsigValueError, StatsigNameError, StatsigRuntimeError) as e:
+ raise e
+ except Exception as e:
+ print("[Statsig]: An unexpected exception occurred.")
+ traceback.print_exc()
+
+ self.log_exception(e)
+
+ if (recover is None):
+ return None
+ return recover()
+
+ def log_exception(self, exception: Exception):
+ try:
+ name = type(exception).__name__
+ if (self._api_key is None or name in self._seen):
+ return
+
+ self._seen.add(name)
+ meta = _StatsigMetadata.get()
+ requests.post(self.endpoint, json={
+ "exception": type(exception).__name__,
+ "info": traceback.format_exc(),
+ "statsigMetadata": _StatsigMetadata.get()
+ }, headers={
+ 'Content-type': 'application/json',
+ 'STATSIG-API-KEY': self._api_key,
+ 'STATSIG-SDK-TYPE': meta["sdkType"],
+ 'STATSIG-SDK-VERSION': meta["sdkVersion"]
+ })
+ except:
+ pass

statsig/statsig_errors.py

@@ -0,0 +1,10 @@
+class StatsigValueError(ValueError):
+ pass
+
+
+class StatsigRuntimeError(RuntimeError):
+ pass
+
+
+class StatsigNameError(NameError):
+ pass

statsig/statsig_event.py

@@ -2,6 +2,7 @@
 from typing import Union, Optional
 
 from dataclasses import dataclass
+from statsig.statsig_errors import StatsigValueError
 from statsig.statsig_user import StatsigUser
 
 
@@ -21,11 +22,13 @@ class StatsigEvent:
 
  def __post_init__(self):
  if self.user is None or not isinstance(self.user, StatsigUser):
- raise ValueError('StatsigEvent.user must be set')
+ raise StatsigValueError('StatsigEvent.user must be set')
  if self.event_name is None or self.event_name == "":
- raise ValueError('StatsigEvent.event_name must be a valid str')
+ raise StatsigValueError(
+ 'StatsigEvent.event_name must be a valid str')
  if self.value is not None and not isinstance(self.value, str) and not isinstance(self.value, float) and not isinstance(self.value, int):
- raise ValueError('StatsigEvent.value must be a str, float, or int')
+ raise StatsigValueError(
+ 'StatsigEvent.value must be a str, float, or int')
 
  def to_dict(self):
  evt_nullable = {

statsig/statsig_metadata.py

@@ -0,0 +1,10 @@
+from .version import __version__
+
+
+class _StatsigMetadata:
+ @staticmethod
+ def get():
+ return {
+ "sdkVersion": __version__,
+ "sdkType": "py-server"
+ }

statsig/statsig_options.py

@@ -1,3 +1,4 @@
+from statsig.statsig_errors import StatsigValueError
 from .statsig_environment_tier import StatsigEnvironmentTier
 import typing
 
@@ -12,7 +13,7 @@ def __init__(
  timeout: typing.Optional[int] = None,
  rulesets_sync_interval: int = 10,
  idlists_sync_interval: int = 60,
- local_mode: bool=False,
+ local_mode: bool = False,
  bootstrap_values: typing.Optional[str] = None,
  rules_updated_callback: typing.Optional[typing.Callable] = None,
  ):
@@ -23,7 +24,7 @@ def __init__(
  tier, StatsigEnvironmentTier) else tier
  self.set_environment_parameter("tier", tier_str)
  else:
- raise ValueError(
+ raise StatsigValueError(
  'StatsigOptions.tier must be a str or StatsigEnvironmentTier')
  if api is None:
  api = "https://statsigapi.net/v1/"
@@ -34,7 +35,7 @@ def __init__(
  self.local_mode = local_mode
  self.bootstrap_values = bootstrap_values
  self.rules_updated_callback = rules_updated_callback
- 
+
  def set_environment_parameter(self, key: str, value: str):
  if self._environment is None:
  self._environment = {}