# EZPubSub
[](https://pypi.org/project/ezpubsub/)
[](https://github.com/edward-jazzhands/ezpubsub/releases/latest)
[](https://python.org)
[](https://mypy-lang.org/)
[](https://opensource.org/license/mit)
A tiny, modern alternative to [Blinker](https://github.com/pallets-eco/blinker) – typed, thread-safe, sync or async, and designed for today’s Python.
EZPubSub is a zero-dependency pub/sub library focused on one thing: **making event publishing and subscribing easy, safe, and predictable.** No over-engineered, confusing API. No unnecessary features. Just clean, easy pub/sub that works anywhere.
The core design is inspired by the internal signal system in [Textual](https://textual.textualize.io/), refined into a standalone library built for general use.
## Quick Start
Synchronous signal:
```python
from ezpubsub import Signal
data_signal = Signal[str]()
def on_data(data: str) -> None:
print("Received:", data)
data_signal.subscribe(on_data)
data_signal.publish("Hello World")
# Output: Received: Hello World
```
Asynchronous signal with callback:
```python
from ezpubsub import Signal
async_data_signal = Signal[str]("My Database Update")
async def on_async_data(data: str) -> None:
await asyncio.sleep(1) # Simulate async work
print("Async Received:", data)
async_data_signal.asubscribe(on_async_data)
await async_data_signal.apublish("Hello Async World")
# Output: Async Received: Hello Async World
```
That’s it. You create a signal, subscribe to it, and publish events.
## Why Another Pub/Sub Library?
Because pub/sub in Python is either **old and untyped** or **overengineered and async-only**.
Writing a naive pub/sub system is easy. Just keep a list of callbacks and fire them. Writing one that actually works in production is not. You need to handle thread safety, memory management (weak refs for bound methods), error isolation, subscription lifecycles, and type safety. Most libraries get at least one of these wrong.
The last great attempt was Blinker, 15 years ago. It was excellent for its time, but Python has moved on. EZPubSub is what a pub/sub library should look like in 2025: type-safe, thread-safe, ergonomic, and designed for modern Python.
## Features
* **Thread-Safe by Default** – Publish and subscribe safely across threads.
* **Strongly Typed with Generics** – `Signal[str]`, `Signal[MyClass]`, or even TypedDict/dataclasses for structured events. Pyright/MyPy catches mistakes before runtime.
* **Sync or Async** – Works in any environment, including mixed sync/async projects.
* **Automatic Memory Management** – Bound methods are weakly referenced and auto-unsubscribed when their objects are deleted.
* **No Runtime Guesswork** – No `**kwargs`, no stringly-typed namespaces, no dynamic channel lookups. Opinionated design that enforces type safety and clarity.
* **Lightweight & Zero Dependencies** – Only what you need.
## How It Compares
### EZPubSub vs Blinker
Blinker is great for simple, single-threaded Flask-style apps. But:
| Feature | EZPubSub | Blinker |
| ----------------- | ----------------------------------- | -------------------------------------------------- |
| **Design** | ✅ Instance-based, type-safe | ⚠️ Channel-based (runtime filtering, string keys) |
| **Weak Refs** | ✅ Automatic | ✅ Automatic |
| **Type Checking** | ✅ Full static typing (`Signal[T]`) | ❌ Untyped (`Any`) |
| **Thread Safety** | ✅ Built-in | ❌ Single-threaded only |
If you’re starting a new project in 2025, you deserve type checking and thread safety out of the box.
### EZPubSub vs AioSignal
[`aiosignal`](https://github.com/aio-libs/aiosignal) is excellent for its niche—managing fixed async callbacks inside `aiohttp`—but unsuitable as a general pub/sub system:
| Feature | EZPubSub | AioSignal |
| ------------------------ | ------------------------------------ | --------------------------------------------------------------------- |
| **Sync and Async** | ✅ Sync and Async friendly | ❌ Sync publishing not available |
| **Freezing Subscribers** | ✅ Optional in both Sync and Async | ❌ `freeze()` required to publish; no dynamic add/remove at runtime |
| **Type Checking** | ✅ Full static typing (`Signal[T]`) | ⚠️ Allows arbitrary `**kwargs`, undermining type safety |
| **Thread Safety** | ✅ Built-in | ❌ Single-threaded only |
`aiosignal` is great if you’re writing an `aiohttp` extension. But being required to use async everywhere just to publish signals is unnecessary for most applications. Synchronous first, with optional async support, is simpler and more predictable. That’s why Blinker, Celery, and Django's internal PubSub (based on PyDispatcher) all share this design.
## Design Philosophy
### Signals vs Channels
EZPubSub uses **one object per signal**, instead of Blinker’s **“one channel, many signals”** model.
**Blinker (channel-based):**
```python
user_signal = Signal()
user_signal.connect(login_handler, sender=LoginService)
user_signal.send(sender=LoginService, user=user)
```
**EZPubSub (instance-based):**
```python
login_signal = Signal[LoginEvent]("user_login")
login_signal.subscribe(login_handler)
login_signal.publish(LoginEvent(user=user))
```
This matters because:
* **No filtering** – Each signal already represents one event type.
* **No runtime lookups** – You never hunt down signals by string name.
* **Type safety** – Wrong event types are caught by your IDE/type checker.
Fewer magic strings, fewer runtime bugs, and code that reads like what it does.
### Why No `**kwargs`?
Allowing arbitrary keyword arguments is convenient, but it destroys type safety.
```python
# Bad: fragile, stringly typed
signal.publish(user, session_id="abc123", ip="1.2.3.4")
# Good: explicit, type-safe
@dataclass
class UserLoginEvent:
user: User
session_id: str
ip: str
signal.publish(UserLoginEvent(user, "abc123", "1.2.3.4"))
```
This forces better API design and catches mistakes at compile time instead of runtime. EZPubSub is opinionated about this: **no `**kwargs`**. Every signal has a specific type.
It is of course possible to simply not use any type hinting when creating a signal (or use `Any`), as type hints in Python are optional. But the library is designed to encourage type safety by default. As for why you would ever want to create a signal using `Any` or without a type hint, I won't ask any questions ;)
## Installation
```sh
pip install ezpubsub
```
Or with [UV](https://github.com/astral-sh/uv):
```sh
uv add ezpubsub
```
Requires Python 3.10+.
---
## Documentation
Full docs: [**Click here**](https://edward-jazzhands.github.io/libraries/ezpubsub/docs/)
---
## License
MIT License. See [LICENSE](LICENSE) for details.
Raw data
{
"_id": null,
"home_page": null,
"name": "ezpubsub",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": null,
"keywords": "python, pubsub, publish-subscribe, typed, thread-safe, event-driven, events",
"author": "Edward Jazzhands",
"author_email": "Edward Jazzhands <ed.jazzhands@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/7e/9d/f0ba8b794d315cd0fb00f2d03e2036c8301afc678aabd72d8931aca4b28a/ezpubsub-0.3.0.tar.gz",
"platform": null,
"description": "# EZPubSub\n\n[](https://pypi.org/project/ezpubsub/)\n[](https://github.com/edward-jazzhands/ezpubsub/releases/latest)\n[](https://python.org)\n[](https://mypy-lang.org/)\n[](https://opensource.org/license/mit)\n\nA tiny, modern alternative to [Blinker](https://github.com/pallets-eco/blinker) \u2013 typed, thread-safe, sync or async, and designed for today\u2019s Python.\n\nEZPubSub is a zero-dependency pub/sub library focused on one thing: **making event publishing and subscribing easy, safe, and predictable.** No over-engineered, confusing API. No unnecessary features. Just clean, easy pub/sub that works anywhere.\n\nThe core design is inspired by the internal signal system in [Textual](https://textual.textualize.io/), refined into a standalone library built for general use.\n\n## Quick Start\n\nSynchronous signal:\n\n```python\nfrom ezpubsub import Signal\n\ndata_signal = Signal[str]()\n\ndef on_data(data: str) -> None:\n print(\"Received:\", data)\n\ndata_signal.subscribe(on_data)\ndata_signal.publish(\"Hello World\")\n# Output: Received: Hello World\n```\n\nAsynchronous signal with callback:\n\n```python\nfrom ezpubsub import Signal\n\nasync_data_signal = Signal[str](\"My Database Update\")\n\nasync def on_async_data(data: str) -> None:\n await asyncio.sleep(1) # Simulate async work\n print(\"Async Received:\", data)\n\nasync_data_signal.asubscribe(on_async_data)\nawait async_data_signal.apublish(\"Hello Async World\")\n# Output: Async Received: Hello Async World\n```\n\nThat\u2019s it. You create a signal, subscribe to it, and publish events.\n\n## Why Another Pub/Sub Library?\n\nBecause pub/sub in Python is either **old and untyped** or **overengineered and async-only**.\n\nWriting a naive pub/sub system is easy. Just keep a list of callbacks and fire them. Writing one that actually works in production is not. You need to handle thread safety, memory management (weak refs for bound methods), error isolation, subscription lifecycles, and type safety. Most libraries get at least one of these wrong.\n\nThe last great attempt was Blinker, 15 years ago. It was excellent for its time, but Python has moved on. EZPubSub is what a pub/sub library should look like in 2025: type-safe, thread-safe, ergonomic, and designed for modern Python.\n\n## Features\n\n* **Thread-Safe by Default** \u2013 Publish and subscribe safely across threads.\n* **Strongly Typed with Generics** \u2013 `Signal[str]`, `Signal[MyClass]`, or even TypedDict/dataclasses for structured events. Pyright/MyPy catches mistakes before runtime.\n* **Sync or Async** \u2013 Works in any environment, including mixed sync/async projects.\n* **Automatic Memory Management** \u2013 Bound methods are weakly referenced and auto-unsubscribed when their objects are deleted.\n* **No Runtime Guesswork** \u2013 No `**kwargs`, no stringly-typed namespaces, no dynamic channel lookups. Opinionated design that enforces type safety and clarity.\n* **Lightweight & Zero Dependencies** \u2013 Only what you need.\n\n## How It Compares\n\n### EZPubSub vs Blinker\n\nBlinker is great for simple, single-threaded Flask-style apps. But:\n\n| Feature | EZPubSub | Blinker |\n| ----------------- | ----------------------------------- | -------------------------------------------------- |\n| **Design** | \u2705 Instance-based, type-safe | \u26a0\ufe0f Channel-based (runtime filtering, string keys) |\n| **Weak Refs** | \u2705 Automatic | \u2705 Automatic |\n| **Type Checking** | \u2705 Full static typing (`Signal[T]`) | \u274c Untyped (`Any`) |\n| **Thread Safety** | \u2705 Built-in | \u274c Single-threaded only |\n\nIf you\u2019re starting a new project in 2025, you deserve type checking and thread safety out of the box.\n\n### EZPubSub vs AioSignal\n\n[`aiosignal`](https://github.com/aio-libs/aiosignal) is excellent for its niche\u2014managing fixed async callbacks inside `aiohttp`\u2014but unsuitable as a general pub/sub system:\n\n| Feature | EZPubSub | AioSignal |\n| ------------------------ | ------------------------------------ | --------------------------------------------------------------------- |\n| **Sync and Async** | \u2705 Sync and Async friendly | \u274c Sync publishing not available |\n| **Freezing Subscribers** | \u2705 Optional in both Sync and Async | \u274c `freeze()` required to publish; no dynamic add/remove at runtime |\n| **Type Checking** | \u2705 Full static typing (`Signal[T]`) | \u26a0\ufe0f Allows arbitrary `**kwargs`, undermining type safety |\n| **Thread Safety** | \u2705 Built-in | \u274c Single-threaded only |\n\n`aiosignal` is great if you\u2019re writing an `aiohttp` extension. But being required to use async everywhere just to publish signals is unnecessary for most applications. Synchronous first, with optional async support, is simpler and more predictable. That\u2019s why Blinker, Celery, and Django's internal PubSub (based on PyDispatcher) all share this design.\n\n## Design Philosophy\n\n### Signals vs Channels\n\nEZPubSub uses **one object per signal**, instead of Blinker\u2019s **\u201cone channel, many signals\u201d** model.\n\n**Blinker (channel-based):**\n\n```python\nuser_signal = Signal() \nuser_signal.connect(login_handler, sender=LoginService)\nuser_signal.send(sender=LoginService, user=user)\n```\n\n**EZPubSub (instance-based):**\n\n```python\nlogin_signal = Signal[LoginEvent](\"user_login\")\nlogin_signal.subscribe(login_handler)\nlogin_signal.publish(LoginEvent(user=user))\n```\n\nThis matters because:\n\n* **No filtering** \u2013 Each signal already represents one event type.\n* **No runtime lookups** \u2013 You never hunt down signals by string name.\n* **Type safety** \u2013 Wrong event types are caught by your IDE/type checker.\n\nFewer magic strings, fewer runtime bugs, and code that reads like what it does.\n\n### Why No `**kwargs`?\n\nAllowing arbitrary keyword arguments is convenient, but it destroys type safety.\n\n```python\n# Bad: fragile, stringly typed\nsignal.publish(user, session_id=\"abc123\", ip=\"1.2.3.4\")\n\n# Good: explicit, type-safe\n@dataclass\nclass UserLoginEvent:\n user: User\n session_id: str\n ip: str\n\nsignal.publish(UserLoginEvent(user, \"abc123\", \"1.2.3.4\"))\n```\n\nThis forces better API design and catches mistakes at compile time instead of runtime. EZPubSub is opinionated about this: **no `**kwargs`**. Every signal has a specific type.\n\nIt is of course possible to simply not use any type hinting when creating a signal (or use `Any`), as type hints in Python are optional. But the library is designed to encourage type safety by default. As for why you would ever want to create a signal using `Any` or without a type hint, I won't ask any questions ;)\n\n## Installation\n\n```sh\npip install ezpubsub\n```\n\nOr with [UV](https://github.com/astral-sh/uv):\n\n```sh\nuv add ezpubsub\n```\n\nRequires Python 3.10+.\n\n---\n\n## Documentation\n\nFull docs: [**Click here**](https://edward-jazzhands.github.io/libraries/ezpubsub/docs/)\n\n---\n\n## License\n\nMIT License. See [LICENSE](LICENSE) for details.\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "An ultra simple, modern pub/sub library and blinker alternative for Python",
"version": "0.3.0",
"project_urls": {
"Changelog": "https://github.com/edward-jazzhands/ezpubsub/blob/master/CHANGELOG.md",
"Repository": "https://github.com/edward-jazzhands/ezpubsub"
},
"split_keywords": [
"python",
" pubsub",
" publish-subscribe",
" typed",
" thread-safe",
" event-driven",
" events"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "642913f734871548d821eebd51b5ddedef29f9c2791d83dc910347c6411b7c19",
"md5": "707288566eabc77b5052491335865e39",
"sha256": "82af16d3bc3df43c58ab1fc99d0afd20457dd292ce34569fd12d30f1353c209a"
},
"downloads": -1,
"filename": "ezpubsub-0.3.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "707288566eabc77b5052491335865e39",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 7530,
"upload_time": "2025-08-03T09:32:01",
"upload_time_iso_8601": "2025-08-03T09:32:01.945432Z",
"url": "https://files.pythonhosted.org/packages/64/29/13f734871548d821eebd51b5ddedef29f9c2791d83dc910347c6411b7c19/ezpubsub-0.3.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "7e9df0ba8b794d315cd0fb00f2d03e2036c8301afc678aabd72d8931aca4b28a",
"md5": "daea840d21bcf94da7578d15721f29c0",
"sha256": "5242b2229561e96f0903163807170560d183afe8e601011c5f32ebf43535f0d5"
},
"downloads": -1,
"filename": "ezpubsub-0.3.0.tar.gz",
"has_sig": false,
"md5_digest": "daea840d21bcf94da7578d15721f29c0",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 6875,
"upload_time": "2025-08-03T09:32:03",
"upload_time_iso_8601": "2025-08-03T09:32:03.397255Z",
"url": "https://files.pythonhosted.org/packages/7e/9d/f0ba8b794d315cd0fb00f2d03e2036c8301afc678aabd72d8931aca4b28a/ezpubsub-0.3.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-03 09:32:03",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "edward-jazzhands",
"github_project": "ezpubsub",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "ezpubsub"
}
JSON
