Function is_using_quorum_queues

examples/quorum-queues/myapp.py:51–58

(app)

def is_using_quorum_queues(app) -> bool:
    queues = app.amqp.queues
    for qname in queues:
        qarguments = queues[qname].queue_arguments or {}
        if qarguments.get("x-queue-type") == "quorum":
            return True

    return False


@app.task