# Persistent Pub/Sub (/sdks/python/tutorials/persistent-pubsub)



## Overview [#overview]

This tutorial builds a publisher and subscriber on a KubeMQ Events Store channel — reach for this pattern when a subscriber can't guarantee it's listening the instant a message is published. Plain events are fire-and-forget: publish with no one subscribed and the message is gone. Events Store persists every event to a durable, ordered log, so a subscriber connecting seconds or a full restart later still catches up — useful for anything needing a complete history, like an audit trail or event-sourced state.

The two calls involved: `send_event_store` publishes and returns an `EventStoreResult` confirming storage plus a broker-assigned sequence number, and `subscribe_to_events_store` takes a required start position telling the broker where to start — new events only (`EventStoreStartPosition.StartFromNew`, used here), from the first stored event, or a given sequence or time. Production subscribers usually resume from a saved checkpoint instead of starting fresh.

**Gotchas:** starting from new events means anything published earlier is silently skipped — this sample papers over that race with a fixed `asyncio.sleep` instead of a ready signal, fine for a demo but not production. Replaying from the first event on every restart replays the whole log, which gets costly on a busy channel. Persistence isn't consumer coordination: each independent subscriber gets its own full replay unless grouped with a consumer group.

## Prerequisites [#prerequisites]

* KubeMQ server running on `localhost:50000`
* Python SDK installed (`pip install kubemq`)

## Code [#code]

```python title="persistent_pubsub.py"
"""Example: Persistent pub/sub — publish and subscribe to events store with persistence."""

from __future__ import annotations

import asyncio

from kubemq import (
    AsyncCancellationToken,
    AsyncPubSubClient,
    EventStoreMessage,
    EventStoreReceived,
    EventsStoreSubscription,
    KubeMQConnectionError,
    KubeMQError,
)
from kubemq.pubsub import EventStoreStartPosition


async def main() -> None:
    try:
        async with AsyncPubSubClient(
            address="localhost:50000",
            client_id="python-events-store-persistent-pubsub-client",
        ) as client:
            token = AsyncCancellationToken()

            async def subscriber() -> None:
                async for event in client.subscribe_to_events_store(
                    subscription=EventsStoreSubscription(
                        channel="python-events-store.persistent-pubsub",
                        on_receive_event_callback=lambda e: None,
                        on_error_callback=lambda e: print(f"Error: {e}"),
                        events_store_type=EventStoreStartPosition.StartFromNew,
                    ),
                    cancellation_token=token,
                ):
                    print(
                        f"Received — Id:{event.id}, Seq:{event.sequence}, "
                        f"Body:{event.body.decode('utf-8')}"
                    )

            task = asyncio.create_task(subscriber())
            await asyncio.sleep(1)

            result = await client.send_event_store(
                EventStoreMessage(
                    channel="python-events-store.persistent-pubsub",
                    body=b"hello kubemq",
                )
            )
            print(f"Send result: {result}")

            await asyncio.sleep(2)
            token.cancel()
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass
    except KubeMQConnectionError as e:
        print(f"Connection error: {e}")
    except KubeMQError as e:
        print(f"KubeMQ error: {e}")


if __name__ == "__main__":
    asyncio.run(main())

# Expected output:
# Received — Id:<message-id>, Seq:<sequence>, Body:hello kubemq
# Send result: <result>

```

## How It Works [#how-it-works]

* `send_event_store` returns an `EventStoreResult` with an `id` and `sent` flag, confirming the event was persisted on the broker.
* `EventStoreStartPosition.StartFromNew` means the subscriber only receives events published after the subscription is established.
* The `subscribe_to_events_store` async generator yields `EventStoreReceived` objects that include a monotonically increasing `sequence` number.
* Unlike plain events, stored events survive subscriber restarts; a new subscriber can replay history by changing the start position.

## Related [#related]

* [Pattern overview](/learn/events-store/getting-started)
* [Python SDK Reference](/sdks/python/reference/events-store)
* [Cancel Subscription](/sdks/python/how-to/events-store/cancel-subscription)
* [Consumer Group](/sdks/python/how-to/events-store/consumer-group)