aio-pydispatch 0.0.1.dev0

asyncio pydipatch (Signal Manager)

aio_pydispatch

Asyncio pydispatch (Signal Manager)

This is based on pyDispatcher reference scrapy SignalManager implementation on Asyncio

Usage

Like the situation often encountered on the web

Init some signals and a signal manager

import asyncio
import logging
from aio_pydispatch import SignalManager

logging.basicConfig(level=logging.INFO)

server_start = object()
server_stop = object()

signal_manager = SignalManager()

Do something when signal is trigger. Another expression: the signal subscribe the event

def start(data: str) -> None:
    # Do something when server is start.
    logging.info(f'started. {data}')

def stop(data: str) -> None:
    # Do something when server is stop.
    logging.info(f'stoped. {data}')

signal_manager.connect(start, server_start)
signal_manager.connect(start, server_stop)

Let's run fake server.

async def run():
    await signal_manager.send(server_start, data=f'xxx')
    await asyncio.sleep(5)
    await signal_manager.send(server_stop, data=f'xxx')


if __name__ == '__main__':
    asyncio.run(run())

There is all code:

import asyncio
import logging
from aio_pydispatch import SignalManager

logging.basicConfig(level=logging.INFO)


def start(data: str) -> None:
    logging.info(f'started. {data}')


def stop(data: str) -> None:
    logging.info(f'stopped. {data}')


server_start = object()
server_stop = object()

signal_manager = SignalManager()

signal_manager.connect(start, server_start)
signal_manager.connect(start, server_stop)


async def run():
    await signal_manager.send(server_start, data=f'xxx')
    await asyncio.sleep(5)
    await signal_manager.send(server_stop, data=f'xxx')


if __name__ == '__main__':
    asyncio.run(run())

Similar design

sync

• pyDispatcher
• Django.dispatch
• scrapy SignalManager
• blinker

Others

Event system in Python