---

PlanQ is a transport-agnostic async task queue for Python. Define a task once and run it over an in-memory, Redis, or SQS broker — the wire format is JSON-RPC 2.0, so producers and consumers stay decoupled from the transport.

Installation

pip install planq            # in-memory only
pip install "planq[redis]"   # Redis broker
pip install "planq[sqs]"     # AWS SQS broker

Example

from planq import Planq
from planq.providers.memory import InMemoryBroker

app = Planq(broker=InMemoryBroker())

@app.task()
async def greet(name: str, say: str = "hi") -> None:
    print(f"{say}, {name}!")

await greet.send("world")   # enqueue

License

Apache-2.0