Metadata-Version: 2.4
Name: mystekz-live-client
Version: 0.2.0
Summary: Python client for the MyStekz Live API
Author-email: Stekz <info@stez.com>
Project-URL: Homepage, https://gitlab.com/stekz/mystekz_live_clients/mystekz-live-client-python/-/issues
Classifier: Programming Language :: Python :: 3
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Requires-Python: >=3.7
Description-Content-Type: text/markdown
Requires-Dist: python-dotenv>=0.21.1
Requires-Dist: requests
Provides-Extra: test
Requires-Dist: pytest; extra == "test"
Requires-Dist: pytest-mock; extra == "test"

# MyStekz Live Python Client

This is the official Python client for the MyStekz Live API. It allows you to easily track and ingest events into the MyStekz platform from your Python applications.

## Features

-   **Event Tracking**: Track events with automatic buffering and batching for optimal performance.
-   **Metrics Retrieval**: Get aggregated metrics for Steks, Domains, Products, or Processes.
-   **Release Management**: List all available releases for a Stek.
-   **Efficient Buffering**: Events are stored in an internal memory buffer, making `track()` calls non-blocking and fast.
-   **Automatic Batching**: Buffered events are sent to the API in batches to minimize network overhead and improve performance.
-   **Reliability**: Built-in retry logic with exponential backoff handles temporary network issues and server errors (5xx, 429) automatically.
-   **Thread Safe**: Designed to be used safely across multiple threads.

## Installation

```bash
pip install mystekz-live-client
```

*Note: If you are installing from source:*
```bash
pip install .
```

## How It Works

1.  **Initialization**: You create an instance of `MyStekzClient` with your authentication credentials (`api_key`, `organization`, `stek`) and configuration options.
2.  **Tracking**: When you call `client.track(...)`, the event is validated and immediately added to an internal queue. The control returns to your application instantly.
3.  **Buffering & Flushing**:
    *   The client monitors the queue size. If it reaches `max_batch_size` (default: 10), a batch of events is sent immediately.
    *   A background timer ensures that events don't sit in the queue too long. If events are pending for more than `flush_interval` seconds (default: 5.0), they are flushed automatically.
4.  **Graceful Shutdown**: Calling `client.close()` stops the background timers and flushes any remaining events in the buffer to ensure no data is lost upon application exit.

## Usage

```python
from mystekz_live_client import MyStekzClient
import os

# Initialize the client
client = MyStekzClient(
    endpoint=os.environ.get("MYSTEKZ_ENDPOINT"),
    api_key=os.environ.get("MYSTEKZ_API_KEY"),
    organization=os.environ.get("MYSTEKZ_ORGANIZATION"),
    stek=os.environ.get("MYSTEKZ_STEK"),
    stek_release=os.environ.get("MYSTEKZ_STEK_RELEASE"),
    # Optional configuration:
    # flush_interval=5.0,  # Flush every 5 seconds
    # max_batch_size=10,   # Flush when 10 events are queued
)
```

### Tracking Events

```python
client.track(
    event_type="user.signed_up",
    payload={
        "user_id": "u_12345",
        "email": "user@example.com",
        "plan": "pro"
    },
    # Optional context fields:
    business_key="u_12345",
    domain_id="identity-mgt",
    product_id="auth-service",
    process_id="registration-flow",
    process_instance_id="pi_abc_123",
    process_element_id="task_verify_email",
    # Optional override:
    stek_release="v1.1.0-beta",
    # Optional explicit timestamp (datetime object):
    timestamp=datetime.now(timezone.utc)
)
```

The `track()` method accepts several optional parameters to provide more context for the event:

- `event_type`: Defaults to `"generic.event"` if not provided
- `payload`: Event data (default: empty dict)
- `timestamp`: Defaults to current UTC time if not provided (naive datetimes assumed to be UTC)
- `process_instance_id`: Auto-generated UUID if not provided (required by backend)
- Context fields: `business_key`, `domain_id`, `product_id`, `process_id`, `process_element_id`

**Note**: The `process_instance_id` field is required by the backend. If you don't provide one, the client will automatically generate a unique UUID for each event. You can provide your own `process_instance_id` if you want to group related events together.

Providing any of `process_element_id`, `process_id`, or `product_id` allows the backend to automatically resolve the full parent hierarchy.

```python
# Track with default event_type
client.track(payload={"info": "Simple generic event"})

# Track another event
client.track(
    event_type="payment.processed",
    payload={"amount": 99.00, "currency": "USD"}
)
```

### Compliance Proof for Task and Process Completion

The MyStekz API supports attaching compliance evidence to events for audit purposes. This is particularly useful for `task.completed` and `process.completed` events where auditors may need to verify the results.

Compliance evidence is added to the event `payload` using the `ComplianceProof` structure, which contains:
- **evidence**: A list of evidence items (minimum 1 required)
- **metadata**: Optional dictionary for additional context

Each evidence item has:
- **type**: Either `"uri"` (link to external document) or `"description"` (text description)
- **value**: The evidence content (URL for uri type, text for description type)
- **description**: Human-readable explanation of what this evidence proves
- **timestamp**: When the evidence was created (auto-generated if not provided)
- **created_by**: Optional user ID or email of who created the evidence

