tessl/pypi-google-cloud-retail

Python client library for Google Cloud Retail API enabling end-to-end personalized recommendation systems

—

Pending

Overview

Eval results

Files

User Event Tracking

Name: tessl/pypi-google-cloud-retail
Author: tessl

Collection and management of user interaction events that power recommendation algorithms and analytics. The User Event Service enables comprehensive tracking of customer behavior including page views, purchases, add-to-cart actions, and custom events.

Capabilities

Event Collection

Real-time and batch collection of user interaction events for recommendation training and analytics.

class UserEventServiceClient:
    def write_user_event(self, request: WriteUserEventRequest) -> UserEvent:
        """
        Records a single user event in real-time.
        
        Args:
            request: Contains parent catalog and user event data
            
        Returns:
            UserEvent: The recorded event with server-generated metadata
            
        Raises:
            InvalidArgument: If event data is invalid or incomplete
        """

    def collect_user_event(self, request: CollectUserEventRequest) -> HttpBody:
        """
        Collects user events via HTTP GET request (typically used with pixel tracking).
        
        Args:
            request: Contains parent, user_event query parameter, and optional parameters
            
        Returns:
            HttpBody: Response suitable for pixel tracking (usually empty)
        """

    def import_user_events(self, request: ImportUserEventsRequest) -> Operation:
        """
        Imports user events in bulk from external sources (long-running operation).
        
        Args:
            request: Contains parent, input configuration, and error handling settings
            
        Returns:
            Operation: Resolves to ImportUserEventsResponse with import statistics
        """

    def rejoin_user_events(self, request: RejoinUserEventsRequest) -> Operation:
        """
        Re-associates user events with updated product catalog (long-running operation).
        
        Args:
            request: Contains parent and rejoin configuration
            
        Returns:
            Operation: Resolves to RejoinUserEventsResponse with rejoin statistics
        """

    def purge_user_events(self, request: PurgeUserEventsRequest) -> Operation:
        """
        Permanently deletes user events matching filter criteria (long-running operation).
        
        Args:
            request: Contains parent, filter, and force flag
            
        Returns:
            Operation: Resolves to PurgeUserEventsResponse with purge count
        """

class UserEventServiceAsyncClient:
    async def write_user_event(self, request: WriteUserEventRequest) -> UserEvent:
        """
        Records a single user event in real-time (async).
        
        Args:
            request: Contains parent catalog and user event data
            
        Returns:
            UserEvent: The recorded event with server-generated metadata
            
        Raises:
            InvalidArgument: If event data is invalid or incomplete
        """

    async def collect_user_event(self, request: CollectUserEventRequest) -> HttpBody:
        """
        Collects user events via HTTP GET request - async version (typically used with pixel tracking).
        
        Args:
            request: Contains parent, user_event query parameter, and optional parameters
            
        Returns:
            HttpBody: Response suitable for pixel tracking (usually empty)
        """

    async def import_user_events(self, request: ImportUserEventsRequest) -> Operation:
        """
        Imports user events in bulk from external sources - async version (long-running operation).
        
        Args:
            request: Contains parent, input configuration, and error handling settings
            
        Returns:
            Operation: Resolves to ImportUserEventsResponse with import statistics
        """

    async def rejoin_user_events(self, request: RejoinUserEventsRequest) -> Operation:
        """
        Re-associates user events with updated product catalog - async version (long-running operation).
        
        Args:
            request: Contains parent and rejoin configuration
            
        Returns:
            Operation: Resolves to RejoinUserEventsResponse with rejoin statistics
        """

    async def purge_user_events(self, request: PurgeUserEventsRequest) -> Operation:
        """
        Permanently deletes user events matching filter criteria - async version (long-running operation).
        
        Args:
            request: Contains parent, filter, and force flag
            
        Returns:
            Operation: Resolves to PurgeUserEventsResponse with purge count
        """

Data Types

User Event

Comprehensive user interaction event with product details and context information.

