Python SDK for Whitewater Riverbeds. Implements the versioned

whitewater-protocol wire contract defined in the main repository under

sdk/spec/ (PROTOCOL.md, endpoints.json, conformance/).

Requires Python 3.10+. Runtime dependency: httpx.

pip install whitewater-client   # once published; pip install -e . for development

A complete microservice walkthrough — provision, produce, pump-process, dead-letter, shutdown — lives in docs/demo-service.md.

from whitewater import WhitewaterClient



client = WhitewaterClient(

    api_key=os.environ["WHITEWATER_API_KEY"],

    endpoints=[

        "https://node1:7070",

        "https://node2:7070",

        "https://node3:7070",

    ],

)

| Client | Covers |

|---|---|

| client (direct) | WCL + typed command execution (/v1/admin/wcl, /v1/admin/commands), node health — plus typed resource methods (create_domain, create_feed, create_writer, create_reader, create_subscription, describe_feed, rename_feed, drop_feed, list_feeds, seek_reader, open_writer_session, …) that build the same WCL for you. ReaderStart (beginning(), now(), cursor(c), timestamp_ns(ns)) supplies START AT positions |

| client.domain(name) / client.get_domain(name) / client.create_domain(name) | The Domain is the front door — create_domain() returns the handle too. domain() is lazy (free name composition, no I/O); get_domain() verifies the Domain exists and materializes its catalog view in one batched read — feeds("order"), feed_names(), readers(), writers(), subscriptions(), pipes(), indexes(), state_stores(), describe(), history_policy, refresh(). Every resource resolves inside the Domain: domain.feed("orders") binds orders under the Domain's prefix; domain.create_feed("orders") declares it; create_writer/create_reader/create_subscription/pipe/state_store/define_*_pipe/query_index/drop_*/rename_*/seek_reader/open_writer_session all follow the same rule |

| domain.writer(name, epoch) | Idempotent single and batch appends (request_id dedup) |

| domain.reader(name) | Named-Reader sessions: open / fetch / ack / close / resume |

| domain.member_client(sub) | Join, claim, renew, fetch, ack, release under fenced leases |

| domain.feed(name[, value_type/value_codec]) | Typed feed facade: binds codecs once so keys/payloads serialize automatically in both directions — feed.writer(name) (declares + opens a session), feed.reader(name, max_page_size) (decoded fetch/ack, context-managed), feed.fetch/records/load (decoded ad-hoc reads), feed.subscribe (decoded batches), feed.consumer(name) (supervised loop). Same codec vocabulary as state_store: str/bytes/JSON dataclasses out of the box, protobuf/msgpack/custom via codecs.of |

See docs/stores.md for the full store guide.

A Domain owns every Feed it prefixes (shop owns shop.orders), so the friendliest surface navigates domain → feed → reader/writer instead of composing dotted names by hand:

from whitewater import ReaderStart

shop = client.get_domain("shop")          # verifies + materializes now
# client.domain("shop") is the lazy form — zero I/O until needed
shop.create_feed("orders")              # CREATE FEED shop.orders;

shop.feed_names()                         # ['shop.orders', 'shop.returns']
[f.name for f in shop.feeds("order")]     # prefix-filtered Feed handles
shop.readers(); shop.writers()            # scoped catalog views
shop.describe(); shop.history_policy

orders = shop.feed("orders", value_type=Order)   # typed Feed handle

# Everything a service needs hangs off the Domain — relative names
# resolve inside it:
writer = (shop.create_writer("orders")
          .writer_id("orders.ingest")
          .open())
writer.append("order-1", Order("order-1", 42))

with (shop.create_reader("orders")
      .reader_id("fulfillment")                    # durable id — the Kafka groupId analogue
      .reader_start_point(ReaderStart.BEGINNING)   # first-declare only
      .max_page_size(256)
      .open()) as reader:
    pump = reader.listen(lambda m: handle(m.value))  # SDK-owned thread — returns now
    # pump.stop() releases thread + session; pump.failure surfaces a raised handler
    # or iterate: `for m in reader:` / `reader.for_each(h)` — same ack-per-page semantics
    # or stay manual: fetch() → handle records → ack()

reader_id is durable application state: reader_start_point applies only when the Reader is first declared — an existing Reader resumes its acknowledged frontier (the auto.offset.reset mapping). A Feed bound to another Domain is refused locally, before any network call. The first catalog accessor issues ONE batched control read (describe domain + every show listing) and caches the snapshot — refresh() re-pulls.

