Add twisted support launchdarkly#18

Author: mrdon

.gitignore

@@ -2,4 +2,6 @@ venv
 *.pyc
 dist
 ldclient_py.egg-info
-build/
+build/
+.idea
+*.iml

README.md

@@ -32,10 +32,11 @@ Development information (for developing this module itself)
 
         pip install -r requirements.txt
         pip install -r test-requirements.txt
+        pip install -r twisted-requirements.txt
 
 2. Run tests:
 
-        $ py.test
+        $ py.test testing
 
 
 Learn more

circle.yml

@@ -4,6 +4,8 @@ dependencies:
     - pyenv shell 3.3.3; $(pyenv which pip) install --upgrade pip
     - pyenv shell 2.6.8; $(pyenv which pip) install -r test-requirements.txt
     - pyenv shell 3.3.3; $(pyenv which pip) install -r test-requirements.txt
+    - pyenv shell 2.6.8; $(pyenv which pip) install -r twisted-requirements.txt
+    - pyenv shell 3.3.3; $(pyenv which pip) install -r twisted-requirements.txt
     - pyenv shell 2.6.8; $(pyenv which python) setup.py install
     - pyenv shell 3.3.3; $(pyenv which python) setup.py install
 

demo/demo_twisted.py

@@ -0,0 +1,20 @@
+from __future__ import print_function
+from ldclient.twisted import TwistedLDClient
+from twisted.internet import task, defer
+
+@defer.inlineCallbacks
+def main(reactor):
+    apiKey = 'whatever'
+    client = TwistedLDClient(apiKey)
+    user = {
+        u'key': u'xyz',
+        u'custom': {
+            u'bizzle': u'def'
+        }
+    }
+    val = yield client.toggle('foo', user)
+    yield client.flush()
+    print("Value: {}".format(val))
+
+if __name__ == '__main__':
+    task.react(main)

ldclient/__init__.py

@@ -161,6 +161,7 @@ def _send(self, event):
         if self._queue.full():
             log.warning("Event queue is full-- dropped an event")
         else:
+            print("putting in {}".format(event))
             self._queue.put(event)
 
     def track(self, event_name, user, data = None):

ldclient/twisted.py

