sensorswave-sdk 0.1.0

SensorsWave backend SDK for Python

SensorsWave SDK

English | 中文

A lightweight Python SDK for event tracking and A/B testing.

Features

• Event Tracking: Track user events with custom properties
• User Profiles: Set, increment, append, and manage user profile properties
• A/B Testing: Evaluate feature gates, experiments, and feature configs
• Automatic Exposure Logging: Automatically track A/B test impressions

Installation

pip install sensorswave-sdk

Requires Python 3.10 or newer.

Quick Start

Basic Event Tracking

from sensorswave import Config, Properties, SensorsWave, User

# Create client with minimal configuration
client = SensorsWave.create(
    "https://your-endpoint.com",
    "your-source-token",
    Config(),
)
try:
    # Track events
    user = User(login_id="user-123", anon_id="device-456")

    client.track_event(user, "PageView", Properties({
        "page_name": "/home",
    }))
finally:
    client.close()

Enable A/B Testing (Optional)

To enable A/B testing, provide an ABConfig:

from sensorswave import ABConfig, Config, SensorsWave

config = Config(ab=ABConfig(
    project_secret="your-project-secret",
))

client = SensorsWave.create(
    "https://your-endpoint.com",
    "your-source-token",
    config,
)

# Now you can use A/B testing methods
result = client.get_experiment(user, "my_experiment")

# Get parameters from the experiment result
btn_color = result.get_string("button_color", "blue")
show_banner = result.get_bool("show_banner", False)
discount = int(result.get_number("discount_percent", 0))

print(f"Experiment: {result.key}, Button: {btn_color}, "
      f"Banner: {show_banner}, Discount: {discount}%")

API Reference

Client Interface

The SDK provides a SensorsWave client class with methods organized into the following categories:

class SensorsWave:
    # ========== Lifecycle Management ==========

    # Gracefully shut down the client, flushing any pending events.
    # Always call this before your application exits.
    def close(self) -> None: ...

    # ========== User Identity ==========

    # Identify links an anonymous ID with a login ID (signup event).
    # This creates a $Identify event that connects the user's anonymous
    # session with their authenticated identity.
    def identify(self, user: User) -> None: ...

    # ========== Event Tracking ==========

    # track_event tracks a custom event with properties.
    # This is the primary method for tracking user actions.
    def track_event(
        self, user: User, event_name: str, properties: Properties | None = None
    ) -> None: ...

    # ========== User Profile Operations ==========

    # profile_set sets user profile properties ($set).
    # Overwrites existing values.
    def profile_set(self, user: User, properties: Properties) -> None: ...

    # profile_set_once sets user profile properties only if they don't exist
    # ($set_once). Useful for recording first-time values like registration date.
    def profile_set_once(self, user: User, properties: Properties) -> None: ...

    # profile_increment increments numeric user profile properties ($increment).
    # Use for counters like login_count or points.
    def profile_increment(self, user: User, properties: Properties) -> None: ...

    # profile_append appends values to list user profile properties ($append).
    # Allows duplicates in the list.
    def profile_append(
        self, user: User, list_properties: ListProperties
    ) -> None: ...

    # profile_union adds unique values to list user profile properties ($union).
    # Ensures no duplicates in the list.
    def profile_union(
        self, user: User, list_properties: ListProperties
    ) -> None: ...

    # profile_unset removes user profile properties ($unset).
    # Deletes the specified properties from the user profile.
    def profile_unset(self, user: User, *keys: str) -> None: ...

    # profile_delete deletes the entire user profile ($delete).
    # This is irreversible — use with caution.
    def profile_delete(self, user: User) -> None: ...

    # ========== A/B Testing ==========

    # check_feature_gate evaluates a feature gate and returns whether it passes.
    # Returns False if the key doesn't exist or is not a gate type.
    def check_feature_gate(self, user: User, key: str) -> bool: ...

    # get_feature_config evaluates a feature config for a user.
    # Returns an empty result if the key doesn't exist or is not a config type.
    def get_feature_config(self, user: User, key: str) -> ABResult: ...

    # get_experiment evaluates an experiment for a user.
    # Returns an empty result if the key doesn't exist or is not an experiment type.
    def get_experiment(self, user: User, key: str) -> ABResult: ...

    # get_ab_specs exports the current A/B testing metadata as JSON bytes.
    # Use this to cache the A/B configuration for faster startup in future sessions.
    # Pass the returned bytes to ABConfig.load_ab_specs on next initialization.
    def get_ab_specs(self) -> bytes: ...