class UserEvent:
    event_type: str  # Event type (required): page-view, add-to-cart, purchase, etc.
    visitor_id: str  # Unique visitor identifier (required)
    session_id: str  # Session identifier for grouping events
    event_time: Timestamp  # When event occurred (auto-set if not provided)
    experiment_ids: List[str]  # A/B test experiment identifiers
    attribution_token: str  # Token from search/prediction response
    product_details: List[ProductDetail]  # Products involved in the event
    completion_detail: CompletionDetail  # Details for search completion events
    attributes: Dict[str, CustomAttribute]  # Custom event attributes
    cart_id: str  # Shopping cart identifier
    purchase_transaction: PurchaseTransaction  # Transaction details for purchase events
    search_query: str  # Search query for search events
    filter: str  # Applied filters for search events
    order_by: str  # Sort order for search events
    offset: int  # Pagination offset for search events
    page_categories: List[str]  # Page context categories
    user_info: UserInfo  # User context information
    uri: str  # Page URI where event occurred
    referrer_uri: str  # Referrer page URI
    page_view_id: str  # Unique page view identifier

Product Detail

Product-specific information within user events including quantities and custom data.

class ProductDetail:
    product: Product  # Product information (id is required, other fields optional)
    quantity: int  # Quantity involved in the event (default: 1)
    
    # For purchases, ratings, and other detailed interactions
    rating: float  # User rating for the product (1-5 scale)
    value: Money  # Monetary value associated with this product
    cost: Money  # Cost associated with this product
    promotion_id: str  # Promotion identifier if applicable

Purchase Transaction

Comprehensive transaction information for purchase events and revenue tracking.

class PurchaseTransaction:
    id: str  # Unique transaction identifier
    revenue: float  # Total transaction revenue (required)
    tax: float  # Tax amount
    cost: float  # Total cost of goods sold
    currency_code: str  # ISO 4217 currency code (required)
    
    # Additional transaction metadata
    transaction_attributes: Dict[str, CustomAttribute]  # Custom transaction attributes

Completion Detail

Details for search query auto-completion interaction events.

class CompletionDetail:
    completion_attribution_token: str  # Attribution token from completion response
    selected_suggestion: str  # The completion suggestion selected by user
    selected_position: int  # Position of selected suggestion in completion list

Request Types

Comprehensive request configurations for various user event operations.

class WriteUserEventRequest:
    parent: str  # Catalog resource name (required)
    user_event: UserEvent  # User event to record (required)
    write_async: bool  # Whether to write asynchronously (default: false)

class CollectUserEventRequest:
    parent: str  # Catalog resource name (required)
    user_event: str  # URL-encoded user event data (required)
    uri: str  # Page URI where event occurred
    ets: int  # Event timestamp (Unix timestamp in milliseconds)
    raw_json: str  # Raw JSON event data (alternative to user_event parameter)

class ImportUserEventsRequest:
    parent: str  # Catalog resource name (required)
    input_config: UserEventInputConfig  # Data source configuration (required)
    errors_config: ImportErrorsConfig  # Error handling configuration

class RejoinUserEventsRequest:
    parent: str  # Catalog resource name (required)
    rejoin_type: RejoinUserEventsRequestRejoinType  # Type of rejoin operation

class PurgeUserEventsRequest:
    parent: str  # Catalog resource name (required)
    filter: str  # Filter expression for events to purge (required)
    force: bool  # Force purge without confirmation (required)

Import Configuration

Configuration for bulk user event import from various data sources.

class UserEventInputConfig:
    user_event_inline_source: UserEventInlineSource  # Inline event data
    gcs_source: GcsSource  # Google Cloud Storage source
    big_query_source: BigQuerySource  # BigQuery source

class UserEventInlineSource:
    user_events: List[UserEvent]  # List of events to import

class UserEventImportSummary:
    joined_events_count: int  # Number of events successfully joined with products
    unjoined_events_count: int  # Number of events that couldn't be joined

Event Types

Standard event types for different user interactions and custom event support.

