snowplow_tracker package

snowplow_tracker.contracts module

snowplow_tracker.contracts.contracts_enabled() → bool
snowplow_tracker.contracts.disable_contracts() → None
snowplow_tracker.contracts.enable_contracts() → None
snowplow_tracker.contracts.form_element(element: Dict[str, Any]) → None
snowplow_tracker.contracts.greater_than(value: float, compared_to: float) → None
snowplow_tracker.contracts.non_empty(seq: Sized) → None
snowplow_tracker.contracts.non_empty_string(s: str) → None
snowplow_tracker.contracts.one_of(value: Any, supported: Iterable) → None
snowplow_tracker.contracts.satisfies(value: Any, check: Callable[[Any], bool]) → None

snowplow_tracker.emitters module

class snowplow_tracker.emitters.AsyncEmitter(endpoint: str, protocol: typing_extensions.Literal['http', 'https'][http, https] = 'http', port: Optional[int] = None, method: typing_extensions.Literal['get', 'post'][get, post] = 'post', batch_size: Optional[int] = None, on_success: Optional[Callable[[List[Dict[str, Any]]], None]] = None, on_failure: Optional[Callable[[int, List[Dict[str, Any]]], None]] = None, thread_count: int = 1, byte_limit: Optional[int] = None, request_timeout: Union[float, Tuple[float, float], None] = None, max_retry_delay_seconds: int = 60, buffer_capacity: int = None, custom_retry_codes: Dict[int, bool] = {}, event_store: Optional[snowplow_tracker.event_store.EventStore] = None, session: Optional[requests.sessions.Session] = None)

Bases: snowplow_tracker.emitters.Emitter

Uses threads to send HTTP requests asynchronously

consume() → None
flush() → None

Removes all dead threads, then creates a new thread which executes the flush method of the base Emitter class

sync_flush() → None

Calls the flush method of the base Emitter class. This is guaranteed to be blocking, not asynchronous.

class snowplow_tracker.emitters.Emitter(endpoint: str, protocol: typing_extensions.Literal['http', 'https'][http, https] = 'https', port: Optional[int] = None, method: typing_extensions.Literal['get', 'post'][get, post] = 'post', batch_size: Optional[int] = None, on_success: Optional[Callable[[List[Dict[str, Any]]], None]] = None, on_failure: Optional[Callable[[int, List[Dict[str, Any]]], None]] = None, byte_limit: Optional[int] = None, request_timeout: Union[float, Tuple[float, float], None] = None, max_retry_delay_seconds: int = 60, buffer_capacity: Optional[int] = None, custom_retry_codes: Dict[int, bool] = {}, event_store: Optional[snowplow_tracker.event_store.EventStore] = None, session: Optional[requests.sessions.Session] = None)

Bases: object

Synchronously send Snowplow events to a Snowplow collector Supports both GET and POST requests

static as_collector_uri(endpoint: str, protocol: typing_extensions.Literal['http', 'https'][http, https] = 'https', port: Optional[int] = None, method: typing_extensions.Literal['get', 'post'][get, post] = 'post') → str
Parameters

  • endpoint (string) – The raw endpoint provided by the user

  • protocol (protocol) – The protocol to use - http or https

  • port (int | None) – The collector port to connect to

  • method (method) – Either get or post HTTP method

Return type

string

static attach_sent_timestamp(events: List[Dict[str, Any]]) → None

Attach (by mutating in-place) current timestamp in milliseconds as stm param

Parameters

events (list(dict(string:*))) – Array of events to be sent

Return type

None

cancel_flush_timer() → None

Abort automatic async flushing

flush() → None

Sends all events in the buffer to the collector.

http_get(payload: Dict[str, Any]) → int
Parameters

payload (dict(string:*)) – The event properties

http_post(data: str) → int
Parameters

data (string) – The array of JSONs to be sent

input(payload: Dict[str, Any]) → None

Adds an event to the buffer. If the maximum size has been reached, flushes the buffer.

Parameters

