A custom event that can notify multiple listeners when triggered. Use this class to create your own event system within Python code. Listeners can be either regular functions or async functions. ```python from pyscript.events import Event # Create a custom event. data
| 21 | |
| 22 | |
| 23 | class Event: |
| 24 | """ |
| 25 | A custom event that can notify multiple listeners when triggered. |
| 26 | |
| 27 | Use this class to create your own event system within Python code. |
| 28 | Listeners can be either regular functions or async functions. |
| 29 | |
| 30 | ```python |
| 31 | from pyscript.events import Event |
| 32 | |
| 33 | # Create a custom event. |
| 34 | data_loaded = Event() |
| 35 | |
| 36 | # Add a listener. |
| 37 | def on_data_loaded(result): |
| 38 | print(f"Data loaded: {result}") |
| 39 | |
| 40 | data_loaded.add_listener(on_data_loaded) |
| 41 | |
| 42 | # Time passes.... trigger the event. |
| 43 | data_loaded.trigger({"data": 123}) |
| 44 | ``` |
| 45 | """ |
| 46 | |
| 47 | def __init__(self): |
| 48 | self._listeners = [] |
| 49 | |
| 50 | def trigger(self, result): |
| 51 | """ |
| 52 | Trigger the event and notify all listeners with the given `result`. |
| 53 | """ |
| 54 | for listener in self._listeners: |
| 55 | if is_awaitable(listener): |
| 56 | asyncio.create_task(listener(result)) |
| 57 | else: |
| 58 | listener(result) |
| 59 | |
| 60 | def add_listener(self, listener): |
| 61 | """ |
| 62 | Add a function to be called when this event is triggered. |
| 63 | |
| 64 | The `listener` must be callable. It can be either a regular function |
| 65 | or an async function. Duplicate listeners are ignored. |
| 66 | """ |
| 67 | if not callable(listener): |
| 68 | msg = "Listener must be callable." |
| 69 | raise ValueError(msg) |
| 70 | if listener not in self._listeners: |
| 71 | self._listeners.append(listener) |
| 72 | |
| 73 | def remove_listener(self, *listeners): |
| 74 | """ |
| 75 | Remove specified `listeners`. If none specified, remove all listeners. |
| 76 | """ |
| 77 | if listeners: |
| 78 | for listener in listeners: |
| 79 | try: |
| 80 | self._listeners.remove(listener) |
no outgoing calls