---

User Type

[!WARNING]

User Identity Requirements (MUST READ)

For ALL methods EXCEPT identify:

• At least one of anon_id or login_id must be non-empty
• If both are provided, login_id takes priority for user identification

For the identify method ONLY:

• Both anon_id AND login_id must be non-empty
• This creates a $Identify event linking anonymous and authenticated identities

User Type Definition

The User dataclass represents a user identity for both event tracking and A/B testing:

from dataclasses import dataclass, field
from typing import Any

@dataclass(frozen=True)
class User:
    anon_id: str = ""              # Anonymous or device ID
    login_id: str = ""             # Login user ID
    ab_user_props: dict[str, Any] = field(default_factory=dict)  # A/B targeting properties

Usage Examples

Creating users with different ID combinations:

from sensorswave import User

# Valid: login_id only (for logged-in users)
user = User(login_id="user-123")

# Valid: anon_id only (for anonymous users)
user = User(anon_id="device-456")

# Valid: both IDs (login_id takes priority for identification)
user = User(login_id="user-123", anon_id="device-456")

# INVALID: neither ID provided — this will FAIL at runtime
user = User()

For the identify method — both IDs are REQUIRED:

# Correct: both IDs provided
client.identify(User(anon_id="device-456", login_id="user-123"))

# INVALID: only one ID — identify will FAIL
client.identify(User(login_id="user-123"))  # missing anon_id

Adding A/B targeting properties:

from dataclasses import replace
from sensorswave import PSP_APP_VER, User

user = User(login_id="user-123", anon_id="device-456")

# User is frozen — use dataclasses.replace to add A/B targeting properties.
user = replace(user, ab_user_props={
    PSP_APP_VER: "11.0",
    "is_premium": True,
})

---

Event Tracking

Identify User

Links an anonymous ID with a login ID (sign-up event).

from sensorswave import User

user = User(anon_id="anon-123", login_id="user-456")
try:
    client.identify(user)
except Exception as exc:
    print(f"Identify failed: {exc}")

Track Custom Event

from sensorswave import Properties, User

user = User(anon_id="anon-123", login_id="user-456")

try:
    client.track_event(user, "Purchase", Properties({
        "product_id":   "SKU-001",
        "total_amount": 99.99,
        "item_count":   2,
    }))
except Exception as exc:
    print(f"Track event failed: {exc}")

Track with Full Event Structure

For advanced scenarios, construct an Event directly and call the internal enqueue path through track_event. The high-level track_event is preferred for normal usage.

from sensorswave import Event, Properties

# Build a fully populated event payload (low-level usage).
event = Event(
    anon_id="anon-123",
    login_id="user-456",
    event="PageView",
    properties=dict(Properties({
        "page_name": "/home",
        "referrer":  "google.com",
    })),
)

# Normalize attaches `time`, `trace_id`, and SDK-lib metadata.
event.normalize()

# In normal code, prefer client.track_event(...). Direct event submission is
# reserved for testing or replay scenarios where the caller controls the
# trace_id / time fields.

---

User Profile Management

Set Profile Properties

from sensorswave import Properties, User

user = User(anon_id="anon-123", login_id="user-456")

try:
    client.profile_set(user, Properties({
        "name":             "John Doe",
        "email":            "[e-mail]",
        "membership_level": 5,
    }))
except Exception as exc:
    print(f"profile_set failed: {exc}")

Set Once (Only if Not Exists)

try:
    client.profile_set_once(user, Properties({
        "first_login_date": "2026-01-20",
    }))
except Exception as exc:
    print(f"profile_set_once failed: {exc}")

Increment Numeric Properties

try:
    client.profile_increment(user, Properties({
        "login_count": 1,
        "points":      100,
    }))
except Exception as exc:
    print(f"profile_increment failed: {exc}")

Append to List Properties

from sensorswave import ListProperties

try:
    client.profile_append(user, ListProperties({
        "tags": ["premium"],
    }))