# Standard Event Types
EVENT_TYPE_PAGE_VIEW = "page-view"  # User viewed a page/product
EVENT_TYPE_ADD_TO_CART = "add-to-cart"  # User added item to cart
EVENT_TYPE_REMOVE_FROM_CART = "remove-from-cart"  # User removed item from cart
EVENT_TYPE_PURCHASE_COMPLETE = "purchase-complete"  # User completed purchase
EVENT_TYPE_SEARCH = "search"  # User performed search
EVENT_TYPE_DETAIL_PAGE_VIEW = "detail-page-view"  # User viewed product detail page
EVENT_TYPE_HOME_PAGE_VIEW = "home-page-view"  # User viewed homepage
EVENT_TYPE_CATEGORY_PAGE_VIEW = "category-page-view"  # User viewed category page
EVENT_TYPE_SHOPPING_CART_PAGE_VIEW = "shopping-cart-page-view"  # User viewed cart
EVENT_TYPE_LOGIN = "login"  # User logged in
EVENT_TYPE_SIGNUP = "signup"  # User signed up

# Custom event types are also supported - use descriptive names

Usage Examples

Recording Real-time User Events

from google.cloud import retail
from google.protobuf.timestamp_pb2 import Timestamp
import time

client = retail.UserEventServiceClient()

# Record a product page view
page_view_event = retail.UserEvent(
    event_type="detail-page-view",
    visitor_id="visitor-123",
    session_id="session-456",
    user_info=retail.UserInfo(
        user_id="user-789",
        ip_address="192.168.1.100",
        user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    ),
    product_details=[
        retail.ProductDetail(
            product=retail.Product(id="product-123"),
            quantity=1
        )
    ],
    uri="https://example.com/products/product-123",
    page_categories=["Electronics", "Laptops"]
)

request = retail.WriteUserEventRequest(
    parent="projects/my-project/locations/global/catalogs/default_catalog",
    user_event=page_view_event
)

recorded_event = client.write_user_event(request=request)
print(f"Recorded event: {recorded_event.event_type} at {recorded_event.event_time}")

Recording Purchase Events

# Record a purchase transaction
purchase_event = retail.UserEvent(
    event_type="purchase-complete",
    visitor_id="visitor-123",
    session_id="session-456",
    user_info=retail.UserInfo(
        user_id="user-789",
        ip_address="192.168.1.100"
    ),
    product_details=[
        retail.ProductDetail(
            product=retail.Product(id="laptop-456"),
            quantity=1
        ),
        retail.ProductDetail(
            product=retail.Product(id="mouse-789"),
            quantity=2
        )
    ],
    purchase_transaction=retail.PurchaseTransaction(
        id="txn-12345",
        revenue=1599.97,
        tax=128.00,
        cost=1200.00,
        currency_code="USD"
    ),
    uri="https://example.com/checkout/success",
    cart_id="cart-abc123"
)

request = retail.WriteUserEventRequest(
    parent="projects/my-project/locations/global/catalogs/default_catalog",
    user_event=purchase_event
)

recorded_event = client.write_user_event(request=request)
print(f"Recorded purchase: ${purchase_event.purchase_transaction.revenue}")

Recording Search Events

# Record a search event with results
search_event = retail.UserEvent(
    event_type="search",
    visitor_id="visitor-123",
    session_id="session-456",
    search_query="gaming laptop",
    filter='(categories: ANY("Electronics")) AND (price_info.price: [500, 2000])',
    order_by="price_info.price desc",
    offset=0,
    user_info=retail.UserInfo(
        user_id="user-789"
    ),
    uri="https://example.com/search?q=gaming+laptop",
    page_categories=["Search Results"]
)

request = retail.WriteUserEventRequest(
    parent="projects/my-project/locations/global/catalogs/default_catalog",
    user_event=search_event
)

recorded_event = client.write_user_event(request=request)
print(f"Recorded search for: {search_event.search_query}")

Recording Add-to-Cart Events

# Record add-to-cart event
add_to_cart_event = retail.UserEvent(
    event_type="add-to-cart",
    visitor_id="visitor-123",
    session_id="session-456",
    user_info=retail.UserInfo(
        user_id="user-789"
    ),
    product_details=[
        retail.ProductDetail(
            product=retail.Product(id="laptop-456"),
            quantity=1
        )
    ],
    cart_id="cart-abc123",
    uri="https://example.com/products/laptop-456",
    attribution_token="attribution_token_from_search_or_prediction"  # If user came from search/recommendations
)

request = retail.WriteUserEventRequest(
    parent="projects/my-project/locations/global/catalogs/default_catalog",
    user_event=add_to_cart_event
)

