pyrx-synapse 0.1.1

PYRX Synapse Python SDK — async + sync client for event tracking, contacts, and email

pyrx-synapse

Python SDK for the PYRX Synapse customer communications platform. Provides both synchronous and asynchronous clients for event tracking, contact management, email sending, and template management.

Install

pip install pyrx-synapse

Quick Start

Sync

from pyrx_synapse import Synapse

with Synapse(api_key="gck_live_xxx", workspace_id="ws_xxx") as client:
    # Track an event
    client.track(external_id="user_123", event_name="purchase", attributes={"amount": 99.99})

    # Identify a contact
    client.identify(external_id="user_123", email="[e-mail]", properties={"tier": "pro"})

    # Send a transactional email
    client.send(
        template_slug="welcome",
        to={"user_id": "user_123", "email": "[e-mail]"},
        attributes={"first_name": "Jane"},
    )

    # Manage contacts
    contacts = client.contacts.list(search="jane", page=1, per_page=50)
    contact = client.contacts.get(contact_id="uuid")
    client.contacts.update("user_123", {"properties": {"tier": "enterprise"}})
    client.contacts.delete("user_123")

    # Manage templates
    templates = client.templates.list()
    template = client.templates.get("welcome")
    preview = client.templates.preview("welcome", {"contact": {"first_name": "Jane"}, "event": {}})

Async

from pyrx_synapse import AsyncSynapse

async with AsyncSynapse(api_key="gck_live_xxx", workspace_id="ws_xxx") as client:
    await client.track(external_id="user_123", event_name="purchase")
    await client.identify(external_id="user_123", email="[e-mail]")
    await client.send(template_slug="welcome", to={"user_id": "u1", "email": "[e-mail]"})

Webhook Verification

from pyrx_synapse import verify_webhook

event = verify_webhook(
    payload=request.body,
    headers={
        "svix-id": request.headers["svix-id"],
        "svix-timestamp": request.headers["svix-timestamp"],
        "svix-signature": request.headers["svix-signature"],
    },
    secret="whsec_xxx",
)
print(event["type"])  # e.g. "contact.created"

Error Handling

from pyrx_synapse import (
    Synapse,
    SynapseAuthError,
    SynapseRateLimitError,
    SynapsePlanLimitError,
    SynapseValidationError,
    SynapseError,
)

with Synapse(api_key="gck_live_xxx", workspace_id="ws_xxx") as client:
    try:
        client.track(external_id="u1", event_name="purchase")
    except SynapseAuthError as e:
        print(f"Auth failed: {e.message}")          # 401 or 403
    except SynapseRateLimitError as e:
        print(f"Rate limited, retry after {e.retry_after}s")  # 429
    except SynapsePlanLimitError as e:
        print(f"Plan limit: {e.limit_type} ({e.current}/{e.maximum})")  # 403
    except SynapseValidationError as e:
        print(f"Validation: {e.errors}")             # 422
    except SynapseError as e:
        print(f"API error {e.status}: {e.message}")  # Other

Configuration

Parameter	Default	Description
api_key	(required)	Your API key (gck_live_... or gck_test_...)
workspace_id	(required)	Your workspace ID
base_url	https://synapse-api.pyrx.tech	API base URL
timeout	30.0	Request timeout in seconds
max_retries	3	Max retry attempts for retryable errors

Retryable status codes: 429, 500, 502, 503, 504. Backoff: exponential with jitter, capped at 30s. On 429, uses Retry-After header when present.

API Reference

See PYRX Synapse API docs.