#### Single Evidence Item Example

```python
from mystekz_live_client import (
    MyStekzClient,
    ComplianceProof,
    ComplianceEvidenceItem,
    ComplianceEvidenceType
)

# Create a URI evidence item pointing to external proof
evidence = ComplianceEvidenceItem(
    type=ComplianceEvidenceType.URI,
    value="https://storage.example.com/audit/task_results.json",
    description="Test results document for task completion",
    created_by="test-runner@example.com"
)

# Create compliance proof with metadata
compliance_proof = ComplianceProof(
    evidence=[evidence],
    metadata={"control_id": "QA-001", "reviewer": "QA Team"}
)

# Add to payload and track event
payload = {
    "task_id": "task_789",
    "status": "completed"
}
compliance_proof.add_to_payload(payload)

client.track(
    event_type="task.completed",
    payload=payload,
    process_element_id="task-element-id"
)
```

#### Multiple Evidence Items Example

```python
from datetime import datetime, timezone

# Create multiple evidence items for comprehensive proof
evidence1 = ComplianceEvidenceItem(
    type=ComplianceEvidenceType.URI,
    value="https://storage.example.com/audit/completion_certificate.pdf",
    description="Process completion certificate signed by manager",
    timestamp=datetime(2026, 1, 31, 10, 30, 0, tzinfo=timezone.utc),
    created_by="manager@example.com"
)

evidence2 = ComplianceEvidenceItem(
    type=ComplianceEvidenceType.DESCRIPTION,
    value="Manual verification performed. All steps completed according to SOP-2024-001.",
    description="Manual verification notes",
    created_by="auditor@example.com"
)

evidence3 = ComplianceEvidenceItem(
    type=ComplianceEvidenceType.URI,
    value="https://storage.example.com/audit/screenshots.zip",
    description="Screenshot evidence archive",
    created_by="operator@example.com"
)

# Create compliance proof with all evidence
compliance_proof = ComplianceProof(
    evidence=[evidence1, evidence2, evidence3],
    metadata={
        "control_id": "SOC2-CC6.1",
        "reviewer": "John Doe",
        "audit_period": "Q1-2026"
    }
)

payload = {"process_id": "proc_999", "status": "success"}
compliance_proof.add_to_payload(payload)

client.track(
    event_type="process.completed",
    payload=payload,
    process_id="process-id"
)
```

#### Manual Payload Construction

You can also construct the compliance proof structure manually:

```python
payload = {
    "task_id": "task_123",
    "status": "completed",
    "compliance_proof": {
        "evidence": [
            {
                "id": "evidence_001",
                "type": "uri",
                "value": "https://example.com/proof.pdf",
                "description": "Approval document",
                "timestamp": "2026-01-31T10:30:00Z",
                "created_by": "user@example.com"
            }
        ],
        "metadata": {
            "control_id": "CTRL-001"
        }
    }
}

client.track(
    event_type="task.completed",
    payload=payload
)
```

### Retrieving Metrics

Get aggregated metrics for a specific entity over a time range:

```python
# Get metrics for a Stek (last 24 hours)
metrics = client.get_metrics(stek_id="your-stek-id", hours=24)

# Get metrics for a Domain (last 48 hours)
metrics = client.get_metrics(domain_id="your-domain-id", hours=48)

# Get metrics for a Product, optionally filtered by release
metrics = client.get_metrics(
    product_id="your-product-id",
    hours=72,
    stek_release="v1.0.0"
)

# Get metrics for a Process
metrics = client.get_metrics(process_id="your-process-id", hours=168)

# Access metrics data
print(f"Total events: {metrics.total_events}")
print(f"Error rate: {metrics.error_rate:.2%}")
print(f"Events by type: {metrics.events_by_type}")
print(f"Time range: {metrics.time_range_start} to {metrics.time_range_end}")
```

**Note**: Exactly one of `stek_id`, `domain_id`, `product_id`, or `process_id` must be provided. The `hours` parameter must be between 1 and 168 (1 week).

### Listing Available Releases

Get all unique release versions for your Stek:

```python
releases = client.get_releases()
print(f"Available releases: {releases}")
# Output: ["v1.0.0", "v1.1.0", "v2.0.0"]
```

### Cleanup

Always close the client when done to ensure all buffered events are sent:

```python
# Manually flush without closing
client.flush()

# Ensure all events are sent before exiting
client.close()
```

## Configuration Options

| Parameter | Type | Default | Description |
| :--- | :--- | :--- | :--- |
| `endpoint` | `str` | **Required** | The MyStekz Ingest API URL. |
| `api_key` | `str` | **Required** | Your MyStekz API Key. |
| `organization`| `str` | **Required** | Your Organization ID. |
| `stek` | `str` | **Required** | Your Stek ID. |
| `stek_release`| `str` | **Required** | Release version tag for your events. |
| `flush_interval`| `float` | `5.0` | Maximum time (in seconds) to hold events before flushing. |
| `max_batch_size`| `int` | `10` | Maximum number of events to batch together in one request. |
| `max_queue_size`| `int` | `1000` | Maximum number of events to keep in memory buffer. |
| `max_retries` | `int` | `3` | Number of times to retry failed requests (5xx/429 errors). |