@@ -0,0 +1,140 @@
+import json
+from queue import Empty
+import errno
+from cachecontrol import CacheControl
+from ldclient import LDClient, _headers, log, _evaluate
+from requests.packages.urllib3.exceptions import ProtocolError
+from twisted.internet import task, defer
+import txrequests
+
+
+class TwistedLDClient(LDClient):
+    def __init__(self, api_key, config=None):
+        super().__init__(api_key, config)
+        self._session = CacheControl(txrequests.Session())
+
+    def _check_consumer(self):
+        if not self._consumer or not self._consumer.is_alive():
+            self._consumer = TwistedConsumer(self._session, self._queue, self._api_key, self._config)
+            self._consumer.start()
+
+    def flush(self):
+        if self._offline:
+            print("offline")
+            return defer.succeed(True)
+        self._check_consumer()
+        return self._consumer.flush()
+
+    def toggle(self, key, user, default=False):
+        @defer.inlineCallbacks
+        def run(should_retry):
+            # noinspection PyBroadException
+            try:
+                if self._offline:
+                    return default
+                val = yield self._toggle(key, user, default)
+                self._send({'kind': 'feature', 'key': key, 'user': user, 'value': val})
+                defer.returnValue(val)
+            except ProtocolError as e:
+                inner = e.args[1]
+                if inner.errno == errno.ECONNRESET and should_retry:
+                    log.warning('ProtocolError exception caught while getting flag. Retrying.')
+                    d = yield run(False)
+                    defer.returnValue(d)
+                else:
+                    log.exception('Unhandled exception. Returning default value for flag.')
+                    defer.returnValue(default)
+            except Exception:
+                log.exception('Unhandled exception. Returning default value for flag.')
+                defer.returnValue(default)
+
+        return run(True)
+
+    @defer.inlineCallbacks
+    def _toggle(self, key, user, default):
+        hdrs = _headers(self._api_key)
+        uri = self._config._base_uri + '/api/eval/features/' + key
+        r = yield self._session.get(uri, headers=hdrs, timeout=(self._config._connect, self._config._read))
+        r.raise_for_status()
+        hash = r.json()
+        val = _evaluate(hash, user)
+        if val is None:
+            val = default
+        defer.returnValue(val)
+
+
+class TwistedConsumer(object):
+    def __init__(self, session, queue, api_key, config):
+        self._queue = queue
+        """ @type: queue.Queue """
+        self._session = session
+        """ :type: txrequests.Session """
+
+        self._api_key = api_key
+        self._config = config
+        self._flushed = None
+        """ :type: Deferred """
+        self._looping_call = None
+        """ :type: LoopingCall"""
+
+    def start(self):
+        self._flushed = defer.Deferred()
+        self._looping_call = task.LoopingCall(self._consume)
+        self._looping_call.start(5)
+
+    def stop(self):
+        self._looping_call.stop()
+
+    def is_alive(self):
+        return self._looping_call is not None and self._looping_call.running
+
+    def flush(self):
+        return self._flushed
+
+    def _consume(self):
+        items = []
+        try:
+            while True:
+                items.append(self._queue.get_nowait())
+        except Empty:
+            pass
+
+        if items:
+            def on_batch_done(*_):
+                print("========== batch done")
+                self._flushed.callback(True)
+                self._flushed = defer.Deferred()
+
+            d = self.send_batch(items)
+            """ :type: Deferred """
+            d.addBoth(on_batch_done)
+
+    @defer.inlineCallbacks
+    def send_batch(self, events):
+        @defer.inlineCallbacks
+        def do_send(should_retry):
+            # noinspection PyBroadException
+            try:
+                if isinstance(events, dict):
+                    body = [events]
+                else:
+                    body = events
+                hdrs = _headers(self._api_key)
+                uri = self._config._base_uri + '/api/events/bulk'
+                r = yield self._session.post(uri, headers=hdrs, timeout=(self._config._connect, self._config._read),
+                                             data=json.dumps(body))
+                r.raise_for_status()
+            except ProtocolError as e:
+                inner = e.args[1]
+                if inner.errno == errno.ECONNRESET and should_retry:
+                    log.warning('ProtocolError exception caught while sending events. Retrying.')
+                    yield do_send(False)
+                else:
+                    log.exception('Unhandled exception in event consumer. Analytics events were not processed.')
+            except:
+                log.exception('Unhandled exception in event consumer. Analytics events were not processed.')
+        try:
+            yield do_send(True)
+        finally:
+            for _ in events:
+                self._queue.task_done()

pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+twisted = 1

setup.py

@@ -11,11 +11,13 @@
 # parse_requirements() returns generator of pip.req.InstallRequirement objects
 install_reqs = parse_requirements('requirements.txt', session=uuid.uuid1())
 test_reqs = parse_requirements('test-requirements.txt', session=uuid.uuid1())
+twisted_reqs = parse_requirements('twisted-requirements.txt', session=uuid.uuid1())
 
 # reqs is a list of requirement
 # e.g. ['django==1.5.1', 'mezzanine==1.4.6']
 reqs = [str(ir.req) for ir in install_reqs]
 testreqs = [str(ir.req) for ir in test_reqs]
+txreqs = [str(ir.req) for ir in twisted_reqs]
 
 class PyTest(Command):
     user_options = []
@@ -43,6 +45,9 @@ def run(self):
         'Operating System :: OS Independent',
         'Programming Language :: Python :: 2 :: Only',
     ],
+    extras_require={
+        "twisted": txreqs
+    },
     tests_require=testreqs,
     cmdclass = {'test': PyTest},
 )

test-requirements.txt

@@ -1 +1,2 @@
 pytest==2.6.4
+pytest-twisted==1.5

testing/test_twisted.py

