Quick PoC of sending tracker data to custom Snowplow Collector.
python simple_snowflow_collector.pyUsed to send events to collector. More about python collector in Snowplow wiki
python test_tracker.pyOutput one should see after executing above commands.
| #!/usr/bin/env python | |
| from flask import Flask | |
| from flask import request | |
| import logging | |
| import sys | |
| app = Flask(__name__) | |
| @app.route('/i') | |
| def hello_world(): | |
| return 'Hello, World!' | |
| @app.route('/com.snowplowanalytics.snowplow/tp2', methods=['POST']) | |
| def hello_world_post(): | |
| rqd = request.data | |
| app.logger.info(rqd) | |
| return 'Hello, World!' | |
| if __name__ == '__main__': | |
| app.logger.addHandler(logging.StreamHandler(sys.stdout)) | |
| app.logger.setLevel(logging.DEBUG) | |
| app.run() |
| #!/usr/bin/python | |
| from snowplow_tracker import Subject, Tracker, Emitter | |
| s1 = Subject() | |
| s1.set_user_id("0a78f2867de") | |
| s1.set_timezone("America/Los_Angeles") | |
| e_post = Emitter("localhost:5000", method="post", buffer_size=1) | |
| e_get = Emitter("localhost:5000", method="get", buffer_size=1) | |
| t_get = Tracker(e_get, subject=s1) | |
| t_post = Tracker(e_post, subject=s1) | |
| t_get.track_struct_event("shop_GET", "add-to-basket", None, "pcs", 2) | |
| t_post.track_struct_event("shop_POST", "add-to-basket", None, "pcs", 2) |
