Metadata-Version: 2.4
Name: sensorswave-sdk
Version: 0.1.0
Summary: SensorsWave backend SDK for Python
Project-URL: Homepage, https://github.com/sensorswave/sdk-python
Project-URL: Source, https://github.com/sensorswave/sdk-python
Author: SensorsWave
License: Proprietary
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: Other/Proprietary License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Requires-Python: >=3.10
Provides-Extra: dev
Requires-Dist: build>=1.0; extra == 'dev'
Requires-Dist: mypy>=1.0; extra == 'dev'
Requires-Dist: pytest-xdist>=3; extra == 'dev'
Requires-Dist: pytest>=8; extra == 'dev'
Requires-Dist: ruff>=0.9; extra == 'dev'
Requires-Dist: twine>=4.0; extra == 'dev'
Description-Content-Type: text/markdown

# SensorsWave SDK

[![Release](https://img.shields.io/github/v/release/sensorswave/sdk-python.svg)](https://github.com/sensorswave/sdk-python/releases)
[![PyPI](https://img.shields.io/pypi/v/sensorswave-sdk.svg)](https://pypi.org/project/sensorswave-sdk/)
[![Python Versions](https://img.shields.io/pypi/pyversions/sensorswave-sdk.svg)](https://pypi.org/project/sensorswave-sdk/)
[![Test](https://github.com/sensorswave/sdk-python/actions/workflows/test.yml/badge.svg)](https://github.com/sensorswave/sdk-python/actions/workflows/test.yml)
[![Lint](https://github.com/sensorswave/sdk-python/actions/workflows/lint.yml/badge.svg)](https://github.com/sensorswave/sdk-python/actions/workflows/lint.yml)

**English** | [中文](README.zh-CN.md)

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

```bash
pip install sensorswave-sdk
```

Requires Python 3.10 or newer.

## Quick Start

### Basic Event Tracking

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

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

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

```python
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:**

```python
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:**

```python
# 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:**

```python
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).

```python
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

```python
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.

```python
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

```python
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":            "john@example.com",
        "membership_level": 5,
    }))
except Exception as exc:
    print(f"profile_set failed: {exc}")
```

### Set Once (Only if Not Exists)

```python
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

```python
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

```python
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

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

### Unset Properties

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

### Delete User Profile

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

---

## A/B Testing

### Get Feature Config Values

```python
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

```python
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<br/>`event_name`: Event name<br/>`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.

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

```python
# 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:

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

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

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

A/B testing example:

```bash
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.
