How to Add Pub/Sub to a Flask App

Flask routes often kick off work that should not hold up the response, such as welcome emails, CRM syncs or webhooks to partners. Moving it out brings a broker, a worker process, retries and the duplicate deliveries that come with them.

This guide adds Pub/Sub to a Flask signup route with Encore, RQ, Celery and Flask's Blinker signals. Each version publishes a user-signed-up event to a welcome email handler.

What you need to decide:

• Whether one event has one consumer or several.
• What runs the handlers, in which process.
• How a handler behaves when a message arrives twice.
• How you run the broker locally.

Pub/Sub options for Flask at a glance

Encore: Pub/Sub for your existing Flask app

Encore combines topics and subscriptions in your Flask code with the infrastructure behind them, delivering messages locally under encore run and running on SNS and SQS on AWS or Pub/Sub on GCP in your own account after deploy. It is an infrastructure SDK added to an existing app, and the routes and blueprints stay as they are.

The event is a Pydantic model, and the topic is typed with it:

from encore import pubsub
from pydantic import BaseModel


class UserSignedUp(BaseModel):
    user_id: str
    email: str


# SNS and SQS on AWS, Pub/Sub on GCP, an in-process emulator under `encore run`.
signups = pubsub.Topic[UserSignedUp]("user-signed-up")

The subscriber skips users it already handled, because delivery is at least once. The set stands in for a table with a unique key, the idempotent receiver pattern:

welcomed: set[str] = set()  # Stand-in for a table with a unique key.


async def send_welcome(msg: pubsub.Message[UserSignedUp]) -> None:
    # Delivery is at-least-once, so skip users that were already handled.
    if msg.data.user_id in welcomed:
        return
    welcomed.add(msg.data.user_id)
    print(f"welcome email to {msg.data.email} (attempt {msg.delivery_attempt})")


welcome = pubsub.Subscription[UserSignedUp](
    signups, "send-welcome", handler=send_welcome, max_retries=5
)

The route is an async Flask view that publishes and returns 202:

from flask import Flask, request

flask_app = Flask(__name__)


@flask_app.post("/signup")
async def signup():
    data = request.get_json()
    msg_id = await signups.publish(UserSignedUp(user_id=data["user_id"], email=data["email"]))
    return {"queued": msg_id}, 202

Subscriptions need a running event loop, so Flask is served under Uvicorn through asgiref's WsgiToAsgi, and a lifespan hook calls encore.start():

import os

import encore
import uvicorn
from asgiref.wsgi import WsgiToAsgi

wsgi = WsgiToAsgi(flask_app)


async def app(scope, receive, send):
    # The lifespan hook starts Encore's subscription handlers on Uvicorn's loop.
    if scope["type"] == "lifespan":
        while True:
            message = await receive()
            if message["type"] == "lifespan.startup":
                await encore.start()
                await send({"type": "lifespan.startup.complete"})
            elif message["type"] == "lifespan.shutdown":
                await send({"type": "lifespan.shutdown.complete"})
                return
    else:
        await wsgi(scope, receive, send)


if __name__ == "__main__":
    host, _, port = os.environ.get("ENCORE_LISTEN_ADDR", "0.0.0.0:8000").rpartition(":")
    uvicorn.run(app, host=host or "0.0.0.0", port=int(port))

Under encore run, a signup is delivered to the handler, and posting the same user again logs nothing new:

On deploy, Encore creates an SNS topic and an SQS queue per subscription on AWS, or a Pub/Sub topic and subscription on GCP, in your own account, and provisions an IAM role for the service. Local traces follow each signup from the route into the handler.

An agent adding a second consumer, such as a CRM sync, writes one more Subscription on the same topic and checks the delivery in the local trace.

Good to know

Publishing after the database write means a crash between the two can drop the event. When that matters, write the event to an outbox table in the same transaction and publish from there. Failed handlers are retried with backoff up to max_retries, so raise an exception for errors worth retrying.

Try Encore →

RQ

RQ keeps jobs in Redis and runs them in rq worker processes:

from flask import Flask, request
from redis import Redis
from rq import Queue, Retry

app = Flask(__name__)
queue = Queue(connection=Redis.from_url("redis://localhost:6379"))


def send_welcome(user_id: str, email: str) -> None:
    print(f"welcome email to {email}")


@app.post("/signup")
def signup():
    data = request.get_json()
    queue.enqueue(send_welcome, data["user_id"], data["email"], retry=Retry(max=5))
    return {"queued": True}, 202

Good to know

Each consumer is its own job, so a second reaction to a signup means a second enqueue in the route. You run Redis and workers in every environment, and the worker imports the job function by path, so it needs the same code.

Celery

Celery is the common choice in larger Flask apps, with Redis or RabbitMQ as the broker:

from celery import Celery

celery = Celery("accounts", broker="redis://localhost:6379/0")


@celery.task(acks_late=True, autoretry_for=(Exception,), retry_backoff=True, max_retries=5)
def send_welcome(user_id: str, email: str) -> None:
    print(f"welcome email to {email}")

Good to know

Routes call send_welcome.delay(...), and a celery worker runs next to the app. Flask's application context is not present in the worker by default, so tasks that use extensions need the app factory pattern from the Flask docs.

Blinker signals

Flask ships with Blinker for in-process signals:

from blinker import signal

user_signed_up = signal("user-signed-up")


@user_signed_up.connect
def send_welcome(sender, **data) -> None:
    print(f"welcome email to {data['email']}")


# In the route: user_signed_up.send(None, user_id="u1", email="ada@example.com")

Good to know

Receivers run synchronously in the request, so a slow email provider slows the signup, and an exception fails it. Nothing is stored, so a crash loses the event.

How to choose

• Encore if you want topics with several subscribers that run locally and deploy to SNS and SQS or Pub/Sub in your own account from the same code.
• RQ for simple jobs in an app that already runs Redis.
• Celery if the team already operates it and each event has one consumer.
• Blinker signals for in-process hooks that can fail with the request.

Python support in Encore is in beta, and you can sign up for access to try it with your Flask app.

Related guides

• How to add Pub/Sub to a FastAPI app
• How to add Pub/Sub to a Django app
• How to add cron jobs to a Flask app
• Celery alternatives for Python background jobs
• Message queues vs Pub/Sub