rocketmq-sdk 0.1.1

Schema-aware AMQP client for the RocketMQ broker — Pydantic + aio-pika.

rocketmq-sdk

Schema-aware AMQP client for the RocketMQ broker — Pydantic v2 + aio-pika.

Define message schemas as Pydantic models, publish and consume with full type safety. The broker validates every message against a compiled protobuf schema at wire speed.

from rocketmq import connect
from rocketmq.schema import BaseSchema

class Order(BaseSchema):
    id: str
    customer_id: str
    qty: int

async def main():
    mq = await connect(url="amqp://localhost")
    orders = await mq.queue("orders", Order)

    await orders.send(Order(id="1", customer_id="c1", qty=5))
    await orders.consume(lambda msg: print(msg))

    await mq.close()

---

Installation

pip install rocketmq-sdk

Requires Python ≥ 3.11.

Quick Start

1. Connect

from rocketmq import connect

mq = await connect(url="amqp://guest:guest@localhost:5672")

2. Declare a queue with a schema

from rocketmq.schema import BaseSchema

class Notification(BaseSchema):
    id: int
    content: str
    timestamp: float

# Declares the queue + registers the proto3 schema on the broker
notifications = await mq.queue("notifications", Notification)

3. Publish

await notifications.send(Notification(
    id=1,
    content="Hello from Python",
    timestamp=1717520000.0,
))

4. Consume

async def on_notification(msg: Notification) -> None:
    print(f"Got: {msg.id} — {msg.content}")

await notifications.consume(on_notification)

5. Close

await mq.close()

---

API Reference

connect(url, serializer?)

Opens a connection and returns a RocketMQ client.

mq = await connect(url="amqp://localhost")
mq = await connect(url="amqp://localhost", serializer=MySerializer())

RocketMQ

Method	Description
await mq.queue(name, Schema)	Declare queue with schema → returns QueueHandle
await mq.assert_queue(name, Schema?)	Declare queue without returning a handle
await mq.send_to_queue(name, dict)	Publish a dict directly (untyped)
await mq.assert_exchange(name, type)	Declare an exchange
await mq.bind_queue(queue, exchange, routing_key)	Bind queue to exchange
await mq.publish(exchange, routing_key, dict)	Publish to exchange
await mq.consume(queue, Schema, handler)	Subscribe with typed handler
await mq.prefetch(count)	Set channel prefetch
await mq.close()	Close channel + connection

QueueHandle[T]

Typed wrapper returned by mq.queue(). All operations are bound to one queue + schema.

orders = await mq.queue("orders", Order)

# Typed publish — payload must be an Order instance
await orders.send(Order(id="1", customer_id="c1", qty=5))

# Typed consume — handler receives Order, not raw bytes
await orders.consume(lambda msg: print(msg.id))

---

Schemas

Defining a schema

Every message schema is a frozen Pydantic model:

from rocketmq.schema import BaseSchema

class OrderEvent(BaseSchema):
    order_id: str
    action: str
    amount: float

The class name (OrderEvent) becomes the protobuf message name. Fields are automatically mapped to proto3 types.

Type mapping

Python	Proto3	Notes
str	string
int	int32	Default for integers
float	double
bool	bool
bytes	bytes
datetime	google.protobuf.Timestamp	Adds import
list[str]	repeated string	Any inner type
int | None	optional int32	Optional fields

Overriding proto types

Use Proto() annotation for cross-language compatibility:

from typing import Annotated
from rocketmq.schema import BaseSchema, Proto

class Metric(BaseSchema):
    sensor_id: str
    value: Annotated[float, Proto("float")]    # float instead of double
    count: Annotated[int, Proto("int64")]       # int64 instead of int32
    big_id: Annotated[int, Proto("uint64")]     # unsigned

Available proto types: double, float, int32, int64, uint32, uint64, sint32, sint64, fixed32, fixed64, sfixed32, sfixed64, bool, string, bytes.

Schema validation

