# Pub/Sub

> Decoupling services and building asynchronous systems

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.

```python
-- 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)
```

```python
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:

```python
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](https://en.wikipedia.org/wiki/Idempotence#Computer_science_meaning)**, 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:

```python
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.

> **Important**
>
> Exactly-once delivery doesn't deduplicate on the publishing side. If you publish the same message twice, it's delivered twice.

### 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:

```python
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](https://en.wikipedia.org/wiki/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:

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

The message can be:

- A top-level [Pydantic](https://docs.pydantic.dev) 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`:

```python
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`:

```python
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:

```python
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:

```python
-- 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](/docs/py/develop/shutdown).

### Message payloads

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

| Declaration | `message.data` |
| --- | --- |
| `Subscription[MyModel]` | An instance of the Pydantic v2 model, validated with `model_validate_json`. |
| `Subscription[pubsub.JSON]` or plain `Subscription` | Ordinary Python values from `json.loads`, without validation. |
| `Subscription[bytes]` | The raw payload, unchanged. |

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:

| Field | Description |
| --- | --- |
| `data` | The decoded payload. |
| `raw_data` | The original payload bytes. |
| `attributes` | The attributes published with the message, as `dict[str, str]`. |
| `id` | The message ID, as returned by `publish`. |
| `delivery_attempt` | Which delivery attempt this is, starting at 1. |
| `publish_time` | When the message was published, as a UTC `datetime`. |
| `ordering_key` | The ordering key, for a message published to an `OrderedTopic`; otherwise `None`. |

### Subscription options

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

```python
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,
)
```

| Option | Default | Description |
| --- | --- | --- |
| `ack_deadline` | `"30s"` | How long a handler has to process a message before it may be redelivered. |
| `message_retention` | `"7d"` | How long undelivered messages are kept. |
| `min_backoff` | `"10s"` | The shortest delay before retrying a failed message. |
| `max_backoff` | `"10m"` | The longest delay before retrying a failed message. |
| `max_retries` | `100` | How many times a failed message is retried before it's dead-lettered. `-1` retries forever; `-2` dead-letters a failed message without retrying. |
| `max_concurrency` | provider default | The most messages handled at once by each instance of the service. |

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

> **Note**
>
> Encore reads these options from your source code to provision the subscription, so write them as literals in the `Subscription` call. A value computed at run time, such as a variable or a function call, isn't seen, and the default is used instead.

## 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:

```shell
$ 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](/docs/cli/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](/docs/py/develop/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](https://microservices.io/patterns/data/transactional-outbox.html): write the event to an outbox table in the same transaction as your other changes, and publish it from there.