Records stay schema-neutral bytes on the wire; a Feed can still declare schema metadata the catalog tracks for tooling, validation, and compatibility. The Domain's schema is the default every Feed inherits; a Feed's own declaration overrides the named sides:

from whitewater import FeedSchema, SchemaDecl

shop = client.domain("shop")
client.set_domain_schema("shop", FeedSchema(value=SchemaDecl("json", subject="order-v1")))
shop.set_feed_schema("orders", FeedSchema(key=SchemaDecl("text")))

shop.schema()                            # the Domain default
shop.feed_schema("orders")               # effective: key=text, value=json/order-v1
shop.declared_feed_schema("orders")      # the Feed's own sides only
shop.clear_feed_schema("orders")         # reverts to inheritance

SET SCHEMA ON DOMAIN|FEED <name> (KEY|VALUE <format> [SUBJECT 'x'] [DEFINITION 'y'])+ in WCL does the same; SHOW SCHEMAS lists declared assignments. Formats today: json, jsonschema, avro, protobuf, text, bytes, or custom.<name> — compatibility enforcement is still a no-op.

feed.consumer(name) binds the durable named Reader (the consumer-group equivalent) and owns the fetch → deliver → ack cycle: transient failures retry with backoff, a fenced session re-opens, progress resumes from the server-side cursor. Your code is a loop over records — nothing else:

for msg in in_feed.consumer("enricher"):
    process(msg)          # decoded value; page acks after each yield

# or a handler, blocking forever until close():
consumer = in_feed.consumer("enricher").on_record(process)
consumer.run()

• Delivery is per-record; the page's delivered cursor acks only after every handler returns — a raised exception leaves it unacked, so the page redelivers (at-least-once: make effects idempotent via request_id, and route poison pills to a DLQ yourself).
• run() raises only for startup refusals (bad names, denied auth) — everything after the first open is supervised forever until consumer.close() (safe from signals/atexit; the in-flight page finishes and acks first).
• run_async(handler) is the same supervised consumer on its own thread; .backoff(initial_s, max_s), .max_page_size(n), .start_at(...), .on_error(fn), .last_error, .is_running tune and observe it.
• feed.subscribe(...) remains for page-oriented loops that want raw control.

Value-only store lookups skip the row wrapper when freshness evidence isn't needed:

user  = users.find(email)                    # value or None
user  = users.find_by("user_id", user_id)    # first named-key match
count = counters.get_or_default(bucket, 0)

get/get_by still return StoreRow/StoreRows with applied-cursor freshness evidence. A complete enrichment service — durable consumer, named-key fallback, DLQ, idempotent outputs, counters — is examples/enrich_users.py. | domain.feed_ops() | Raw-byte ad-hoc Feed operations: committed record reads, anonymous temporary fetches (tail/wait_ms/new_only), ad-hoc appends, load(feed, start, sink) -- the escape hatch under domain.feed(...) |

| domain.subscribe(reader, feed, start, on_records) | Managed durable consume: provisions the Reader, runs open -> fetch -> deliver -> ack on a background thread; sub.close() (or with) to stop |

| domain.state_store(name[, value_type/value_codec]) | Typed facade over the node-side StateStore: put/get/get_by/delete/iterate/clear/drop with typed named keys and freshness evidence. value_type= maps a class/dataclass to JSON; value_codec= binds any codecs.Codec · domain.define_state_store(spec[, source]) declares it |

accounts = client.domain("accounts")

store = (accounts.define_state_store(

        accounts.state_store("employees")

        .add_unique_key("passport").add_non_unique_key("department")))  # up to 10 named keys



# named-key postings: a typed StoreKeySet subclass, a builder lambda,
# a plain dict, or a single string for one declared named key —
store.put("user-1", profile,
          PassportKeySet(passport="P111", department="eng"))    # typed values
store.put("user-1", profile,
          named={"passport": "P111", "department": "eng"})      # dict form

rows = store.get_by("department", "eng").rows                    # every match



store.get("emp-1")                       # by primary key

store.get_by("passport", "P123")         # by named key — every match

for row in store.iterate(): export(row)   # primary-key order, bounded pages

feed.load(start, sink) remains the lighter one-shot

read when you don't need named-key projections. Codecs are shared

across feeds and stores — bind once, never .encode()/.decode() in

application code:

orders = client.domain("orders").feed("events", value_type=Order)

orders.writer("ingest").append("order-1", Order("order-1", 42))

with orders.reader("audit", max_page_size=64) as reader:

    order = reader.fetch().records[0].value          # decoded Order



