Event API JSON file exporter

Author: Kenneth Chan

examples/event_fileexport.py

@@ -0,0 +1,39 @@
+import predictionio
+from datetime import datetime
+import pytz
+
+exporter = predictionio.FileExporter(file_name="test.json")
+
+first_event_properties = {
+    "prop1" : 1,
+    "prop2" : "value2",
+    "prop3" : [1, 2, 3],
+    "prop4" : True,
+    "prop5" : ["a", "b", "c"],
+    "prop6" : 4.56 ,
+    }
+first_event_time = datetime(
+    2004, 12, 13, 21, 39, 45, 618000, pytz.timezone('US/Mountain'))
+exporter.create_event(
+    event="my_event",
+    entity_type="user",
+    entity_id="uid",
+    properties=first_event_properties,
+    event_time=first_event_time,
+    )
+
+# Second event
+second_event_properties = {
+    "someProperty" : "value1",
+    "anotherProperty" : "value2",
+    }
+exporter.create_event(
+    event="my_event",
+    entity_type="user",
+    entity_id="uid",
+    target_entity_type="item",
+    target_entity_id="iid",
+    properties=second_event_properties,
+    event_time=datetime(2014, 12, 13, 21, 38, 45, 618000, pytz.utc))
+
+exporter.close()

predictionio/__init__.py

@@ -167,7 +167,7 @@ class EventClient(BaseClient):
     Default value is 5.
   """
 
-  def __init__(self, access_key, 
+  def __init__(self, access_key,
       url="http://localhost:7070",
       threads=1, qsize=0, timeout=5):
     assert type(access_key) is str, ("access_key must be string. "
@@ -201,7 +201,7 @@ def acreate_event(self, event, entity_type, entity_id,
     :param target_entity_id: target entity id. type str.
     :param properties: a custom dict associated with an event. type dict.
     :param event_time: the time of the event. type datetime, must contain
-      timezone info. 
+      timezone info.
 
     :returns:
       AsyncRequest object. You can call the get_response() method using this
@@ -227,7 +227,7 @@ def acreate_event(self, event, entity_type, entity_id,
     # need to skip the last three digits.
     et_str = et.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + et.strftime("%z")
     data["eventTime"] = et_str
-    
+
     path = "/events.json?accessKey=%s" % (self.access_key, )
     request = AsyncRequest("POST", path, **data)
     request.set_rfunc(self._acreate_resp)
@@ -239,7 +239,7 @@ def create_event(self, event, entity_type, entity_id,
       event_time=None):
     """Synchronously (blocking) create an event."""
     return self.acreate_event(event, entity_type, entity_id,
-        target_entity_type, target_entity_id, properties, 
+        target_entity_type, target_entity_id, properties,
         event_time).get_response()
 
   def aget_event(self, event_id):
@@ -249,7 +249,7 @@ def aget_event(self, event_id):
       event.
 
     :returns:
-      AsyncRequest object. 
+      AsyncRequest object.
     """
     enc_event_id = urllib.quote(event_id, "") # replace special char with %xx
     path = "/events/%s.json" % enc_event_id
@@ -269,7 +269,7 @@ def adelete_event(self, event_id):
       event.
 
     :returns:
-      AsyncRequest object. 
+      AsyncRequest object.
     """
     enc_event_id = urllib.quote(event_id, "") # replace special char with %xx
     path = "/events/%s.json" % (enc_event_id, )
@@ -286,7 +286,7 @@ def delete_event(self, event_id):
 
   def aset_user(self, uid, properties={}, event_time=None):
     """Set properties of a user.
-   
+
     Wrapper of acreate_event function, setting event to "$set" and entity_type
     to "user".
     """
@@ -304,7 +304,7 @@ def set_user(self, uid, properties={}, event_time=None):
 
   def aunset_user(self, uid, properties, event_time=None):
     """Unset properties of an user.
-   
+
     Wrapper of acreate_event function, setting event to "$unset" and entity_type
     to "user".
     """
@@ -323,7 +323,7 @@ def unset_user(self, uid, properties, event_time=None):
 
   def adelete_user(self, uid, event_time=None):
     """Delete a user.
-   
+
     Wrapper of acreate_event function, setting event to "$delete" and entity_type
     to "user".
     """
@@ -339,7 +339,7 @@ def delete_user(self, uid, event_time=None):
 
   def aset_item(self, iid, properties={}, event_time=None):
     """Set properties of an item.
-   
+
     Wrapper of acreate_event function, setting event to "$set" and entity_type
     to "item".
     """
@@ -356,7 +356,7 @@ def set_item(self, iid, properties={}, event_time=None):
 
   def aunset_item(self, iid, properties={}, event_time=None):
     """Unset properties of an item.
-   
+
     Wrapper of acreate_event function, setting event to "$unset" and entity_type
     to "item".
     """
@@ -373,7 +373,7 @@ def unset_item(self, iid, properties={}, event_time=None):
 
   def adelete_item(self, iid, event_time=None):
     """Delete an item.
-   
+
     Wrapper of acreate_event function, setting event to "$delete" and entity_type
     to "item".
     """
@@ -413,7 +413,7 @@ def record_user_action_on_item(self, action, uid, iid, properties={},
 class EngineClient(BaseClient):
   """Client for extracting prediction results from an PredictionIO Engine
   Instance.
-  
+
   :param url: the url of the PredictionIO Engine Instance.
   :param threads: number of threads to handle PredictionIO API requests.
           Must be >= 1.
@@ -424,7 +424,7 @@ class EngineClient(BaseClient):
   :param timeout: timeout for HTTP connection attempts and requests in
     seconds (optional).
     Default value is 5.
-  
+
   """
   def __init__(self, url="http://localhost:8000", threads=1,
       qsize=0, timeout=5):
@@ -435,7 +435,7 @@ def asend_query(self, data):
     query.
 
     :param data: the query: It is coverted to an json object using json.dumps
-      method. type dict. 
+      method. type dict.
 
     :returns:
       AsyncRequest object. You can call the get_response() method using this
@@ -449,10 +449,56 @@ def asend_query(self, data):
 
   def send_query(self, data):
     """Synchronously send a request.
-    
+
     :param data: the query: It is coverted to an json object using json.dumps
-      method. type dict. 
+      method. type dict.
 
     :returns: the prediction.
     """
     return self.asend_query(data).get_response()
+
+class FileExporter(object):
+  """File exporter to export events to JSON file for batch import
+
+  :param file_name: the destination file name
+  """
+  def __init__(self, file_name):
+    """Constructor of Exporter.
+
+    """
+    self._file = open(file_name, 'w')
+
+  def create_event(self, event, entity_type, entity_id,
+      target_entity_type=None, target_entity_id=None, properties=None,
+      event_time=None):
+    """write event to the file"""
+    data = {
+        "event": event,
+        "entityType": entity_type,
+        "entityId": entity_id,
+        }
+
+    if target_entity_type is not None:
+      data["targetEntityType"] = target_entity_type
+
+    if target_entity_id is not None:
+      data["targetEntityId"] = target_entity_id
+
+    if properties is not None:
+      data["properties"] = properties
+
+    et = event_time_validation(event_time)
+    # EventServer uses milliseconds, but python datetime class uses micro. Hence
+    # need to skip the last three digits.
+    et_str = et.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + et.strftime("%z")
+    data["eventTime"] = et_str
+
+    j = json.dumps(data)
+    self._file.write(j+"\n")
+
+  def close(self):
+    """Close the FileExporter
+
+    Call this method when you finish writing all events to JSON file
+    """
+    self._file.close()