feat: python sdk batch operations & smart batching

Merged
VI
7mo ago
+1218-57

Summary

  • Add batch_enqueue() method for explicit multi-message RPCs on both sync and async clients
  • Add smart batching via BatchMode enum (AUTO/DISABLED) and Linger dataclass, routing enqueue() through a background batcher thread by default
  • Add delivery batching: consume stream transparently unpacks ConsumeResponse.messages repeated field with backward-compatible fallback to singular message field
  • Single-item optimization: when flushing 1 message, use singular Enqueue RPC to preserve error types (QueueNotFoundError vs BatchEnqueueError)
  • close() drains pending batched messages before disconnecting
  • Update proto to include BatchEnqueue RPC and ConsumeResponse.messages repeated field
  • 16 new unit tests (batcher mechanics, flush functions, batch mode types) + 9 integration tests (batch enqueue, smart batching modes)

Test plan

  • Unit tests for _flush_single and _flush_batch with mock stubs
  • Unit tests for AutoBatcher (single message, concurrent batching, close drains, stub update)
  • Unit tests for LingerBatcher (batch size trigger, linger timeout, close drains)
  • Unit tests for BatchMode, Linger, BatchEnqueueResult types
  • Integration tests for explicit batch_enqueue() (multiple messages, single message, consume verification)
  • Integration tests for async batch_enqueue()
  • Integration tests for smart batching modes (AUTO, DISABLED, LINGER, default-is-AUTO)
  • Existing integration tests still pass (enqueue/consume/ack lifecycle, nack redelivery, API key auth)
  • ruff lint clean, mypy clean (no new errors)

Generated with Claude Code

Summary by cubic

Add explicit batch_enqueue() to the Python SDK and enable smart batching by default (AUTO) for enqueue(). Also add batched delivery in consume() and update the proto with BatchEnqueue and ConsumeResponse.messages.

  • New Features

    • Client.batch_enqueue() and AsyncClient.batch_enqueue() send many messages in one RPC; return per-message BatchEnqueueResult.
    • Smart batching via BatchMode (AUTO default, DISABLED) and Linger; background batcher with single-item optimization; close() drains pending; reconnect updates batcher stub.
    • Delivery batching: consume() transparently iterates ConsumeResponse.messages with fallback to singular message.
    • Proto changes: add BatchEnqueue RPC and ConsumeResponse.messages; new BatchEnqueueError; public exports updated.

  • Migration

    • enqueue() now routes through a background batcher by default; disable with batch_mode=BatchMode.DISABLED, or use Linger(linger_ms, batch_size) for timer-based batching.
    • Call close() to flush any pending batched messages before shutdown.
    • No changes needed for consumers; batched deliveries are unpacked automatically.