payload (dict(string:*)) – The name-value pairs for the event

static is_good_status_code(status_code: int) → bool
Parameters

status_code (int) – HTTP status code

Return type

bool

reached_limit() → bool

Checks if event-size or bytes limit are reached

Return type

bool

send_events(evts: List[Dict[str, Any]]) → None
Parameters

evts (list(dict(string:*))) – Array of events to be sent

set_flush_timer(timeout: float) → None

Set an interval at which the buffer will be flushed :param timeout: interval in seconds :type timeout: int | float

sync_flush() → None

Calls the flush method of the base Emitter class. This is guaranteed to be blocking, not asynchronous.

class snowplow_tracker.emitters.FlushTimer(emitter: snowplow_tracker.emitters.Emitter, repeating: bool)

Bases: object

Internal class used by the Emitter to schedule flush calls for later.

cancel() → None
is_active() → bool
start(timeout: float) → bool

snowplow_tracker.payload module

class snowplow_tracker.payload.Payload(dict_: Optional[Dict[str, Any]] = None)

Bases: object

add(name: str, value: Any) → None

Add a name value pair to the Payload object

add_dict(dict_: Dict[str, Any], base64: bool = False) → None

Add a dict of name value pairs to the Payload object

Parameters

dict (dict(string:*)) – Dictionary to be added to the Payload

add_json(dict_: Optional[Dict[str, Any]], encode_base64: bool, type_when_encoded: str, type_when_not_encoded: str, json_encoder: Optional[Callable[[Any], Any]] = None) → None

Add an encoded or unencoded JSON to the payload

Parameters

  • dict (dict(string:*) | None) – Custom context for the event

  • encode_base64 (bool) – If the payload is base64 encoded

  • type_when_encoded (string) – Name of the field when encode_base64 is set

  • type_when_not_encoded (string) – Name of the field when encode_base64 is not set

  • json_encoder (function | None) – Custom JSON serializer that gets called on non-serializable object

get() → Dict[str, Any]

Returns the context dictionary from the Payload object

snowplow_tracker.self_describing_json module

class snowplow_tracker.self_describing_json.SelfDescribingJson(schema: str, data: Union[Dict[str, Any], List[Dict[str, Any]]])

Bases: object

property schema
to_json() → Dict[str, Any]
to_string() → str

snowplow_tracker.subject module

class snowplow_tracker.subject.Subject

Bases: object

Class for an event subject, where we view events as of the form

(Subject) -> (Verb) -> (Object)

combine_subject(subject: Optional[Subject]) → Dict[str, Any]

Merges another instance of Subject, with self taking priority :param subject Subject to update :type subject subject :rtype PayloadDict

set_color_depth(depth: int) → snowplow_tracker.subject.Subject
Parameters

depth (int) – Depth of the color on the screen

Return type

subject

set_domain_session_id(sid: str) → snowplow_tracker.subject.Subject

Set the domain session ID :param sid: Domain session ID :type sid: string :rtype: subject

set_domain_session_index(vid: int) → snowplow_tracker.subject.Subject

Set the domain session Index :param vid: Domain session Index :type vid: int :rtype: subject

set_domain_user_id(duid: str) → snowplow_tracker.subject.Subject

Set the domain user ID

Parameters

duid (string) – Domain user ID

Return type

subject

set_ip_address(ip: str) → snowplow_tracker.subject.Subject

Set the domain user ID

Parameters

ip (string) – IP address

Return type

subject

set_lang(lang: str) → snowplow_tracker.subject.Subject

Set language.

Parameters

lang (string) – Language the application is set to

Return type

subject

set_network_user_id(nuid: str) → snowplow_tracker.subject.Subject

Set the network user ID field This overwrites the nuid field set by the collector

Parameters

nuid (string) – Network user ID

Return type

subject

set_platform(value: typing_extensions.Literal['pc', 'tv', 'mob', 'cnsl', 'iot', 'web', 'srv', 'app'][pc, tv, mob, cnsl, iot, web, srv, app]) → snowplow_tracker.subject.Subject
Parameters

