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', port: int | None = None, method: typing_extensions.Literal[get, post] = 'post', batch_size: int | None = None, on_success: Callable[[List[Dict[str, Any]]], None] | None = None, on_failure: Callable[[int, List[Dict[str, Any]]], None] | None = None, thread_count: int = 1, byte_limit: int | None = None, request_timeout: float | Tuple[float, float] | None = None, max_retry_delay_seconds: int = 60, buffer_capacity: int | None = None, custom_retry_codes: Dict[int, bool] = {}, event_store: EventStore | None = None, session: Session | None = None)

Bases: 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] = 'https', port: int | None = None, method: typing_extensions.Literal[get, post] = 'post', batch_size: int | None = None, on_success: Callable[[List[Dict[str, Any]]], None] | None = None, on_failure: Callable[[int, List[Dict[str, Any]]], None] | None = None, byte_limit: int | None = None, request_timeout: float | Tuple[float, float] | None = None, max_retry_delay_seconds: int = 60, buffer_capacity: int | None = None, custom_retry_codes: Dict[int, bool] = {}, event_store: EventStore | None = None, session: Session | None = None)

Bases: EmitterProtocol

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] = 'https', port: int | None = None, method: typing_extensions.Literal[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

async_flush() → None

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: 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

class snowplow_tracker.emitters.Requester(post: Callable, get: Callable)

Bases: object

get: Callable

post: Callable

snowplow_tracker.payload module

class snowplow_tracker.payload.Payload(dict_: Dict[str, Any] | None = 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_: Dict[str, Any] | None, encode_base64: bool, type_when_encoded: str, type_when_not_encoded: str, json_encoder: Callable[[Any], Any] | None = 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: Dict[str, Any] | List[Dict[str, Any]])

Bases: object

property schema: str

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: Subject | None) → 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) → Subject

Parameters:

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

Return type:

subject

set_domain_session_id(sid: str) → Subject

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

set_domain_session_index(vid: int) → Subject

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

set_domain_user_id(duid: str) → Subject

Set the domain user ID

Parameters:

duid (string) – Domain user ID

Return type:

subject

set_ip_address(ip: str) → Subject

Set the domain user ID

Parameters:

ip (string) – IP address

Return type:

subject

set_lang(lang: str) → Subject

Set language.

Parameters:

lang (string) – Language the application is set to

Return type:

subject

set_network_user_id(nuid: str) → 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]) → 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) → Subject

Parameters:

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

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

Return type:

subject

set_timezone(timezone: str) → Subject

Parameters:

timezone (string) – Timezone as a string

Return type:

subject

set_user_id(user_id: str) → Subject

Parameters:

user_id (string) – User ID

Return type:

subject

set_useragent(ua: str) → Subject

Set the user agent

Parameters:

ua (string) – User agent

Return type:

subject

set_viewport(width: int, height: int) → 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: List[EmitterProtocol] | EmitterProtocol, subject: Subject | None = None, app_id: str | None = None, encode_base64: bool = True, json_encoder: Callable[[Any], Any] | None = None)

Bases: object

add_emitter(emitter: EmitterProtocol) → Tracker

Add a new emitter to which events should be passed

Parameters:

emitter (emitter) – New emitter

Return type:

tracker

complete_payload(event: Event) → Payload

flush(is_async: bool = False) → 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: float | None = 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: Subject | None) → Tracker

Set the subject of the events fired by the tracker

Parameters:

subject (subject | None) – Subject to be tracked

Return type:

tracker

track(event: Event) → str | None

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: str | None = None, category: str | None = None, unit_price: float | None = None, currency: str | None = None, context: List[SelfDescribingJson] | None = None, tstamp: float | None = None, event_subject: Subject | None = None) → 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: str | None = None, tax_value: float | None = None, shipping: float | None = None, city: str | None = None, state: str | None = None, country: str | None = None, currency: str | None = None, items: List[Dict[str, Any]] | None = None, context: List[SelfDescribingJson] | None = None, tstamp: float | None = None, event_subject: Subject | None = None) → 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: str | None = None, category: str | None = None, currency: str | None = None, context: List[SelfDescribingJson] | None = None, tstamp: float | None = None, event_subject: Subject | None = None) → 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: str | None, node_name: typing_extensions.Literal[INPUT, TEXTAREA, SELECT], value: str | None, type_: str | None = None, element_classes: List[str] | Tuple[str, Any] | None = None, context: List[SelfDescribingJson] | None = None, tstamp: float | None = None, event_subject: Subject | None = None) → 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: List[str] | Tuple[str, Any] | None = None, elements: List[Dict[str, Any]] | None = None, context: List[SelfDescribingJson] | None = None, tstamp: float | None = None, event_subject: Subject | None = None) → 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

track_link_click(target_url: str, element_id: str | None = None, element_classes: List[str] | Tuple[str, Any] | None = None, element_target: str | None = None, element_content: str | None = None, context: List[SelfDescribingJson] | None = None, tstamp: float | None = None, event_subject: Subject | None = None) → 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_: str | None = None, type: str | None = None, previous_name: str | None = None, previous_id: str | None = None, previous_type: str | None = None, transition_type: str | None = None, context: List[SelfDescribingJson] | None = None, tstamp: float | None = None, event_subject: Subject | None = None) → 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: str | None = None, referrer: str | None = None, min_x: int | None = None, max_x: int | None = None, min_y: int | None = None, max_y: int | None = None, context: List[SelfDescribingJson] | None = None, tstamp: float | None = None, event_subject: Subject | None = None) → 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: str | None = None, referrer: str | None = None, context: List[SelfDescribingJson] | None = None, tstamp: float | None = None, event_subject: Subject | None = None) → 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: str | None = None, category: str | None = None, unit_price: float | None = None, currency: str | None = None, context: List[SelfDescribingJson] | None = None, tstamp: float | None = None, event_subject: Subject | None = None) → 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: str | None = None, id_: str | None = None, context: List[SelfDescribingJson] | None = None, tstamp: float | None = None, event_subject: Subject | None = None) → 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: SelfDescribingJson, context: List[SelfDescribingJson] | None = None, tstamp: float | None = None, event_subject: Subject | None = None) → 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

track_site_search(terms: Sequence[str], filters: Dict[str, str | bool] | None = None, total_results: int | None = None, page_results: int | None = None, context: List[SelfDescribingJson] | None = None, tstamp: float | None = None, event_subject: Subject | None = None) → 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: str | None = None, property_: str | None = None, value: int | float | None = None, context: List[SelfDescribingJson] | None = None, tstamp: float | None = None, event_subject: Subject | None = None) → 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: SelfDescribingJson, context: List[SelfDescribingJson] | None = None, tstamp: float | None = None, event_subject: Subject | None = None) → 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: Protocol

async_flush() → None

flush() → None

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

sync_flush() → None