leanmq 0.2.0

Replace your internal webhooks with a reliable, performant, and scalable MQ with a fantastic DX.

LeanMQ

---

🔥 Just getting started – show your support!

If you find LeanMQ interesting, consider giving it a ⭐ star on GitHub!
It helps others discover it and motivates ongoing improvements.

---

LeanMQ is a lightweight, message queue library with minimal dependencies for internal or async microservice communication. It provides a simple but powerful implementation using Redis Streams with support for —

• Dead Letter Queues: Automatic handling of failed messages
• Message TTL: Set expiration times for messages
• Atomic Transactions: Send multiple messages in a single transaction
• Consumer Groups: Support for multiple consumers
• Message Tracking: Track delivery attempts and failures

Read the docs

---

Installation

pip install leanmq

Quick Start

# Set up a webhook server
from leanmq import LeanMQWebhook

webhook = LeanMQWebhook()

# Register a handler
@webhook.get("/order/status/")
def process_order_status(data):
    print(f"Order {data['order_id']} status: {data['status']}")

# Start processing webhooks in a background thread
service = webhook.run_service()

# In another service, send a webhook
webhook.send("/order/status/", {
    "order_id": "ORD-12345", 
    "status": "shipped"
})

Find rest of the docs here.

License

Apache 2.0 - see LICENSE.md for details.

Copyright

Copyright (c) 2025 Augustus D'Souza