value (supported_platform) – One of [“pc”, “tv”, “mob”, “cnsl”, “iot”, “web”, “srv”, “app”]

Return type

subject

set_screen_resolution(width: int, height: int) → snowplow_tracker.subject.Subject
Parameters

  • width (int,>0) – Width of the screen

  • height (int,>0) – Height of the screen

Return type

subject

set_timezone(timezone: str) → snowplow_tracker.subject.Subject
Parameters

timezone (string) – Timezone as a string

Return type

subject

set_user_id(user_id: str) → snowplow_tracker.subject.Subject
Parameters

user_id (string) – User ID

Return type

subject

set_useragent(ua: str) → snowplow_tracker.subject.Subject

Set the user agent

Parameters

ua (string) – User agent

Return type

subject

set_viewport(width: int, height: int) → snowplow_tracker.subject.Subject
Parameters

  • width (int,>0) – Width of the viewport

  • height (int,>0) – Height of the viewport

Return type

subject

snowplow_tracker.tracker module

class snowplow_tracker.tracker.Tracker(namespace: str, emitters: Union[List[snowplow_tracker.typing.EmitterProtocol], snowplow_tracker.typing.EmitterProtocol], subject: Optional[snowplow_tracker.subject.Subject] = None, app_id: Optional[str] = None, encode_base64: bool = True, json_encoder: Optional[Callable[[Any], Any]] = None)

Bases: object

add_emitter(emitter: snowplow_tracker.typing.EmitterProtocol) → snowplow_tracker.tracker.Tracker

Add a new emitter to which events should be passed

Parameters

emitter (emitter) – New emitter

Return type

tracker

complete_payload(event: snowplow_tracker.events.event.Event) → snowplow_tracker.payload.Payload
flush(is_async: bool = False) → snowplow_tracker.tracker.Tracker

Flush the emitter

Parameters

is_async (bool) – Whether the flush is done asynchronously. Default is False

Return type

tracker

get_namespace() → str
static get_timestamp(tstamp: Optional[float] = None) → int
Parameters

tstamp (int | float | None) – User-input timestamp or None

Return type

int

static get_uuid() → str

Set transaction ID for the payload once during the lifetime of the event.

Return type

string

set_subject(subject: Optional[snowplow_tracker.subject.Subject]) → snowplow_tracker.tracker.Tracker

Set the subject of the events fired by the tracker

Parameters

subject (subject | None) – Subject to be tracked

Return type

tracker

track(event: snowplow_tracker.events.event.Event) → Optional[str]

Send the event payload to a emitter. Returns the tracked event ID. :param event: Event :type event: events.Event :rtype: String

track_add_to_cart(sku: str, quantity: int, name: Optional[str] = None, category: Optional[str] = None, unit_price: Optional[float] = None, currency: Optional[str] = None, context: Optional[List[snowplow_tracker.self_describing_json.SelfDescribingJson]] = None, tstamp: Optional[float] = None, event_subject: Optional[snowplow_tracker.subject.Subject] = None) → snowplow_tracker.tracker.Tracker
Parameters

  • sku (non_empty_string) – Item SKU or ID

  • quantity (int) – Number added to cart

  • name (string_or_none) – Item’s name

  • category (string_or_none) – Item’s category

  • unit_price (int | float | None) – Item’s price

  • currency (string_or_none) – Type of currency the price is in

  • context (context_array | None) – Custom context for the event

  • tstamp (int | float | None) – Optional event timestamp in milliseconds

  • event_subject (subject | None) – Optional per event subject

Return type

Tracker

