Signalbus

支持类型的小型库，用于广播信号

功能

• 异步支持
• 完全类型支持（获取错误）
• 小巧（约100行代码）且快速
• 您可以将一些逻辑封装在信号中

为什么还需要另一个库？

其他信号库没有良好的类型支持。

安装

$ pip install signalbus

用法

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/await之外，其他一切都是一样的

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)

通过参数过滤信号

您可以使用register函数设置任何参数来过滤接收器。只有当对应的参数匹配时，接收器才会被调用。

让我们考虑以下示例

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支持

为了更好地使用mypy进行类型检查，您必须为您的信号设置正确的返回类型

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()

无需使用Pyright进行此操作，因为Pyright可以正确计算类型

错误跟踪器

如果您有任何建议、错误报告或烦恼，请通过https://github.com/klen/signalbus/issues问题跟踪器报告它们

贡献

该库的开发发生在：https://github.com/klen/signalbus

许可协议

许可协议为MIT许可协议