mystekz-live-client 0.2.0

Python client for the MyStekz Live API

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

pip install mystekz-live-client

Note: If you are installing from source:

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

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

client.track(
    event_type="user.signed_up",
    payload={
        "user_id": "u_12345",
        "email": "[e-mail]",
        "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.

# 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

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="[e-mail]"
)

# 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

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="[e-mail]"
)

evidence2 = ComplianceEvidenceItem(
    type=ComplianceEvidenceType.DESCRIPTION,
    value="Manual verification performed. All steps completed according to SOP-2024-001.",
    description="Manual verification notes",
    created_by="[e-mail]"
)

evidence3 = ComplianceEvidenceItem(
    type=ComplianceEvidenceType.URI,
    value="https://storage.example.com/audit/screenshots.zip",
    description="Screenshot evidence archive",
    created_by="[e-mail]"
)

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

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": "[e-mail]"
            }
        ],
        "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:

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

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:

# 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).