latest = client.domain("demo").state_store("latest", value_type=Order)

latest.put("order-1", order)                          # same codec seam

See docs/stores.md for the full store guide.

Errors surface as WhitewaterError subtypes mapped from wire code

values (stale_epoch, lease.busy, lease.stale_lease,

writer.stale_sequence, ...). Unavailable covers transient refusals;

AmbiguousResult marks operations that may have committed — retry

through the same request_id, never blindly re-issue.

Cursors are opaque strings: never parsed, ordered, or fabricated.

event_time_ns/ingest_time_ns are Python ints (arbitrary precision)

and travel as decimal strings on the wire — the full signed 64-bit

nanosecond range round-trips exactly.

Managed Pipes and replicated Indexes share the same topology-free

model — names, never partitions or Nodes. See

docs/pipes.md for the full pipe guide:

from datetime import timedelta

# Pipe declaration: the owning Domain names the business key, source and
# outputs — the server owns panes, leases and the effect journal.
orders = client.domain("orders")
crm = client.domain("crm")
stats = client.domain("stats")
orders.define_forward_pipe("relay", "orders.sub", "relayed")
orders.define_enrich_pipe("enrich", "orders.sub",
    "enriched", crm.state_store("customers"),   # cross-Domain store handle
    PipeOperation.enrich(Extractor.record_key(),
        freshness=EnrichFreshness.LATEST,
        on_miss=EnrichMiss.QUARANTINE),
    quarantine_feed="enrich.dlq")
orders.define_window_pipe("hourly", "orders.sub",
    stats.state_store("orders.hourly"),         # cross-Domain pane store
    PipeOperation.window(timedelta(hours=1),  # 1h panes
        Extractor.record_key(), WindowAggregation.count(),
        time=WindowTime.EVENT,
        watermark=timedelta(minutes=5),  # 5m horizon
        grace=timedelta(seconds=30),     # 30s late revisions
        expire=timedelta(hours=2)),
    emit_feed="hourly.emitted")

# Strings resolve inside the declaring Domain; a Feed/SubscriptionRef/
# StateStore handle or Scoped.absolute crosses Domains explicitly —
# see examples/cross_domain.py.

# Pipe lifecycle: describe, pause, resume, drop.

pipe = orders.pipe("relay")

info = pipe.describe()        # stage, operation, outputs, status

pipe.pause(); pipe.resume(); pipe.drop()

pipes = orders.pipe_infos()   # the Domain-scoped catalog



# Index queries: exact, prefix, or range predicates over typed values.

page = crm.query_index(

    "byregion",

    IndexPredicate.range(

        ["emea"],

        IndexBound.inclusive_bound(0),

        IndexBound.exclusive_bound(99)),

    limit=50, strict=True)

for row in page.rows:         # row.key / row.payload bytes

    ...



# Demand observability: typed rows + per-domain rollups.

report = client.demand_feeds()

appended = report.rows[0].counter("appended_records")

client.demand_readers(); client.demand_subscriptions()

client.demand_indexes();  client.demand_pipes()

hot = client.hot_keys()       # dominant-key evidence per feed

Pipe operation payloads and demand counter fields stay opaque dicts so

server-side variants evolve without an SDK bump; query_index typed

values lower automatically (str/int/bool/None/bytes).

Everything the SDK exposes, grouped by capability. The typed surface (Domain → Feed → writer/reader/consumer) covers the daily path; commands, wcl, and the *SessionClient/SubscriptionMemberClient classes are the explicit escape hatches for the same wire contract.

pip install -e .[dev]

pytest

whitewater.testing.FakeRiverbed serves the wire contract on loopback — appends dedupe by request id, unacked pages redeliver, stale epochs fence, and StateStores model named keys — so application tests exercise the real client without a Riverbed.

from whitewater.testing import FakeRiverbed

with FakeRiverbed() as fake:
    client = fake.client()          # configured WhitewaterClient
    # ...

See tests/test_testing_double.py and the testing guide.

tests/conformance/ executes the shared fixture corpus against a real

three-Node Riverbed:

$env:WHITEWATER_ENDPOINTS = "http://localhost:7071,http://localhost:7072,http://localhost:7073"

$env:WHITEWATER_SPEC_DIR  = "C:\Code\Stackapps\whitewater\sdk\spec\conformance"

pytest tests/conformance

A scenario whose op the SDK cannot express fails explicitly rather

than silently skipping — a spec-required signal, not a bug.