except Exception as exc:
    print(f"profile_append failed: {exc}")

Union List Properties

try:
    client.profile_union(user, ListProperties({
        "categories": ["sports"],
    }))
except Exception as exc:
    print(f"profile_union failed: {exc}")

Unset Properties

try:
    client.profile_unset(user, "temp_field", "old_field")
except Exception as exc:
    print(f"profile_unset failed: {exc}")

Delete User Profile

try:
    client.profile_delete(user)
except Exception as exc:
    print(f"profile_delete failed: {exc}")

---

A/B Testing

Get Feature Config Values

try:
    result = client.get_feature_config(user, "button_color_config")
except Exception as exc:
    print(f"feature config eval error: {exc}")
    return

# Get string value with fallback
color = result.get_string("color", "blue")

# Get number value with fallback
size = result.get_number("size", 14.0)

# Get boolean value with fallback
enabled = result.get_bool("enabled", False)

# Get list value with fallback
items = result.get_slice("items", [])

# Get map value with fallback
settings = result.get_map("settings", {})

Evaluate Experiment

try:
    result = client.get_experiment(user, "pricing_experiment")
except Exception as exc:
    print(f"experiment eval error: {exc}")
    return

# Get experiment variant parameter
pricing_strategy = result.get_string("strategy", "original")

# Execute different logic based on experiment variant
if pricing_strategy == "original":
    show_original_pricing()
elif pricing_strategy == "discount":
    show_discount_pricing(discount)
elif pricing_strategy == "bundle":
    show_bundle_pricing(int(bundle_size))
else:
    show_original_pricing()

---

Complete API Method Reference

Lifecycle Management

Method	Signature	Description	Example
close	close() -> None	Gracefully shuts down the client and flushes pending events. Always call before application exit.	client.close() (or try / finally)

User Identity

Method	Signature	Parameters	Returns	Description
identify	identify(user: User) -> None	user: User with both anon_id and login_id	None	Creates a $Identify event linking anonymous and authenticated identities

Event Tracking

Method	Signature	Parameters	Returns	Description
track_event	track_event(user: User, event_name: str, properties: Properties | None = None) -> None	user: User identity event_name: Event name properties: Event properties	None	Primary method for tracking user actions with custom properties

User Profile Operations

Method	Signature	Description	Use Case
profile_set	profile_set(user: User, properties: Properties) -> None	Sets or overwrites profile properties	Update user name, email, settings
profile_set_once	profile_set_once(user: User, properties: Properties) -> None	Sets properties only if they don't exist	Record registration date, first source
profile_increment	profile_increment(user: User, properties: Properties) -> None	Increments numeric properties	Login count, points, score
profile_append	profile_append(user: User, list_properties: ListProperties) -> None	Appends to list properties (allows duplicates)	Add purchase history, activity log
profile_union	profile_union(user: User, list_properties: ListProperties) -> None	Adds unique values to list properties	Add interests, tags, categories
profile_unset	profile_unset(user: User, *keys: str) -> None	Removes specified properties	Clear temporary or deprecated fields
profile_delete	profile_delete(user: User) -> None	Deletes entire user profile (irreversible)	GDPR data deletion requests

A/B Testing

Method	Signature	Parameters	Returns	Description
check_feature_gate	check_feature_gate(user: User, key: str) -> bool	user: User, key: Gate key	bool	Evaluates a feature gate. Returns False if key not found or wrong type
get_feature_config	get_feature_config(user: User, key: str) -> ABResult	user: User, key: Config key	ABResult	Evaluates a feature config. Returns empty result if key not found or wrong type
get_experiment	get_experiment(user: User, key: str) -> ABResult	user: User, key: Experiment key	ABResult	Evaluates an experiment. Returns empty result if key not found or wrong type
get_ab_specs	get_ab_specs() -> bytes	None	bytes	Exports current A/B metadata as JSON for caching and faster startup

---

Complex Property Input Conventions

The SDK accepts Object (map / dict) and Object Array (list of dicts) values as event properties and profile_set / profile_set_once profile properties. The SDK is pass-through and does not validate property values; the server may silently truncate, drop, or otherwise sanitize values that exceed the limits below. Callers are responsible for staying within these limits to avoid silent data loss.

