Define attributes and service with Python

In this section, you'll define attributes that calculate real-time user behavior metrics from ecommerce events. These attributes will track product views, cart additions, and cart value.

You'll create three attributes to track user shopping behavior:

• count_product_views counts the number of product view events
• count_add_to_cart counts the number of add-to-cart events
• total_cart_value sums the prices of items added to the cart

These attributes use the Snowplow ecommerce plugin, which tracks standardized ecommerce events.

Import required classes

Start by importing the classes you'll need from the Signals Python SDK:

from snowplow_signals import (
    Attribute,
    Event,
    Criterion,
    Criteria,
    EventProperty,
    EntityProperty,
    StreamAttributeGroup,
    Service,
    domain_userid,
)

Define the attributes

Each attribute defines which event it will be calculated from, and what kind of aggregation will be performed.

Product views counter

This attribute counts product view events:

count_product_views = Attribute(
    name="count_product_views",
    type="int32",
    events=[Event(name="snowplow_ecommerce_action")],
    criteria=Criteria(
        all=[
            Criterion.eq(
                property=EventProperty(
                    vendor="com.snowplowanalytics.snowplow.ecommerce",
                    name="snowplow_ecommerce_action",
                    major_version=1,
                    path="type",
                ),
                value="product_view",
            )
        ]
    ),
    aggregation="counter",
)

This attribute:

• Uses the snowplow_ecommerce_action event
• Filters for events where the action type is product_view
• Counts matching events using the counter aggregation

Add to cart counter

This attribute counts when users add items to their cart:

count_add_to_cart = Attribute(
    name="count_add_to_cart",
    type="int32",
    events=[Event(name="snowplow_ecommerce_action")],
    criteria=Criteria(
        all=[
            Criterion.eq(
                property=EventProperty(
                    vendor="com.snowplowanalytics.snowplow.ecommerce",
                    name="snowplow_ecommerce_action",
                    major_version=1,
                    path="type",
                ),
                value="add_to_cart",
            )
        ]
    ),
    aggregation="counter",
)

Total cart value

This attribute sums the prices of products added to the cart:

total_cart_value = Attribute(
    name="total_cart_value",
    type="double",
    events=[Event(name="snowplow_ecommerce_action")],
    criteria=Criteria(
        all=[
            Criterion.eq(
                property=EventProperty(
                    vendor="com.snowplowanalytics.snowplow.ecommerce",
                    name="snowplow_ecommerce_action",
                    major_version=1,
                    path="type",
                ),
                value="add_to_cart",
            )
        ]
    ),
    property=EntityProperty(
        vendor="com.snowplowanalytics.snowplow.ecommerce",
        name="product",
        major_version=1,
        path="price",
    ),
    aggregation="sum",
)

This attribute:

• Filters for add_to_cart events
• Extracts the price field from the product entity
• Sums the prices using the sum aggregation

Create an attribute group

Attribute groups organize related attributes together. They can be considered as "tables" of attributes.

Create a group to hold your ecommerce attributes:

attribute_group = StreamAttributeGroup(
    name="ecom_attributes",
    version=1,
    attribute_key=domain_userid,
    attributes=[
        count_product_views,
        count_add_to_cart,
        total_cart_value,
    ],
    owner="user@company.com",
)

The attribute_key parameter specifies the user identifier that the attributes are grouped by. In this case, domain_userid means the attributes track behavior for each anonymous user.

This is a StreamAttributeGroup, because Signals will process events from the real-time event stream.

note

Attribute groups are immutable and versioned. If you need to modify attributes, create a new version of the group.

Create a service

Services provide an interface for applications to retrieve attributes. Create a service that includes your attribute group:

stream_service = Service(
    name="ecom_attributes",
    attribute_groups=[attribute_group],
    owner="user@company.com",
)

Publish to Signals

Now publish both the attribute group and service to your Signals instance:

sp_signals.publish([attribute_group, stream_service])

Once published, Signals begins calculating these attributes in real time as events arrive at the Collector.

tip

You can retrieve attribute values for a specific user by calling:

stream_service.get_attributes(
    signals=sp_signals,
    attribute_key="domain_userid",
    identifier="your-domain-userid-here", # the value will be a UUID
)

Your attributes are now live and ready to power personalization experiences. Next, you'll define interventions that trigger based on these attribute values.