minimal-snowplow-tracker

import os from snowplow_tracker import Tracker, Emitter from snowplow_tracker.self_describing_json import SelfDescribingJson # Replace with your Snowplow Collector endpoint COLLECTOR_URL = os.environ.get('SNOWPLOW_COLLECTOR_URL', 'http://localhost:9090') # Initialize an emitter to send events to the collector emitter = Emitter(COLLECTOR_URL) # Initialize the tracker with a namespace and application ID tracker = Tracker(emitters=[emitter], namespace='my_app_namespace', app_id='my-python-app') # Track a structured event tracker.track_struct_event( category='example_category', action='example_action', label='example_label', property='example_property', value=1.0 ) # Track an unstructured event with a self-describing JSON schema example_schema = 'iglu:com.example_company/example_event/jsonschema/1-0-0' example_data = { 'event_detail': 'This is an unstructured event', 'session_id': 'abc-123' } unstructured_event = SelfDescribingJson(example_schema, example_data) tracker.track_unstruct_event(unstructured_event) print(f"Events sent to {COLLECTOR_URL} using minimal-snowplow-tracker.") # In a real application, you might want to gracefully close the emitter or ensure all events are sent. # For simple scripts, events are often sent synchronously or on program exit.