Python (Beta)

pip install sailhouse

from sailhouse import SailhouseClient

import os
from sailhouse import SailhouseClient

client = SailhouseClient(
    token=os.getenv("SAILHOUSE_TOKEN")
)

import asyncio
from datetime import datetime

async def main():
    data = {
        "text": "example contents",
        "object": {
            "key": "value"
        }
    }

    # Basic publish
    await client.publish("some-topic", data)

    # Publishing with optional parameters
    await client.publish(
        "some-topic",
        data,
        scheduled_time=datetime(2025, 12, 31),
        metadata={"source": "example"}
    )

asyncio.run(main())

from dataclasses import dataclass
import asyncio

@dataclass
class Message:
    text: str

async def main():
    event = await client.pull(
        "user-created",
        "send-welcome-email"
    )

    message = event.as_type(Message)
    print(message.text)

    await event.ack()

asyncio.run(main())

import asyncio
from sailhouse import Event

async def handle_event(event: Event):
    print(f"Received event: {event.data}")
    await event.ack()

async def handle_error(error: Exception):
    print(f"Error occurred: {error}")

async def main():
    await client.subscribe(
        "awesome-example",
        "awesome-subscription",
        handler=handle_event,
        polling_interval=5.0,
        on_error=handle_error,
        exit_on_error=False
    )

asyncio.run(main())

import asyncio
from sailhouse import SailhouseClient, Filter, ComplexFilter, FilterCondition

async def main():
    client = SailhouseClient(token=os.getenv("SAILHOUSE_TOKEN"))
    
    # Basic push subscription
    await client.admin.register_push_subscription(
        topic="user-events",
        subscription="welcome-email", 
        endpoint="https://api.example.com/webhooks/welcome"
    )
    
    # With simple filter
    await client.admin.register_push_subscription(
        topic="user-events",
        subscription="premium-users",
        endpoint="https://api.example.com/webhooks/premium",
        filter=Filter(
            path="user.type",
            value="premium"
        )
    )
    
    # With advanced filter
    await client.admin.register_push_subscription(
        topic="user-events",
        subscription="active-premium-users",
        endpoint="https://api.example.com/webhooks/active-premium",
        filter=ComplexFilter(
            filters=[
                FilterCondition(
                    path="user.type",
                    condition="equals",
                    value="premium"
                ),
                FilterCondition(
                    path="user.status", 
                    condition="not_equals",
                    value="inactive"
                )
            ],
            operator="and"
        ),
        rate_limit="100/h",
        deduplication="5m"
    )

asyncio.run(main())