recorded_event = client.write_user_event(request=request)
print(f"Recorded add-to-cart for product: {add_to_cart_event.product_details[0].product.id}")

Bulk Import User Events

# Import events from Google Cloud Storage
input_config = retail.UserEventInputConfig(
    gcs_source=retail.GcsSource(
        input_uris=["gs://my-bucket/user-events/events-2024-01.jsonl"]
    )
)

errors_config = retail.ImportErrorsConfig(
    gcs_prefix="gs://my-bucket/import-errors/"
)

request = retail.ImportUserEventsRequest(
    parent="projects/my-project/locations/global/catalogs/default_catalog",
    input_config=input_config,
    errors_config=errors_config
)

operation = client.import_user_events(request=request)
print(f"Import operation: {operation.name}")

# Wait for import to complete
result = operation.result()
print(f"Import completed:")
print(f"- Joined events: {result.import_summary.joined_events_count}")
print(f"- Unjoined events: {result.import_summary.unjoined_events_count}")
print(f"- Total errors: {len(result.error_samples)}")

Inline Event Import

# Import events directly from code
events_to_import = [
    retail.UserEvent(
        event_type="page-view",
        visitor_id="visitor-111",
        product_details=[retail.ProductDetail(product=retail.Product(id="product-1"))]
    ),
    retail.UserEvent(
        event_type="add-to-cart",
        visitor_id="visitor-111",
        product_details=[retail.ProductDetail(product=retail.Product(id="product-1"), quantity=2)]
    ),
    retail.UserEvent(
        event_type="purchase-complete",
        visitor_id="visitor-111",
        product_details=[retail.ProductDetail(product=retail.Product(id="product-1"), quantity=2)],
        purchase_transaction=retail.PurchaseTransaction(
            id="txn-inline-1",
            revenue=199.98,
            currency_code="USD"
        )
    )
]

input_config = retail.UserEventInputConfig(
    user_event_inline_source=retail.UserEventInlineSource(
        user_events=events_to_import
    )
)

request = retail.ImportUserEventsRequest(
    parent="projects/my-project/locations/global/catalogs/default_catalog",
    input_config=input_config
)

operation = client.import_user_events(request=request)
result = operation.result()
print(f"Imported {result.import_summary.joined_events_count} events")

Event Data Maintenance

# Rejoin user events after product catalog updates
rejoin_request = retail.RejoinUserEventsRequest(
    parent="projects/my-project/locations/global/catalogs/default_catalog",
    rejoin_type=retail.RejoinUserEventsRequest.RejoinType.UNJOINED_EVENTS
)

operation = client.rejoin_user_events(request=rejoin_request)
rejoin_result = operation.result()
print(f"Rejoined {rejoin_result.rejoined_user_events_count} events")

# Purge old user events (be very careful with this operation!)
purge_request = retail.PurgeUserEventsRequest(
    parent="projects/my-project/locations/global/catalogs/default_catalog",
    filter='(event_time: ["2023-01-01T00:00:00Z", "2023-12-31T23:59:59Z"])',  # Purge 2023 events
    force=True
)

operation = client.purge_user_events(request=purge_request)
purge_result = operation.result()
print(f"Purged {purge_result.purged_events_count} events")

Pixel Tracking Integration

# For web pixel tracking, you would typically use the collect_user_event method
# This is usually called from JavaScript, but here's how to use it from Python:

import urllib.parse

# Construct event data
event_data = {
    "eventType": "page-view",
    "visitorId": "visitor-123",
    "productDetails": [{"product": {"id": "product-456"}}],
    "uri": "https://example.com/products/product-456"
}

# URL encode the event data
encoded_event = urllib.parse.quote(json.dumps(event_data))

request = retail.CollectUserEventRequest(
    parent="projects/my-project/locations/global/catalogs/default_catalog",
    user_event=encoded_event,
    uri="https://example.com/products/product-456"
)

# This returns HttpBody suitable for pixel tracking
response = client.collect_user_event(request=request)
print("Event collected via pixel tracking")

Install with Tessl CLI

npx tessl i tessl/pypi-google-cloud-retail

docs

advanced-features.md

ai-recommendations.md

analytics-export.md

catalog-config.md

index.md

product-management.md

search-discovery.md

user-events.md

tile.json