Changed logger level, added "supported_platform" contract, and other small changes

Author: fblundun

snowplow_tracker/emitters.py

@@ -30,7 +30,7 @@
 from contracts import contract, new_contract
 
 logger = logging.getLogger(__name__)
-logger.setLevel(logging.DEBUG)
+logger.setLevel(logging.INFO)
 
 DEFAULT_MAX_LENGTH = 10
 THREAD_TIMEOUT = 10
@@ -101,6 +101,8 @@ def __init__(self, endpoint, protocol="http", port=None, method="get", buffer_si
 
         self.threads = []
 
+        logger.info("Emitter initialized with endpoint " + self.endpoint)
+
     @staticmethod
     @contract
     def as_collector_uri(endpoint, protocol="http", port=None):
@@ -184,7 +186,7 @@ def http_post(self, data):
         """
         logger.debug("Sending POST request...")
         r = requests.post(self.endpoint, data=data)
-        logger.debug("POST request finished with status code: " + str(r.status_code))
+        logger.info("POST request finished with status code: " + str(r.status_code))
         return r
 
     @contract
@@ -195,7 +197,7 @@ def http_get(self, payload):
         """
         logger.debug("Sending GET request...")
         r = requests.get(self.endpoint, params=payload)        
-        logger.debug("GET request finished with status code: " + str(r.status_code))
+        logger.info("GET request finished with status code: " + str(r.status_code))
         return r
 
     def sync_flush(self):
@@ -241,6 +243,7 @@ def flush(self):
             Schedules a flush task
         """
         super(CeleryEmitter, self).flush.delay()
+        logger.info("Scheduled a Celery task to flush the event queue")
 
 
 class RedisEmitter(object):
@@ -268,7 +271,7 @@ def input(self, payload):
         """
         logger.debug("Pushing event to Redis queue...")
         self.rdb.rpush(self.key, json.dumps(payload))
-        logger.debug("Finished sending event to Redis.")
+        logger.info("Finished sending event to Redis.")
 
     def flush(self):
         logger.warn("The RedisEmitter class does not need to be flushed")

snowplow_tracker/redis_worker.py

@@ -23,22 +23,22 @@
 import redis
 import gevent
 from gevent.pool import Pool
-from consumer import Consumer
 import json
 import signal
 
-DEFAULT_REDIS = redis.StrictRedis()
 DEFAULT_KEY = "snowplow"
 
 class RedisWorker(object):
     """
-        Asynchronously take events from redis and send them to a consumer
+        Asynchronously take events from redis and send them to an emitter
     """
 
-    def __init__(self, _consumer, key=DEFAULT_KEY, dbr=DEFAULT_REDIS):
-        self.consumer = _consumer
+    def __init__(self, emitter, rdb=None, key=DEFAULT_KEY):
+        self.emitter = emitter
         self.key = key
-        self.dbr = dbr
+        if rdb is None:
+            rdb = redis.StrictRedis()
+        self.rdb = rdb
         self.pool = Pool(5)
 
         signal.signal(signal.SIGTERM, self.request_shutdown)
@@ -47,16 +47,16 @@ def __init__(self, _consumer, key=DEFAULT_KEY, dbr=DEFAULT_REDIS):
 
     def send(self, payload):
         """
-            Send an event to a consumer
+            Send an event to an emitter
         """
-        self.consumer.input(payload)
+        self.emitter.input(payload)
 
     def pop_payload(self):
         """
             Get a single event from Redis and send it
             If the Redis queue is empty, sleep to avoid making continual requests
         """
-        payload = self.dbr.lpop(self.key)
+        payload = self.rdb.lpop(self.key)
         if payload:
             self.pool.spawn(self.send, json.loads(payload.decode("utf-8")))
         else:

snowplow_tracker/subject.py

@@ -21,11 +21,13 @@
 
 from contracts import contract, new_contract
 
-new_contract("subject", lambda x: isinstance(x, Subject))
-
 SUPPORTED_PLATFORMS = set(["pc", "tv", "mob", "cnsl", "iot"])
 DEFAULT_PLATFORM = "pc"
 
+new_contract("subject", lambda x: isinstance(x, Subject))
+
+new_contract("supported_platform", lambda x: x in SUPPORTED_PLATFORMS)
+
 class Subject(object):
     """
         Class for an event subject, where we view events as of the form
@@ -40,13 +42,10 @@ def __init__(self):
     def set_platform(self, value):
         """
             :param  value:          One of ["pc", "tv", "mob", "cnsl", "iot"]
-            :type   value:          string
+            :type   value:          supported_platform
             :rtype:                 subject
         """
-        if value in SUPPORTED_PLATFORMS:
-            self.standard_nv_pairs["p"] = value
-        else:
-            raise RuntimeError(value + " is not a supported platform")
+        self.standard_nv_pairs["p"] = value
         return self
 
     @contract

snowplow_tracker/tracker.py

@@ -54,11 +54,11 @@ class Tracker:
     new_contract("emitter", lambda s: hasattr(s, "input"))
 
     @contract
-    def __init__(self, out_queue, _subject=None,
+    def __init__(self, emitter, _subject=None,
                  namespace=None, app_id=None, encode_base64=DEFAULT_ENCODE_BASE64):
         """
-            :param out_queue:        Emitter to which events will be sent
-            :type  out_queue:        emitter
+            :param emitter:          Emitter to which events will be sent
+            :type  emitter:          emitter
             :param _subject:         Subject to be tracked
             :type  _subject:         subject | None
             :param namespace:        Identifier for the Tracker instance
@@ -71,7 +71,7 @@ def __init__(self, out_queue, _subject=None,
         if _subject is None:
             _subject = subject.Subject()
 
-        self.out_queue = out_queue
+        self.emitter = emitter
         self.subject = _subject
         self.encode_base64 = encode_base64
 
@@ -121,7 +121,7 @@ def track(self, pb):
             :type   pb:              payload
             :rtype:                  tracker | int
         """
-        result = self.out_queue.input(pb.nv_pairs)
+        result = self.emitter.input(pb.nv_pairs)
         if result is not None:
             return result
         else:
@@ -357,10 +357,10 @@ def flush(self, async=False):
             :rtype:         tracker | int
         """
         if async:
-            self.out_queue.flush()
+            self.emitter.flush()
             return self
         else:
-            return self.out_queue.sync_flush()
+            return self.emitter.sync_flush()
 
     @contract
     def set_subject(self, _subject):