mercure-publisher is a typed Python client to publish updates to a Mercure Hub, with Redis-backed caching of the generated publisher JWT so it isn't re-created on every publish call.
Features:
- 🔐 Builds the publisher JWT (
mercure.publishclaim) from a typed config - ⚡ Optional Redis cache for the JWT (checked before minting a new token)
- 🔁 Sync and async clients, both backed by
httpx - 🧩
dataaccepts adictand serializes it to JSON for you (or pass astrfor a raw payload) - 📦 Fully typed with dataclasses and type hints
pip install mercure-publisherRequires a reachable Redis instance (used to cache the publisher JWT); pass
redis_config=None to disable caching without removing the dependency.
from mercure_publisher import MercureConfig, RedisConfig
config = MercureConfig(
hub_url="https://hub.example.com/.well-known/mercure",
publisher_jwt_key="!ChangeThisMercureHubJWTSecretKey!",
publish_topics=["https://example.com/books/{id}"], # or ["*"] for any topic
jwt_ttl_seconds=3600,
redis_config=RedisConfig(
host="localhost",
port=6379,
key_prefix="mercure_publisher:",
),
)redis_config=None (the default) disables token caching — a new JWT is
generated on every publish.
from mercure_publisher import MercurePublisher
with MercurePublisher(config) as publisher:
event_id = publisher.publish(
topic="https://example.com/books/1",
data={"status": "published"},
)from mercure_publisher import AsyncMercurePublisher
async with AsyncMercurePublisher(config) as publisher:
event_id = await publisher.publish(
topic="https://example.com/books/1",
data={"status": "published"},
)publisher.publish(
topic=["https://example.com/books/1", "https://example.com/authors/1"],
data='{"status": "published"}',
private=True,
id="event-123",
type="book.updated",
retry=1000,
)MIT © Not Empty
Not Empty Foundation - Free codes, full minds