track_ecommerce_transaction(order_id: str, total_value: float, affiliation: Optional[str] = None, tax_value: Optional[float] = None, shipping: Optional[float] = None, city: Optional[str] = None, state: Optional[str] = None, country: Optional[str] = None, currency: Optional[str] = None, items: Optional[List[Dict[str, Any]]] = None, context: Optional[List[snowplow_tracker.self_describing_json.SelfDescribingJson]] = None, tstamp: Optional[float] = None, event_subject: Optional[snowplow_tracker.subject.Subject] = None) → snowplow_tracker.tracker.Tracker
Parameters

  • order_id (non_empty_string) – ID of the eCommerce transaction

  • total_value (int | float) – Total transaction value

  • affiliation (string_or_none) – Transaction affiliation

  • tax_value (int | float | None) – Transaction tax value

  • shipping (int | float | None) – Delivery cost charged

  • city (string_or_none) – Delivery address city

  • state (string_or_none) – Delivery address state

  • country (string_or_none) – Delivery address country

  • currency (string_or_none) – The currency the price is expressed in

  • items (list(dict(str:*)) | None) – The items in the transaction

  • context (context_array | None) – Custom context for the event

  • tstamp (int | float | None) – Optional event timestamp in milliseconds

  • event_subject (subject | None) – Optional per event subject

Return type

Tracker

track_ecommerce_transaction_item(order_id: str, sku: str, price: float, quantity: int, name: Optional[str] = None, category: Optional[str] = None, currency: Optional[str] = None, context: Optional[List[snowplow_tracker.self_describing_json.SelfDescribingJson]] = None, tstamp: Optional[float] = None, event_subject: Optional[snowplow_tracker.subject.Subject] = None) → snowplow_tracker.tracker.Tracker

This is an internal method called by track_ecommerce_transaction. It is not for public use.

Parameters

  • order_id (non_empty_string) – Order ID

  • sku (non_empty_string) – Item SKU

  • price (int | float) – Item price

  • quantity (int) – Item quantity

  • name (string_or_none) – Item name

  • category (string_or_none) – Item category

  • currency (string_or_none) – The currency the price is expressed in

  • context (context_array | None) – Custom context for the event

  • tstamp (int | float | None) – Optional event timestamp in milliseconds

  • event_subject (subject | None) – Optional per event subject

Return type

Tracker

track_form_change(form_id: str, element_id: Optional[str], node_name: typing_extensions.Literal['INPUT', 'TEXTAREA', 'SELECT'][INPUT, TEXTAREA, SELECT], value: Optional[str], type_: Optional[str] = None, element_classes: Union[List[str], Tuple[str, Any], None] = None, context: Optional[List[snowplow_tracker.self_describing_json.SelfDescribingJson]] = None, tstamp: Optional[float] = None, event_subject: Optional[snowplow_tracker.subject.Subject] = None) → snowplow_tracker.tracker.Tracker
Parameters

  • form_id (non_empty_string) – ID attribute of the HTML form

  • element_id (string_or_none) – ID attribute of the HTML element

  • node_name (form_node_name) – Type of input element

  • value (string_or_none) – Value of the input element

  • type (non_empty_string, form_type) – Type of data the element represents

  • element_classes (list(str) | tuple(str,*) | None) – Classes of the HTML element

  • context (context_array | None) – Custom context for the event

  • tstamp (int | float | None) – Optional event timestamp in milliseconds

  • event_subject (subject | None) – Optional per event subject

Return type

Tracker

track_form_submit(form_id: str, form_classes: Union[List[str], Tuple[str, Any], None] = None, elements: Optional[List[Dict[str, Any]]] = None, context: Optional[List[snowplow_tracker.self_describing_json.SelfDescribingJson]] = None, tstamp: Optional[float] = None, event_subject: Optional[snowplow_tracker.subject.Subject] = None) → snowplow_tracker.tracker.Tracker
Parameters

  • form_id (non_empty_string) – ID attribute of the HTML form

  • form_classes (list(str) | tuple(str,*) | None) – Classes of the HTML form

  • elements (list(form_element) | None) – Classes of the HTML form

  • context (context_array | None) – Custom context for the event

  • tstamp (int | float | None) – Optional event timestamp in milliseconds

  • event_subject (subject | None) – Optional per event subject

Return type

Tracker

