Metadata-Version: 2.4
Name: rocketmq-sdk
Version: 0.1.2
Summary: High-performance Python SDK for RocketMQ broker with Pydantic schema validation
License-Expression: MIT
License-File: LICENSE
Requires-Python: >=3.11
Requires-Dist: aio-pika<10.0,>=9.0
Requires-Dist: pydantic<3.0,>=2.0
Description-Content-Type: text/markdown

# rocketmq-sdk

Schema-aware AMQP client for the [RocketMQ broker](https://github.com/rocketmq-broker/rocketmq) — **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.

```python
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

```bash
pip install rocketmq-sdk
```

Requires Python ≥ 3.11.

## Quick Start

### 1. Connect

```python
from rocketmq import connect

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

### 2. Declare a queue with a schema

```python
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

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

### 4. Consume

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

await notifications.consume(on_notification)
```

### 5. Close

```python
await mq.close()
```

---

## API Reference

### `connect(url, serializer?)`

Opens a connection and returns a `RocketMQ` client.

```python
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.

```python
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:

```python
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:

```python
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:

```python
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:

```python
# 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:

```python
# 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 (e.g. `orjson` for performance, or `protobuf` binary):

```python
import orjson

class OrjsonSerializer:
    @property
    def content_type(self) -> str:
        return "application/json"

    def serialize(self, value: object) -> bytes:
        return orjson.dumps(value)

    def deserialize(self, data: bytes) -> object:
        return orjson.loads(data)

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

No base class needed — any object with `content_type`, `serialize()`, and `deserialize()` works.
Note: the broker strictly validates payloads. Only JSON or Protobuf encodings are currently supported for schema-enforced queues.

---

## 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
```

```python
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

```bash
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
