# Signalbus
Simple and small library to broadcast signals with typing support
[](https://github.com/klen/signalbus/actions/workflows/tests.yml)
[](https://github.com/klen/signalbus/actions/workflows/publish.yml)
Features:
* Async support
* Full typing support (get errors)
* Small (around 100 lines of code) and fast
* You may incapsulate some logic inside a signal
## Why another library?
Other signals library don't have a good typing support.
## Installation
$ pip install signalbus
## Usage
```python
from signalbus import create_signal
# Create a signals
# Just define a (generator) function and wrap it with `signalbus.create_signal`
@create_signal
def order_changed(order_status: str, *, order_id: int): #
"""
The function contains the signal code.
Feel free to do some operations before and after the sending.
Pay attention to the function's params
All receivers for the signal have to be able to accept the same params.
Typing libraries will show you errors.
"""
# first, you have to get `emit` to be able to send the signal
emit = yield
# then send the signal to the receivers (you may want to skip it in some cases)
res: list = emit(order_status, order_id=order_id)
# you may check the results, do some additional work, etc
# Register a receiver for the signal
# The receiver has to have the same params (types will be checked)
@order_changed.register
def notify_user(order_status: str, *, order_id: int):
...
@order_changed.register
def update_stats(order_status: str, *, order_id: int):
...
# To send the signal just call it like a function with all required params
order_changed('done', order_id=42)
```
### Async Signals
Everything is almost the same except async/await
```python
from signalbus import create_signal
@create_signal
async def order_changed(order_status: str, *, order_id: int):
emit = yield
res: list = await emit(order_status, order_id=order_id)
# Receiver has to be async too
@order_changed.register
async def notify_user(order_status: str, *, order_id: int):
...
@order_changed.register
async def update_stats(order_status: str, *, order_id: int):
...
# Do not forget to await the signal
await order_changed('done', order_id=42)
```
### Filter signals by arguments
You may set any arguments to filter a receiver with the register function. The
receiver would be called only when corresponding arguments match.
Let's consider the following example:
```python
from signalbus import create_signal
@create_signal
async def order_changed(order_status: str, *, order_id: int):
emit = yield
res: list = await emit(order_status, order_id=order_id)
# pay attention to that we define an attribute in register
@order_changed.register('done')
async def notify_user(order_status: str, *, order_id: int):
...
@order_changed.register
async def update_stats(order_status: str, *, order_id: int):
...
await order_changed('done', order_id=42) # both the receivers above will be called
await order_changed('cancel', order_id=42) # only update stats will be called
```
### Mypy support
For better typing with mypy, you have to set correct returning type for your signals:
```python
from signalbus import create_signal
from typing import Generator, AsyncGenerator
@create_signal
def sync_signal() -> Generator:
emit = yield
res: list = await emit()
@create_signal
async def async_signal() -> AsyncGenerator:
emit = yield
res: list = await emit()
```
No need to do it with Pyright, because the Pyright calculates the types correctly
## Bug tracker
If you have any suggestions, bug reports or annoyances please report them to
the issue tracker at https://github.com/klen/signalbus/issues
## Contributing
Development of The library happens at: https://github.com/klen/signalbus
## License
Licensed under a [MIT license](https://opensource.org/license/mit/)
Raw data
{
"_id": null,
"home_page": "https://github.com/klen/signalbus",
"name": "signalbus",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.8,<4.0",
"maintainer_email": "",
"keywords": "signal,emit,events,broadcast",
"author": "Kirill Klenov",
"author_email": "horneds@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/9f/ec/014252c3330fb65920918b89309a9769152fa259123f702851e9cb1c1922/signalbus-0.2.5.tar.gz",
"platform": null,
"description": "# Signalbus\n\nSimple and small library to broadcast signals with typing support\n\n[](https://github.com/klen/signalbus/actions/workflows/tests.yml)\n[](https://github.com/klen/signalbus/actions/workflows/publish.yml)\n\nFeatures:\n\n* Async support\n* Full typing support (get errors)\n* Small (around 100 lines of code) and fast\n* You may incapsulate some logic inside a signal\n\n## Why another library?\n\nOther signals library don't have a good typing support.\n\n## Installation\n\n $ pip install signalbus\n\n## Usage\n\n```python\n\nfrom signalbus import create_signal\n\n# Create a signals\n# Just define a (generator) function and wrap it with `signalbus.create_signal`\n\n@create_signal\ndef order_changed(order_status: str, *, order_id: int): #\n \"\"\"\n The function contains the signal code.\n Feel free to do some operations before and after the sending.\n\n Pay attention to the function's params\n All receivers for the signal have to be able to accept the same params.\n Typing libraries will show you errors.\n \"\"\"\n # first, you have to get `emit` to be able to send the signal\n emit = yield\n\n # then send the signal to the receivers (you may want to skip it in some cases)\n res: list = emit(order_status, order_id=order_id)\n\n # you may check the results, do some additional work, etc\n\n\n# Register a receiver for the signal\n# The receiver has to have the same params (types will be checked)\n@order_changed.register\ndef notify_user(order_status: str, *, order_id: int):\n ...\n\n\n@order_changed.register\ndef update_stats(order_status: str, *, order_id: int):\n ...\n\n\n# To send the signal just call it like a function with all required params\norder_changed('done', order_id=42)\n\n```\n\n\n### Async Signals\n\nEverything is almost the same except async/await\n\n```python\n\nfrom signalbus import create_signal\n\n@create_signal\nasync def order_changed(order_status: str, *, order_id: int):\n emit = yield\n res: list = await emit(order_status, order_id=order_id)\n\n\n# Receiver has to be async too\n@order_changed.register\nasync def notify_user(order_status: str, *, order_id: int):\n ...\n\n\n@order_changed.register\nasync def update_stats(order_status: str, *, order_id: int):\n ...\n\n\n# Do not forget to await the signal\nawait order_changed('done', order_id=42)\n```\n\n### Filter signals by arguments\n\nYou may set any arguments to filter a receiver with the register function. The\nreceiver would be called only when corresponding arguments match.\n\nLet's consider the following example:\n\n```python\nfrom signalbus import create_signal\n\n@create_signal\nasync def order_changed(order_status: str, *, order_id: int):\n emit = yield\n res: list = await emit(order_status, order_id=order_id)\n\n\n# pay attention to that we define an attribute in register\n@order_changed.register('done')\nasync def notify_user(order_status: str, *, order_id: int):\n ...\n\n\n@order_changed.register\nasync def update_stats(order_status: str, *, order_id: int):\n ...\n\n\nawait order_changed('done', order_id=42) # both the receivers above will be called\nawait order_changed('cancel', order_id=42) # only update stats will be called\n```\n\n\n### Mypy support\n\nFor better typing with mypy, you have to set correct returning type for your signals:\n```python\nfrom signalbus import create_signal\n\nfrom typing import Generator, AsyncGenerator\n\n@create_signal\ndef sync_signal() -> Generator:\n emit = yield\n res: list = await emit()\n\n\n@create_signal\nasync def async_signal() -> AsyncGenerator:\n emit = yield\n res: list = await emit()\n```\n\nNo need to do it with Pyright, because the Pyright calculates the types correctly\n\n## Bug tracker\n\nIf you have any suggestions, bug reports or annoyances please report them to\nthe issue tracker at https://github.com/klen/signalbus/issues\n\n\n## Contributing\n\nDevelopment of The library happens at: https://github.com/klen/signalbus\n\n\n## License\n\nLicensed under a [MIT license](https://opensource.org/license/mit/)\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Simple and small library to broadcast signals with typing support",
"version": "0.2.5",
"project_urls": {
"Homepage": "https://github.com/klen/signalbus"
},
"split_keywords": [
"signal",
"emit",
"events",
"broadcast"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "727d5f4a3f2a9079b18de7cfe2f77256760e51737d805fd87fcf4ba2ff6c1a4b",
"md5": "f41084646a4068e21a118417daaf1399",
"sha256": "c6a940863cdaa70b6cbf1e48477cc29626253246b9678d028601d3fbf00294cc"
},
"downloads": -1,
"filename": "signalbus-0.2.5-py3-none-any.whl",
"has_sig": false,
"md5_digest": "f41084646a4068e21a118417daaf1399",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8,<4.0",
"size": 3641,
"upload_time": "2023-11-02T10:04:59",
"upload_time_iso_8601": "2023-11-02T10:04:59.320174Z",
"url": "https://files.pythonhosted.org/packages/72/7d/5f4a3f2a9079b18de7cfe2f77256760e51737d805fd87fcf4ba2ff6c1a4b/signalbus-0.2.5-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "9fec014252c3330fb65920918b89309a9769152fa259123f702851e9cb1c1922",
"md5": "d8357e70e2267a8f523b52fed91796ef",
"sha256": "01233a57ebf0063b7576e9500e1544edc2c0d7ecffb06da048d3c959f997da9e"
},
"downloads": -1,
"filename": "signalbus-0.2.5.tar.gz",
"has_sig": false,
"md5_digest": "d8357e70e2267a8f523b52fed91796ef",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8,<4.0",
"size": 3451,
"upload_time": "2023-11-02T10:05:00",
"upload_time_iso_8601": "2023-11-02T10:05:00.380334Z",
"url": "https://files.pythonhosted.org/packages/9f/ec/014252c3330fb65920918b89309a9769152fa259123f702851e9cb1c1922/signalbus-0.2.5.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-11-02 10:05:00",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "klen",
"github_project": "signalbus",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "signalbus"
}
JSON
