StreamHouse Python SDK

Python client for StreamHouse — S3-native event streaming.

Install

pip install streamhouse

Quick Start

from streamhouse import StreamHouse

sh = StreamHouse("http://localhost:8080", api_key="sk_...")

# Create a topic
sh.admin.create_topic("events", partitions=3)

# Produce
producer = sh.producer()
producer.send("events", key="user-1", value={"action": "click"})

# Consume
consumer = sh.consumer()
for record in consumer.stream("events", partition=0):
    print(f"{record.key_str}: {record.value_str}")

# Schema Registry
registry = sh.schema_registry()
schema_id = registry.register("events-value", '{"type": "object"}')

sh.close()

API

StreamHouse(url, api_key?, organization_id?, timeout?)

Main client. Provides:

• sh.admin — topic, consumer group, agent, and metrics management
• sh.producer() — send records
• sh.consumer() — poll/stream records
• sh.schema_registry() — register and retrieve schemas

Producer

producer = sh.producer()
result = producer.send("topic", value="hello", key="k1", partition=0)
# result.offset, result.partition

Consumer

consumer = sh.consumer()
records = consumer.poll("topic", partition=0, offset=0, max_records=100)
# or stream continuously:
for record in consumer.stream("topic", partition=0):
    process(record)

Admin

sh.admin.create_topic("name", partitions=3)
sh.admin.list_topics()
sh.admin.delete_topic("name")
sh.admin.list_consumer_groups()
sh.admin.get_consumer_lag("group-id")
sh.admin.list_agents()
sh.admin.get_metrics()

Schema Registry

registry = sh.schema_registry()
schema_id = registry.register("subject", schema_str, format=SchemaFormat.AVRO)
schema = registry.get_by_id(schema_id)
subjects = registry.list_subjects()