How to Add Pub/Sub to a FastAPI App
Publishing events from FastAPI and handling them in subscribers with Encore, FastAPI BackgroundTasks, Celery or SNS and SQS with boto3, with retries, idempotency and local testing.
FastAPI endpoints often start work that should not hold up the response, such as receipts, stock reservations or search updates. The usual answer is a broker with subscribers, which brings retries, duplicate deliveries and a broker to run on every laptop and in production.
This guide adds Pub/Sub to a FastAPI order endpoint with Encore, BackgroundTasks, Celery and SNS with SQS. Each version publishes an order-placed event to two subscribers.
What you need to decide:
- Whether the work can be lost on a restart, or must survive it.
- How many consumers react to one event.
- How handlers deal with the same message arriving twice.
- What runs the broker locally and in production.
Pub/Sub options for FastAPI at a glance
Encore: Pub/Sub for your existing FastAPI app
Encore combines typed topics and subscriptions in your FastAPI code with the broker behind them, running locally under encore run and as SNS and SQS on AWS or Pub/Sub on GCP after deploy. It is an infrastructure SDK you add to the app, and FastAPI keeps its routes and models.
The example stores receipts and stock reservations in Postgres, which gives each subscriber a place to record what it has already done. The migration uses the order ID as the primary key in both tables:
CREATE TABLE receipts (
order_id TEXT PRIMARY KEY,
email TEXT NOT NULL,
sent_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE stock_reservations (
order_id TEXT PRIMARY KEY,
sku TEXT NOT NULL,
quantity INTEGER NOT NULL
);
The topic carries a Pydantic model, and each subscription gets its own copy of every message. The receipt subscriber retries up to five times with a one second minimum backoff, and the insert with ON CONFLICT DO NOTHING makes a redelivery a no-op:
import os
from contextlib import asynccontextmanager
import encore
import uvicorn
from encore import pubsub
from encore.storage import postgres
from fastapi import FastAPI
from pydantic import BaseModel
db = postgres.Database("orders", migrations="./migrations")
class OrderPlaced(BaseModel):
order_id: str
email: str
sku: str
quantity: int
# One topic, any number of subscriptions. Each subscription gets its own
# copy of every message (SNS/SQS on AWS, Pub/Sub on GCP).
order_placed = pubsub.Topic[OrderPlaced]("order-placed")
async def send_receipt(msg: pubsub.Message[OrderPlaced]) -> None:
order = msg.data
# Delivery is at-least-once, so the handler must be safe to run twice.
# The primary key turns a redelivery into a no-op.
result = await db.exec(
"INSERT INTO receipts (order_id, email) VALUES ($1, $2) "
"ON CONFLICT (order_id) DO NOTHING",
order.order_id,
order.email,
)
if result.rows_affected == 0:
print(f"receipt for {order.order_id} already sent, skipping")
return
print(f"sending receipt for {order.order_id} to {order.email} (attempt {msg.delivery_attempt})")
async def reserve_stock(msg: pubsub.Message[OrderPlaced]) -> None:
order = msg.data
await db.exec(
"INSERT INTO stock_reservations (order_id, sku, quantity) VALUES ($1, $2, $3) "
"ON CONFLICT (order_id) DO NOTHING",
order.order_id,
order.sku,
order.quantity,
)
print(f"reserved {order.quantity} x {order.sku} for {order.order_id}")
receipts = pubsub.Subscription[OrderPlaced](
order_placed,
"send-receipt",
handler=send_receipt,
max_retries=5,
min_backoff="1s",
)
inventory = pubsub.Subscription[OrderPlaced](
order_placed,
"reserve-stock",
handler=reserve_stock,
)
The endpoint publishes and returns 202 straight away, and encore.start() in the lifespan starts the subscription handlers on the app's event loop:
@asynccontextmanager
async def lifespan(app: FastAPI):
# Starts the subscription handlers on the app's event loop.
await encore.start()
yield
app = FastAPI(lifespan=lifespan)
@app.post("/orders", status_code=202)
async def place_order(order: OrderPlaced) -> dict:
message_id = await order_placed.publish(order)
return {"order_id": order.order_id, "message_id": message_id}
@app.get("/orders/{order_id}/status")
async def order_status(order_id: str) -> dict:
receipt = await db.query(
"SELECT email FROM receipts WHERE order_id = $1", order_id
).first()
stock = await db.query(
"SELECT sku, quantity FROM stock_reservations WHERE order_id = $1", order_id
).first()
return {
"receipt_sent": receipt is not None,
"stock_reserved": dict(stock) if stock else None,
}
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))
encore run starts Postgres locally and delivers messages to both subscribers in the same process. Publishing the same order twice shows the idempotency at work:
A handler that raises is retried with backoff, and msg.delivery_attempt tells it which attempt it is on. Requests carry a trace ID in the local logs, and tracing shows where the time in each request went. On deploy, Encore creates an SNS topic with an SQS queue per subscription on AWS, or a Pub/Sub topic and subscriptions on GCP, with dead-letter handling and IAM permissions for each service, in your own account.
A coding agent that adds a third consumer, for example analytics, adds one Subscription and tests it against the local broker before anything reaches the cloud.
Good to know
The receipt row is written before the side effect here, which keeps the example short. With a real email provider, pass the order ID as the provider's idempotency key, or record the send after the provider accepts it, so a failure between the two steps cannot drop a receipt.
FastAPI BackgroundTasks
BackgroundTasks run a function after the response is sent, in the same process:
from fastapi import BackgroundTasks, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Order(BaseModel):
order_id: str
email: str
def send_receipt(order_id: str, email: str) -> None:
print(f"sending receipt for {order_id} to {email}")
@app.post("/orders", status_code=202)
async def place_order(order: Order, tasks: BackgroundTasks) -> dict:
tasks.add_task(send_receipt, order.order_id, order.email)
return {"order_id": order.order_id}
Good to know
There are no retries, no persistence and no fan-out. The FastAPI documentation itself points to bigger tools like Celery for heavy background computation, and a deploy or crash drops any task that has not run yet.
Celery
Celery is the long-standing Python task queue. One task per consumer approximates fan-out, and the endpoint enqueues each:
from celery import Celery
celery_app = Celery("orders", broker="redis://localhost:6379/0")
@celery_app.task(acks_late=True, autoretry_for=(Exception,), retry_backoff=True, max_retries=5)
def send_receipt(order_id: str, email: str) -> None:
print(f"sending receipt for {order_id} to {email}")
@celery_app.task(acks_late=True)
def reserve_stock(order_id: str, sku: str, quantity: int) -> None:
print(f"reserved {quantity} x {sku} for {order_id}")
In FastAPI, call send_receipt.delay(...) and reserve_stock.delay(...) from the endpoint, and run celery -A orders worker next to the API.
Good to know
By default Celery acknowledges a task when a worker receives it, so a crash mid-task loses it. The acks_late setting acknowledges after the task runs, which makes delivery at least once and the tasks need to be idempotent in the same way as Pub/Sub subscribers. The publisher has to know every consumer, so adding one means changing the endpoint.
SNS and SQS with boto3
Using AWS directly, the endpoint publishes to an SNS topic and each consumer polls its own SQS queue subscribed to it:
import json
import os
import boto3
sns = boto3.client("sns")
TOPIC_ARN = os.environ["ORDER_PLACED_TOPIC_ARN"]
def publish_order(order: dict) -> str:
response = sns.publish(TopicArn=TOPIC_ARN, Message=json.dumps(order))
return response["MessageId"]
Good to know
boto3 is synchronous, so in an async endpoint run it in a thread pool or use aioboto3. Each consumer needs a polling loop that deletes messages only after handling them, a visibility timeout longer than the handler takes, a dead-letter queue and the queue policy that lets SNS deliver to it. Standard queues deliver at least once, so the same idempotency applies. Locally this means LocalStack or a shared development account.
How to choose
- Encore if you want typed topics with several subscribers that run locally and deploy to SNS and SQS or Pub/Sub in your account.
- BackgroundTasks for best-effort work that can be lost on a restart.
- Celery if you already run it and each event has one consumer.
- SNS and SQS with boto3 if you want to own every AWS resource and write the consumers yourself.
Python support in Encore is in beta, and you can sign up for access to try it with your FastAPI app.
Related guides
Frequently asked questions
How do I add Pub/Sub to FastAPI?
Define a topic with a message schema, publish to it from the endpoint, and handle messages in one or more subscribers that are safe to run twice. With Encore, declare pubsub.Topic[OrderPlaced]("order-placed"), publish with await order_placed.publish(order), and attach handlers with pubsub.Subscription[OrderPlaced](order_placed, "send-receipt", handler=send_receipt). encore run delivers messages locally, and on deploy the topic becomes SNS and SQS on AWS or Pub/Sub on GCP.
Are FastAPI BackgroundTasks enough for events?
BackgroundTasks run in the same process after the response is sent, with no retries and no persistence. If the process restarts, queued work is lost, so they suit best-effort work such as logging rather than receipts or payments.
Why do Pub/Sub subscribers need to be idempotent?
SNS with SQS and Google Cloud Pub/Sub deliver at least once, so a handler can see the same message again after a timeout or retry. A unique key, such as the order ID as a primary key with ON CONFLICT DO NOTHING, turns the duplicate into a no-op.
What is the difference between Pub/Sub and a task queue like Celery?
A task queue sends one job to one worker. Pub/Sub sends each event to every subscription, so receipts, inventory and analytics can each react to the same order without the publisher knowing about them.
Deploy to your cloud on AWS and Google Cloud
without managing infrastructure
Encore lets developers and agents define application infrastructure directly in code, then automatically provisions it from local development to production in your AWS or GCP account.