Parameters

  • target_url (non_empty_string) – Target URL of the link

  • element_id (string_or_none) – ID attribute of the HTML element

  • element_classes (list(str) | tuple(str,*) | None) – Classes of the HTML element

  • element_target (string_or_none) – ID attribute of the HTML element

  • element_content (string_or_none) – The content of the HTML element

  • context (context_array | None) – Custom context for the event

  • tstamp (int | float | None) – Optional event timestamp in milliseconds

  • event_subject (subject | None) – Optional per event subject

Return type

Tracker

track_mobile_screen_view(name: str, id_: Optional[str] = None, type: Optional[str] = None, previous_name: Optional[str] = None, previous_id: Optional[str] = None, previous_type: Optional[str] = None, transition_type: Optional[str] = None, context: Optional[List[snowplow_tracker.self_describing_json.SelfDescribingJson]] = None, tstamp: Optional[float] = None, event_subject: Optional[snowplow_tracker.subject.Subject] = None) → snowplow_tracker.tracker.Tracker
Parameters

  • name (string_or_none) – The name of the screen view event

  • id (string | None) – Screen view ID. This must be of type UUID.

  • type (string | None) – The type of screen that was viewed e.g feed / carousel.

  • previous_name (string | None) – The name of the previous screen.

  • previous_id (string | None) – The screenview ID of the previous screenview.

:param previous_type The screen type of the previous screenview :type previous_type string | None :param transition_type The type of transition that led to the screen being viewed. :type transition_type string | None :param context: Custom context for the event :type context: context_array | None :param tstamp: Optional event timestamp in milliseconds :type tstamp: int | float | None :param event_subject: Optional per event subject :type event_subject: subject | None :rtype: Tracker

track_page_ping(page_url: str, page_title: Optional[str] = None, referrer: Optional[str] = None, min_x: Optional[int] = None, max_x: Optional[int] = None, min_y: Optional[int] = None, max_y: Optional[int] = None, context: Optional[List[snowplow_tracker.self_describing_json.SelfDescribingJson]] = None, tstamp: Optional[float] = None, event_subject: Optional[snowplow_tracker.subject.Subject] = None) → snowplow_tracker.tracker.Tracker
Parameters

  • page_url (non_empty_string) – URL of the viewed page

  • page_title (string_or_none) – Title of the viewed page

  • referrer (string_or_none) – Referrer of the page

  • min_x (int | None) – Minimum page x offset seen in the last ping period

  • max_x (int | None) – Maximum page x offset seen in the last ping period

  • min_y (int | None) – Minimum page y offset seen in the last ping period

  • max_y (int | None) – Maximum page y offset seen in the last ping period

  • context (context_array | None) – Custom context for the event

  • tstamp (int | float | None) – Optional event timestamp in milliseconds

  • event_subject (subject | None) – Optional per event subject

Return type

Tracker

track_page_view(page_url: str, page_title: Optional[str] = None, referrer: Optional[str] = None, context: Optional[List[snowplow_tracker.self_describing_json.SelfDescribingJson]] = None, tstamp: Optional[float] = None, event_subject: Optional[snowplow_tracker.subject.Subject] = None) → snowplow_tracker.tracker.Tracker
Parameters

  • page_url (non_empty_string) – URL of the viewed page

  • page_title (string_or_none) – Title of the viewed page

  • referrer (string_or_none) – Referrer of the page

  • context (context_array | None) – Custom context for the event

  • tstamp (int | float | None) – Optional event timestamp in milliseconds

  • event_subject (subject | None) – Optional per event subject

Return type

Tracker

