Pub/Sub

Publishers and subscribers (Pub/Sub) let services communicate by broadcasting events asynchronously. It's a good way to decouple services for better reliability and responsiveness.

With Encore.py you declare topics and subscriptions in code, and Encore provisions the infrastructure: a local broker when you run your app with encore run, and a cloud service such as GCP Pub/Sub or AWS SNS/SQS when you deploy. Publishing and message handling are traced automatically.

orders/events.py
from encore import pubsub
from pydantic import BaseModel


class OrderPlaced(BaseModel):
    order_id: int
    amount: int


orders = pubsub.Topic[OrderPlaced]("order-placed")


async def send_receipt(message: pubsub.Message[OrderPlaced]) -> None:
    print("Sending receipt for order", message.data.order_id)


receipts = pubsub.Subscription[OrderPlaced](orders, "send-receipt", handler=send_receipt)

await orders.publish(OrderPlaced(order_id=1, amount=100))

Creating a topic

A topic is a named channel you publish events to. Create one with pubsub.Topic, assigned to a module-level variable:

from encore import pubsub

signups = pubsub.Topic("user.signed-up")

Encore reads the declaration from your source code to provision the topic, so:

• Declare it as a module-level variable, not inside a function, class, or if block.
• The name must be a string literal: lowercase letters, digits, hyphens and periods, unique within the app.
• Options must be literals, such as exactly_once=True.

A topic can be published to from any service and subscribed to from any service.

The type argument, as in pubsub.Topic[OrderPlaced], is optional. It lets your type checker check what you publish; it doesn't change how messages are encoded.

At-least-once delivery

By default a topic delivers each event at least once to each subscription: if Encore believes a message wasn't processed, it delivers it again. Subscription handlers should therefore be idempotent, so that handling a message twice has the same effect as handling it once. For example, record in a database which events you've already acted on.

Exactly-once delivery

Set exactly_once=True to have the cloud provider minimize redelivery:

payments = pubsub.Topic("payment-received", exactly_once=True)

Messages can still be redelivered in rare cases, for example when a network failure loses the acknowledgement of a processed message, so handlers that must be correct under all circumstances should still be idempotent. Exactly-once delivery also limits throughput: 300 messages per second per topic on AWS, and at least 3,000 messages per second across the region's topics on GCP.

Ordered topics

Topics are unordered by default: messages can be delivered in any order, which lets them be processed in parallel. When messages about the same entity must be handled in the order they were published, declare an OrderedTopic and give each message an ordering key when you publish it:

cart_events = pubsub.OrderedTopic("cart-events")


async def checkout(cart_id: int) -> None:
    # Delivered in this order, since they share an ordering key.
    await cart_events.publish({"event": "checkout_started"}, ordering_key=str(cart_id))
    await cart_events.publish({"event": "checkout_completed"}, ordering_key=str(cart_id))

Messages with the same ordering key are delivered to each subscription in the order they were published. Messages with different keys are delivered in an unspecified order relative to each other.

A message isn't delivered until the one before it with the same key has been processed or dead-lettered, so a failing message holds up the messages behind it (head-of-line blocking). Choose a suitable retry policy, and log and alert on failures. Ordered topics are also limited in throughput: 300 messages per second per topic on AWS, and 1 MBps per ordering key on GCP.

OrderedTopic takes the same exactly_once option as Topic. The local broker delivers messages in ordering-key order too, so you can test the behavior with encore run.

Publishing events

Call publish on the topic. It returns the message's ID:

message_id = await signups.publish({"user_id": 42})

The message can be:

• A top-level Pydantic v2 model, serialized with model_dump_json().
• Any JSON-serializable value (dicts, lists, strings, numbers, booleans, None), serialized with json.dumps.
• bytes, sent unchanged.

Anything else, such as a dataclass or a Pydantic model nested in a plain dict, raises TypeError. Wrap nested models in a top-level model instead.

You can send string attributes alongside the payload. Subscribers receive them in message.attributes:

await signups.publish({"user_id": 42}, attributes={"source": "web"})

For an OrderedTopic, ordering_key is required: publish(message, ordering_key, attributes=None).

Topics declared elsewhere

To publish to or subscribe to a topic declared by another service, possibly one written in another language, reference it by name with named:

from encore import pubsub

signups = pubsub.Topic.named("user.signed-up")

Use pubsub.OrderedTopic.named for an ordered topic. This doesn't create a topic. Like a declaration, it must be the value of a module-level variable, the name must be a string literal, and Encore reports an error if no topic with that name is declared in the app.

