How to publish from a fastapi process?

[albertalexandrov]

Hi all!

I can't figure out how to publish from a fastapi process. We run a fastapi app with uvicorn and and faststream with taskiq scheduler app.tasks:scheduler commands.

Example of faststream with two types of task (scheduled and simple async task) (ran with taskiq scheduler command):

from faststream import FastStream
from faststream.redis import RedisBroker
from taskiq_faststream import BrokerWrapper
from taskiq_faststream import StreamScheduler
from taskiq.schedule_sources import LabelScheduleSource

broker = RedisBroker("redis://localhost:6379")
app = FastStream(broker)

@broker.subscriber("send-confirm-email")
async def send_confirm_email(email: str) -> None:
    # sending confirmation email...


@broker.subscriber("in-channel")
async def handle_msg(user: str, user_id: int) -> str:
    return f"User: {user_id} - {user} registered"

taskiq_broker = BrokerWrapper(broker)

taskiq_broker.task(
    message={"user": "John", "user_id": 1},
    channel="in-channel",
    schedule=[{
        "cron": "* * * * *",
    }],
)

scheduler = StreamScheduler(
    broker=taskiq_broker,
    sources=[LabelScheduleSource(taskiq_broker)],
)

Example of fastapi app (ran with uvicorn):

from fastapi import FastAPI

app = FastAPI()


@app.get('/register')
async def register(data):
     # doing something
     await broker.publish(data.email, channel="send-confirm-email")  # does not work, gives an error Exception <class 'faststream.exceptions.IncorrectState'>: Connection is not available yet. Please, connect the broker first

register view gives an error.

How to do it right?

[darekwojciechowski]

You don't need taskiq for this. The broker works on its own, so open it in the FastAPI lifespan and publish straight from the endpoint:

from contextlib import asynccontextmanager

from fastapi import FastAPI
from faststream.redis import RedisBroker

broker = RedisBroker("redis://localhost:6379")


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with broker:
        yield


app = FastAPI(lifespan=lifespan)


@app.get("/register")
async def register():
    await broker.publish("john@example.com", channel="send-confirm-email")
    return {"ok": True}

async with broker only connects and disconnects. There's no broker.start() here on purpose. That's the call that launches subscribers, and yours already run in the other process.

Keep the broker at module level like above. Create it inside lifespan and the route can't reach it.

Nothing changes on the FastStream side, send_confirm_email picks the message up as usual.

Skip faststream.redis.fastapi.RedisRouter for this, by the way. It's deprecated and moving to a separate faststream_fastapi package before 1.0, and you don't need it if all you do from FastAPI is publish.