Function example

examples/quorum-queues/myapp.py:71–145

()

def example():
    queue = my_quorum_queue.name if my_quorum_queue in (app.conf.task_queues or {}) else "celery"

    while True:
        print("Celery Quorum Queue Example")
        print("===========================")
        print("1. Send a simple identity task")
        print("1.1 Send an ETA identity task")
        print("2. Send a group of add tasks")
        print("3. Inspect the active queues")
        print("4. Shutdown Celery worker")
        print("Q. Quit")
        print("Q! Exit")
        choice = input("Enter your choice (1-4 or Q): ")

        if choice == "1" or choice == "1.1":
            queue_type = "Quorum" if is_using_quorum_queues(app) else "Classic"
            payload = f"Hello, {queue_type} Queue!"
            eta = datetime.now(UTC) + timedelta(seconds=30)
            if choice == "1.1":
                result = identity.si(payload).apply_async(queue=queue, eta=eta)
            else:
                result = identity.si(payload).apply_async(queue=queue)
            print()
            print(f"Task sent with ID: {result.id}")
            print("Task type: identity")

            if choice == "1.1":
                print(f"ETA: {eta}")

            print(f"Payload: {payload}")

        elif choice == "2":
            tasks = [
                (1, 2),
                (3, 4),
                (5, 6),
            ]
            result = group(
                add.s(*tasks[0]),
                add.s(*tasks[1]),
                add.s(*tasks[2]),
            ).apply_async(queue=queue)
            print()
            print("Group of tasks sent.")
            print(f"Group result ID: {result.id}")
            for i, task_args in enumerate(tasks, 1):
                print(f"Task {i} type: add")
                print(f"Payload: {task_args}")

        elif choice == "3":
            active_queues = app.control.inspect().active_queues()
            print()
            print("Active queues:")
            for worker, queues in active_queues.items():
                print(f"Worker: {worker}")
                for q in queues:
                    print(f"  - {q['name']}")