rocketmq-sdk 0.1.0

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

rocketmq.py

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

Python port of the rocketmq.js TypeScript SDK.

Setup

Requires uv.

uv sync          # install deps + editable package

Quick Start

import asyncio
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 asyncio.sleep(2)
    await mq.close()


asyncio.run(main())

Development

make help        # show all commands
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 string
├── serializer.py        # Serializer Protocol + JsonSerializer
├── errors.py            # Error hierarchy
├── error_codes.py       # BrokerErrorCode enum
└── error_parser.py      # Broker JSON error parsing

Cross-Language Type Compatibility

Python types map to proto3 types that match the JS SDK:

Python	TypeScript (Zod)	Proto3
str	z.string()	string
int	z.number()	double
float	z.number()	double
bool	z.boolean()	bool
bytes	z.object(...)	bytes
datetime	z.date()	Timestamp
Annotated[int, Proto("int32")]	z.number().int()	int32