track_remove_from_cart(sku: str, quantity: int, name: Optional[str] = None, category: Optional[str] = None, unit_price: Optional[float] = None, currency: Optional[str] = None, context: Optional[List[snowplow_tracker.self_describing_json.SelfDescribingJson]] = None, tstamp: Optional[float] = None, event_subject: Optional[snowplow_tracker.subject.Subject] = None) → snowplow_tracker.tracker.Tracker
Parameters

  • sku (non_empty_string) – Item SKU or ID

  • quantity (int) – Number added to cart

  • name (string_or_none) – Item’s name

  • category (string_or_none) – Item’s category

  • unit_price (int | float | None) – Item’s price

  • currency (string_or_none) – Type of currency the price is in

  • context (context_array | None) – Custom context for the event

  • tstamp (int | float | None) – Optional event timestamp in milliseconds

  • event_subject (subject | None) – Optional per event subject

Return type

Tracker

track_screen_view(name: Optional[str] = None, id_: Optional[str] = None, context: Optional[List[snowplow_tracker.self_describing_json.SelfDescribingJson]] = None, tstamp: Optional[float] = None, event_subject: Optional[snowplow_tracker.subject.Subject] = None) → snowplow_tracker.tracker.Tracker
Parameters

  • name (string_or_none) – The name of the screen view event

  • id (string_or_none) – Screen view ID

  • context (context_array | None) – Custom context for the event

  • tstamp (int | float | None) – Optional event timestamp in milliseconds

  • event_subject (subject | None) – Optional per event subject

Return type

Tracker

track_self_describing_event(event_json: snowplow_tracker.self_describing_json.SelfDescribingJson, context: Optional[List[snowplow_tracker.self_describing_json.SelfDescribingJson]] = None, tstamp: Optional[float] = None, event_subject: Optional[snowplow_tracker.subject.Subject] = None) → snowplow_tracker.tracker.Tracker
Parameters

  • event_json (self_describing_json) – The properties of the event. Has two field: A “data” field containing the event properties and A “schema” field identifying the schema against which the data is validated

  • context (context_array | None) – Custom context for the event

  • tstamp (int | float | None) – Optional event timestamp in milliseconds

  • event_subject (subject | None) – Optional per event subject

Return type

Tracker

Parameters

  • terms (seq[>=1](str)) – Search terms

  • filters (dict(str:str|bool) | None) – Filters applied to the search

  • total_results (int | None) – Total number of results returned

  • page_results (int | None) – Total number of pages of results

  • context (context_array | None) – Custom context for the event

  • tstamp (int | float | None) – Optional event timestamp in milliseconds

  • event_subject (subject | None) – Optional per event subject

Return type

Tracker

track_struct_event(category: str, action: str, label: Optional[str] = None, property_: Optional[str] = None, value: Optional[float] = None, context: Optional[List[snowplow_tracker.self_describing_json.SelfDescribingJson]] = None, tstamp: Optional[float] = None, event_subject: Optional[snowplow_tracker.subject.Subject] = None) → snowplow_tracker.tracker.Tracker
Parameters

  • category (non_empty_string) – Category of the event

  • action (non_empty_string) – The event itself

  • label (string_or_none) – Refer to the object the action is performed on

  • property (string_or_none) – Property associated with either the action or the object

  • value (int | float | None) – A value associated with the user action

  • context (context_array | None) – Custom context for the event

  • tstamp (int | float | None) – Optional event timestamp in milliseconds

  • event_subject (subject | None) – Optional per event subject

Return type

Tracker

track_unstruct_event(event_json: snowplow_tracker.self_describing_json.SelfDescribingJson, context: Optional[List[snowplow_tracker.self_describing_json.SelfDescribingJson]] = None, tstamp: Optional[float] = None, event_subject: Optional[snowplow_tracker.subject.Subject] = None) → snowplow_tracker.tracker.Tracker
Parameters

  • event_json (self_describing_json) – The properties of the event. Has two field: A “data” field containing the event properties and A “schema” field identifying the schema against which the data is validated

  • context (context_array | None) – Custom context for the event

  • tstamp (int | float | None) – Optional event timestamp in milliseconds

  • event_subject (subject | None) – Optional per event subject

Return type

Tracker

snowplow_tracker.typing module

class snowplow_tracker.typing.EmitterProtocol(*args, **kwargs)

Bases: typing_extensions.Protocol

input(payload: Dict[str, Any]) → None