Encore reads which services publish to each topic from the publish calls in your code, and grants each service the permissions it needs. Call publish on the topic's module-level variable, directly or through an import, rather than passing the topic around.

Subscribing to events

A subscription receives every event published to its topic and calls your handler for each one. Create one with pubsub.Subscription, assigned to a module-level variable:

from encore import pubsub
from pydantic import BaseModel


class SignupEvent(BaseModel):
    user_id: int


signups = pubsub.Topic[SignupEvent]("user.signed-up")


async def send_welcome_email(message: pubsub.Message[SignupEvent]) -> None:
    print("Welcoming user", message.data.user_id)


welcome_emails = pubsub.Subscription[SignupEvent](
    signups,
    "send-welcome-email",
    handler=send_welcome_email,
)

The arguments are:

• The topic: the variable of a topic declared in, imported into, or referenced with named in this module.
• The subscription's name, a string literal that is unique among the topic's subscriptions. Use kebab-case, such as send-welcome-email.
• handler, a keyword-only async function that takes a pubsub.Message. Pass the function itself (handler=send_welcome_email), not the result of calling it.

The service whose code declares the subscription is the one that receives its messages. A topic can have many subscriptions, in the same or different services, and each receives every message.

Starting subscriptions

Subscriptions start receiving messages when you call await encore.start() on your application's event loop. Call it once at startup, after the modules that declare your subscriptions have been imported. With FastAPI, use the lifespan hook:

main.py
from contextlib import asynccontextmanager

import encore
from fastapi import FastAPI

import orders.events  # noqa: F401 (declares the subscriptions)


@asynccontextmanager
async def lifespan(app: FastAPI):
    await encore.start()
    yield


app = FastAPI(lifespan=lifespan)

encore.start() returns once the subscriptions are running; handlers run on the same event loop as your application. Calling it again is safe, and starts any subscriptions declared in modules imported since. If a subscription declared in your service has no handler running, for example because its module was never imported, Encore logs a warning shortly after startup.

When the service shuts down, Encore stops taking new messages and waits for running handlers to finish. See Graceful shutdown.

Message payloads

The subscription's type argument decides how each message is decoded before your handler is called:

The handler's annotation is only for your type checker; it doesn't select the decoder, so keep it in step with the subscription's type argument. Use a Pydantic model when fields need validation or nested models need to be constructed. Dataclasses, unions of models, models nested inside plain JSON containers and string forward references aren't supported; use a top-level model, or Subscription[bytes] and decode the payload yourself.

A payload that fails to decode raises in place of calling your handler, and is retried like any other failure.

The message

The handler receives a pubsub.Message with these fields:

Subscription options

Subscription takes keyword-only options for message retention, retries and concurrency:

welcome_emails = pubsub.Subscription[SignupEvent](
    signups,
    "send-welcome-email",
    handler=send_welcome_email,
    ack_deadline="60s",
    min_backoff="5s",
    max_backoff="5m",
    max_retries=10,
    max_concurrency=20,
)

Durations are strings of numbers and units (ms, s, m, h, d), such as "30s", "1h30m" or "7d".

Error handling

If a handler raises an exception, the message is retried after a backoff between min_backoff and max_backoff. Once it has been retried max_retries times, it's moved to the subscription's dead-letter queue, so the subscription can carry on with other messages until you've fixed the problem. You can then release dead-lettered messages to be processed again.

Locally, use the encore pubsub commands to see and manage what happened to each message:

$ encore pubsub ls                                         # topics, subscriptions and their queues
$ encore pubsub queue user.signed-up send-welcome-email --state dead
$ encore pubsub show <message-id>                          # a message and its delivery attempts
$ encore pubsub release user.signed-up send-welcome-email  # deliver dead messages again
$ encore pubsub publish user.signed-up '{"user_id": 42}'

encore pubsub also has history, pause, resume, discard and purge. See the CLI reference.

Local development and testing

encore run runs a local Pub/Sub broker, so topics and subscriptions work without any cloud account. Exactly-once delivery isn't emulated locally.

Under encore test, each test gets its own isolated instance of the topics and subscriptions it uses, so messages published in one test don't reach another. See Testing.

Ensuring consistency between services

Writing to your database and publishing an event aren't one transaction: if one succeeds and the other fails, your services can disagree. If that matters, consider the transactional outbox pattern: write the event to an outbox table in the same transaction as your other changes, and publish it from there.