The broker validates every published message against the queue's compiled schema. If a field is missing or has the wrong type, the broker rejects the message:

from rocketmq.errors import SchemaValidationError

try:
    await mq.send_to_queue("orders", {"wrong_field": 123})
except SchemaValidationError as err:
    print(err.code)    # "SchemaTypeMismatch"
    print(err.queue)   # "orders"
    print(err.fields)  # [FieldErrorDetail(name="id", expected="str", got="missing")]

Consumer schema verification

When consuming, the SDK sends the consumer's schema to the broker. The broker verifies compatibility before delivering any message:

# Fails at subscribe time if NewOrder is incompatible with the queue's schema
await mq.consume("orders", NewOrder, handler)

# Override the queue's schema with the consumer's schema
await mq.consume("orders", NewOrder, handler, schema_override=True)

# Remove the queue's schema entirely
await mq.consume("orders", NewOrder, handler, schema_delete=True)

---

Exchange Routing

For topic/direct/fanout/headers routing:

# Declare exchange + queues
await mq.assert_exchange("events", "direct")
await mq.assert_queue("events.created", OrderEvent)
await mq.assert_queue("events.cancelled", OrderEvent)
await mq.bind_queue("events.created", "events", "created")
await mq.bind_queue("events.cancelled", "events", "cancelled")

# Publish to specific routing keys
await mq.publish("events", "created", {
    "order_id": "ord-001",
    "action": "created",
    "amount": 99.90,
})

# Consume from specific queues
await mq.consume("events.created", OrderEvent, lambda msg: print(f"NEW: {msg}"))
await mq.consume("events.cancelled", OrderEvent, lambda msg: print(f"CANCEL: {msg}"))

---

Custom Serializer

Swap JSON for any encoding (msgpack, protobuf, avro, etc.):

import msgpack
from typing import cast

class MsgpackSerializer:
    @property
    def content_type(self) -> str:
        return "application/x-msgpack"

    def serialize(self, value: object) -> bytes:
        return cast("bytes", msgpack.packb(value, use_bin_type=True))

    def deserialize(self, data: bytes) -> object:
        return msgpack.unpackb(data, raw=False)

mq = await connect(url="amqp://localhost", serializer=MsgpackSerializer())

No base class needed — any object with content_type, serialize(), and deserialize() works.

---

Error Handling

All errors extend RocketMQError:

RocketMQError
├── ConnectionError_       # Connection failures
├── QueueError             # Queue declaration failures
├── PublishError           # Publish failures (.queue, .payload)
├── ConsumeError           # Subscribe failures
├── SerializationError     # Encode/decode failures (.payload)
├── SchemaError            # Schema compilation issues
│   └── SchemaValidationError  # Type mismatch (.code, .queue, .fields)
└── TimeoutError_          # Operation timeouts

from rocketmq.errors import PublishError, SchemaValidationError

try:
    await mq.send_to_queue("orders", payload)
except SchemaValidationError as err:
    # Structured: err.code, err.queue, err.fields
    for field in err.fields:
        print(f"  {field.name}: expected {field.expected}, got {field.got}")
except PublishError as err:
    # Generic: err.queue, err.payload
    print(f"Failed on {err.queue}")

---

Development

uv sync          # install deps
make fmt         # format (ruff)
make lint        # lint (ruff)
make typecheck   # type-check (pyright)
make test        # run tests (pytest)
make check       # lint + typecheck + test
make all         # format + fix + typecheck + test

Architecture

src/rocketmq/
├── __init__.py          # Public re-exports
├── py.typed             # PEP 561 marker
├── amqp.py              # Thin aio-pika wrapper
├── client.py            # RocketMQ + connect() + QueueHandle
├── schema.py            # BaseSchema + Proto annotation
├── proto.py             # Pydantic model → proto3 generator
├── serializer.py        # Serializer Protocol + JsonSerializer
├── errors.py            # Error hierarchy
├── error_codes.py       # BrokerErrorCode enum
└── error_parser.py      # Broker JSON error parsing

License

MIT