Anlyon Relay Python SDK

Official Python SDK for the Anlyon Relay API - Type-safe message scheduling and delivery.

Installation

pip install anlyon-relay

Quick Start

Synchronous Client

from anlyon_relay import Client, PublishMessageRequest

# Create a client
client = Client(api_key="your-api-key")

# Publish a message
result = client.messages.publish(
    PublishMessageRequest(
        url="https://example.com/webhook",
        body={"hello": "world"}
    )
)

print(f"Message ID: {result.data.message_id}")

# Close when done
client.close()

Asynchronous Client

import asyncio
from anlyon_relay import AsyncClient, PublishMessageRequest

async def main():
    async with AsyncClient(api_key="your-api-key") as client:
        result = await client.messages.publish(
            PublishMessageRequest(
                url="https://example.com/webhook",
                body={"hello": "world"}
            )
        )
        print(f"Message ID: {result.data.message_id}")

asyncio.run(main())

Features

• Type-safe: Full type hints and Pydantic models
• Sync & Async: Both synchronous and asynchronous clients
• Fluent Builders: Intuitive API for building requests
• Automatic Retries: Exponential backoff with jitter
• Webhook Verification: HMAC-SHA256 signature verification

Usage Examples

Fluent Message Builder

from anlyon_relay import Client

client = Client(api_key="your-api-key")

# Using the fluent builder
result = (
    client.messages.create_message()
    .to("https://example.com/webhook")
    .method("POST")
    .header("X-Custom-Header", "value")
    .body({"event": "user.created", "user_id": 123})
    .in_minutes(5)  # Schedule 5 minutes from now
    .max_retries(3)
    .send()
)

Scheduled Messages

from datetime import datetime
from anlyon_relay import Client, PublishMessageRequest

client = Client(api_key="your-api-key")

# Schedule for a specific time
result = client.messages.publish(
    PublishMessageRequest(
        url="https://example.com/webhook",
        scheduled_for="2024-12-25T00:00:00Z"
    )
)

# Or use Unix timestamp
result = client.messages.publish(
    PublishMessageRequest(
        url="https://example.com/webhook",
        execute_at=1735084800  # Unix timestamp in seconds
    )
)

Cron Schedules

from anlyon_relay import Client, CreateScheduleRequest

client = Client(api_key="your-api-key")

# Create a schedule
result = client.schedules.create(
    CreateScheduleRequest(
        name="Daily Report",
        cron_expression="0 9 * * *",  # Every day at 9 AM
        timezone="America/New_York",
        url="https://example.com/generate-report"
    )
)

# Using the fluent builder
result = (
    client.schedules.create_schedule()
    .name("Hourly Check")
    .hourly()
    .to("https://example.com/health-check")
    .create()
)

URL Groups (Broadcasting)

from anlyon_relay import Client, CreateUrlGroupRequest, EndpointInput

client = Client(api_key="your-api-key")

# Create a group for broadcasting
result = client.url_groups.create(
    CreateUrlGroupRequest(
        name="Notification Services",
        endpoints=[
            EndpointInput(url="https://service1.com/webhook"),
            EndpointInput(url="https://service2.com/webhook"),
            EndpointInput(url="https://service3.com/webhook"),
        ]
    )
)

# Publish to all endpoints
from anlyon_relay import PublishGroupMessageRequest

client.url_groups.publish(
    result.data.id,
    PublishGroupMessageRequest(body={"event": "broadcast"})
)

Queues

from anlyon_relay import Client, CreateQueueRequest

client = Client(api_key="your-api-key")

# Create a queue with parallelism
result = client.queues.create(
    CreateQueueRequest(
        name="email-queue",
        parallelism=5  # Process 5 messages at a time
    )
)

# Publish to the queue
from anlyon_relay import PublishMessageRequest

client.messages.publish(
    PublishMessageRequest(
        url="https://example.com/process-email",
        queue="email-queue"
    )
)

Pagination

from anlyon_relay import Client

client = Client(api_key="your-api-key")

# Iterate through all messages
for message in client.messages.list_all():
    print(f"Message: {message.message_id} - {message.status}")

# Or collect to a list
all_messages = client.messages.list_all().to_list()

# Take first 10
first_10 = client.messages.list_all().take(10)

# Find specific message
pending = client.messages.list_all().find(
    lambda m: m.status == "pending"
)

Async Pagination

import asyncio
from anlyon_relay import AsyncClient

async def main():
    async with AsyncClient(api_key="your-api-key") as client:
        async for message in client.messages.list_all():
            print(f"Message: {message.message_id}")

        # Or collect to list
        all_messages = await client.messages.list_all().to_list()

asyncio.run(main())

Webhook Verification

from anlyon_relay import Client, Receiver, ReceiverConfig

# Option 1: Using Client static method
result = Client.verify_webhook_signature(
    payload=request_body,
    signature_header=request.headers["x-anlyon-signature"],
    secret="your-webhook-secret"
)

if result.valid:
    # Process the webhook
    pass

# Option 2: Using Receiver class
receiver = Receiver(ReceiverConfig(signing_secret="your-webhook-secret"))

try:
    receiver.verify_or_throw(
        signature=request.headers["x-anlyon-signature"],
        body=request_body
    )
    # Signature is valid, process the webhook
except SignatureVerificationError as e:
    # Invalid signature
    pass

Error Handling

from anlyon_relay import (
    Client,
    AnlyonError,
    RateLimitError,
    ValidationError,
    AuthenticationError,
)

client = Client(api_key="your-api-key")

try:
    result = client.messages.publish(...)
except RateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after} seconds")
except ValidationError as e:
    print(f"Validation failed: {e.details}")
except AuthenticationError as e:
    print("Invalid API key")
except AnlyonError as e:
    print(f"API error: {e.code} - {e.message}")

Custom Configuration

from anlyon_relay import Client, RetryConfig, LoggingConfig, LogLevel

client = Client(
    api_key="your-api-key",
    base_url="https://api.queue.anlyon.com",
    timeout=60.0,
    retry=RetryConfig(
        max_attempts=5,
        initial_delay=0.5,
        max_delay=30.0,
        backoff_multiplier=2.0,
        jitter=True,
    ),
    logging=LoggingConfig(
        enabled=True,
        level=LogLevel.DEBUG,
        redact_sensitive=True,
    ),
    default_headers={"X-Custom-Header": "value"},
)

API Reference

Resources

• client.messages - Publish and manage messages
• client.schedules - Create and manage cron schedules
• client.url_groups - Manage URL groups for broadcasting
• client.queues - Create and manage message queues
• client.dlq - Dead letter queue operations
• client.analytics - Usage analytics and metrics
• client.notifications - Notification management
• client.billing - Billing and usage information

Types

All request and response types are available as Pydantic models with full type hints.

License

MIT