Limit	Value	Scope
String value byte length	≤ 1024 (UTF-8 bytes)	Any string property value
Properties per event	≤ 256	Caller-supplied keys in the event properties map
OBJECT_ARRAY element count	≤ 100	List of dicts used as a property value

Exceeding any of these may be silently truncated/dropped by the server.

Profile list operations

profile_append and profile_union are list operations and do not accept Object (dict) or Object Array (list of dicts) values. Pass scalars only. The SDK does not reject complex values here, but the server will treat them as OBJECT_ARRAY, conflicting with list semantics.

---

Configuration Options

Client Config

Field	Description	Default
track_uri_path	Event tracking endpoint path	/in/track
transport	Custom HTTP transport	UrllibTransport()
logger	Custom logger implementation	Console logger
event_queue	Maximum queued tracking events before producers block; None uses the SDK default	500
flush_interval	Background worker flush interval in seconds; flush(), close(), and batch limits also flush pending events	10.0
http_concurrency	Maximum worker-side request concurrency	1
http_timeout	HTTP request timeout in seconds	3.0
http_retry	HTTP retry count	2
gzip_threshold_bytes	Gzip tracking request bodies larger than this threshold	1048576
on_track_fail_handler	Callback invoked on dispatch failure	None
ab	A/B testing configuration	None (disabled)

ABConfig

Field	Description	Default
project_secret	Project secret for authentication	Required
meta_endpoint	A/B metadata server URL	Falls back to main endpoint
meta_uri_path	A/B metadata path	/ab/all4eval
meta_load_interval	Metadata polling interval (seconds)	60.0 (minimum effective interval: 30 seconds)
load_ab_specs	Cached A/B specs from get_ab_specs() for fast startup	None
sticky_handler	Custom sticky session handler	None
meta_loader	Custom metadata loader	None

Advanced: Caching A/B Specs

To improve startup performance, you can cache the A/B specifications and load them upon client initialization.

from sensorswave import ABConfig, Config, SensorsWave

# 1. Get specs from an initialized client
specs = client.get_ab_specs()

# 2. Save specs to persistent storage (e.g. file, database, redis)
# save_to_storage(specs)

# 3. Load specs when creating a new client
saved_specs = load_from_storage()

config = Config(ab=ABConfig(
    project_secret="your-project-secret",
    load_ab_specs=saved_specs,  # Inject cached specs
))

# Client will be immediately ready for A/B evaluation using cached specs
client = SensorsWave.create(
    "https://your-endpoint.com",
    "your-source-token",
    config,
)

---

Predefined Properties

The SDK provides predefined property constants for event tracking and user properties:

# Device and system properties
PSP_APP_VER     = "$app_version"      # Application version
PSP_BROWSER     = "$browser"          # Browser name
PSP_BROWSER_VER = "$browser_version"  # Browser version
PSP_MODEL       = "$model"            # Device model
PSP_IP          = "$ip"               # IP address
PSP_OS          = "$os"               # Operating system: ios/android/harmony
PSP_OS_VER      = "$os_version"       # OS version

# Geographic properties
PSP_COUNTRY  = "$country"   # Country
PSP_PROVINCE = "$province"  # Province/State
PSP_CITY     = "$city"      # City

Usage in events:

from sensorswave import PSP_APP_VER, PSP_COUNTRY, Properties

client.track_event(user, "Purchase", Properties({
    PSP_APP_VER:  "2.1.0",
    PSP_COUNTRY:  "US",
    "product_id": "SKU-001",
}))

Usage in A/B testing:

from dataclasses import replace
from sensorswave import PSP_APP_VER, PSP_COUNTRY, User

user = replace(user, ab_user_props={
    PSP_APP_VER: "2.1.0",
    PSP_COUNTRY: "US",
})

---

Running the Example

Track / Identify / profile_set example:

python3 examples/quickstart.py \
    --source-token=your_token \
    --endpoint=your_event_tracking_endpoint

A/B testing example:

python3 examples/ab_example.py \
    --source-token=your_token \
    --project-secret=your_secret \
    --endpoint=your_event_tracking_endpoint \
    --gate-key=my_feature_gate \
    --experiment-key=my_experiment \
    --feature-config-key=my_feature_config

---

License

Proprietary. All rights reserved. SensorsWave.