@@ -0,0 +1,158 @@
+from builtins import next
+from builtins import filter
+from builtins import object
+import ldclient
+from ldclient.twisted import TwistedLDClient
+import pytest
+from twisted.internet import defer
+
+try:
+    import queue
+except:
+    import Queue as queue
+
+client = TwistedLDClient("API_KEY", ldclient.Config("http://localhost:3000"))
+
+user = {
+  u'key': u'xyz', 
+  u'custom': { 
+    u'bizzle': u'def' 
+    }
+  }
+
+class MockConsumer(object):
+    def __init__(self):
+      self._running = False
+
+    def stop(self):
+      self._running = False
+
+    def start(self):
+      self._running = True
+
+    def is_alive(self):
+      return self._running
+
+    def flush(self):
+        return defer.succeed(True)
+
+
+def mock_consumer():
+  return MockConsumer()  
+
+def noop_consumer():
+  return    
+
+def mock_toggle(key, user, default):
+    hash = minimal_feature = {
+      u'key': u'feature.key', 
+      u'salt': u'abc',
+      u'on': True,
+      u'variations': [
+        {
+          u'value': True,
+          u'weight': 100,
+          u'targets': []
+        },
+        {
+          u'value': False,
+          u'weight': 0,
+          u'targets': []
+        }
+      ]
+    }
+    val = ldclient._evaluate(hash, user)
+    if val is None:
+        return defer.succeed(default)
+    return defer.succeed(val)
+
+def setup_function(function):
+  client.set_online()
+  client._queue = queue.Queue(10)
+  client._consumer = mock_consumer()
+
+@pytest.fixture(autouse=True)
+def noop_check_consumer(monkeypatch):
+  monkeypatch.setattr(client, '_check_consumer', noop_consumer)
+
+@pytest.fixture(autouse=True)
+def no_remote_toggle(monkeypatch):
+  monkeypatch.setattr(client, '_toggle', mock_toggle)
+
+def test_set_offline():
+  client.set_offline()
+  assert client.is_offline() == True
+
+def test_set_online():
+  client.set_offline()
+  client.set_online()
+  assert client.is_offline() == False
+
+@pytest.inlineCallbacks
+def test_toggle():
+  result = yield client.toggle('xyz', user, default=None)
+  assert result == True
+
+@pytest.inlineCallbacks
+def test_toggle_offline():
+  client.set_offline()
+  assert (yield client.toggle('xyz', user, default=None)) == None
+
+@pytest.inlineCallbacks
+def test_toggle_event():
+  val = yield client.toggle('xyz', user, default=None)
+  def expected_event(e):
+    return e['kind'] == 'feature' and e['key']  == 'xyz' and e['user'] == user and e['value'] == True
+  assert expected_event(client._queue.get(False))
+
+@pytest.inlineCallbacks
+def test_toggle_event_offline():
+  client.set_offline()
+  yield client.toggle('xyz', user, default=None)
+  assert client._queue.empty()
+
+
+def test_identify():
+  client.identify(user)
+  def expected_event(e):
+    return e['kind'] == 'identify' and e['key']  == u'xyz' and e['user'] == user
+  assert expected_event(client._queue.get(False))
+
+def test_identify_offline():
+  client.set_offline()
+  client.identify(user)
+  assert client._queue.empty()
+
+def test_track():
+  client.track('my_event', user, 42)
+  def expected_event(e):
+    return e['kind'] == 'custom' and e['key']  == 'my_event' and e['user'] == user and e['data'] == 42
+  assert expected_event(client._queue.get(False))
+
+def test_track_offline():
+  client.set_offline()
+  client.track('my_event', user, 42)
+  assert client._queue.empty()
+
+def drain(queue):
+    while not queue.empty():
+      queue.get()
+      queue.task_done()
+    return
+
+@pytest.inlineCallbacks
+def test_flush_empties_queue():
+  client.track('my_event', user, 42)
+  client.track('my_event', user, 33)
+  drain(client._queue)
+  yield client.flush()
+  assert client._queue.empty()
+
+
+@pytest.inlineCallbacks
+def test_flush_offline_does_not_empty_queue():
+  client.track('my_event', user, 42)
+  client.track('my_event', user, 33)
+  client.set_offline()
+  yield client.flush()
+  assert not client._queue.empty()

twisted-requirements.txt

@@ -0,0 +1 @@
+txrequests==0.9.2