How to use Schedule with Fastapi

[vunhatchuong]

I'm new to this architecture and read multiple examples but I still have some questions.

Basically you need to run 3 things right?

• The broker which runs with fastapi through uvicorn main:app.
• The scheduler
• Workers

tkb.py

import taskiq_fastapi
from taskiq import TaskiqScheduler, ZeroMQBroker
from taskiq.schedule_sources import LabelScheduleSource

broker = ZeroMQBroker()


scheduler = TaskiqScheduler(
    broker=broker,
    sources=[LabelScheduleSource(broker)],
)

taskiq_fastapi.init(broker, "main:app")


@broker.task(schedule=[{"cron": "*/1 * * * *"}])
async def add_one() -> None:
    print("Hello")

main.py

@asynccontextmanager
async def lifespan(app: FastAPI):
    if not broker.is_worker_process:
        print("Starting worker")
        await broker.startup()
    yield
    if not broker.is_worker_process:
        print("Shutdown worker")
        await broker.shutdown()

app = FastAPI(lifespan=lifespan)

---

1. Run fastapi

uvicorn main:app --port 8080

What does this do? From my understanding, this starts FastAPI with the broker. But does it start the scheduler?
Because when I

2. Run taskiq scheduler schedule.taskiq_scheduler:scheduler it err out:

[2024-11-06 12:52:29,946][INFO   ][run:run_scheduler:206] Starting scheduler.
Traceback (most recent call last):
...
zmq.error.ZMQError: Address already in use (addr='tcp://0.0.0.0:5555')

---

3. We also need to run a worker right?

$ taskiq worker schedule.taskiq_scheduler:broker

This returns a small part of my fastapi main.py which is weird to me.

[2024-11-06 12:46:24,926][taskiq.worker][INFO   ][MainProcess] Starting 2 worker processes.
[2024-11-06 12:46:24,929][taskiq.process-manager][INFO   ][MainProcess] Started process worker-0 with pid 36479
[2024-11-06 12:46:24,930][taskiq.process-manager][INFO   ][MainProcess] Started process worker-1 with pid 36481
INFO:     [06-11-2024 12:46:25]  main.py <module>(81): Not using GZIP
INFO:     [06-11-2024 12:46:25]  main.py <module>(81): Not using GZIP
INFO:     [06-11-2024 12:46:25]  receiver.py listen(330): Listening started.
INFO:     [06-11-2024 12:46:25]  receiver.py listen(330): Listening started.