tiny-listener 1.1.0

lightning-fast, high-performance event handle framework

A lightweight event framework

---

Documentation: https://molto0504.github.io/tiny-listener

Source Code: https://github.com/molto0504/tiny-listener

---

Introduction

Tiny-listener is a lightweight and flexible event framework.

Requirements

Python 3.8+

Installation

$ pip install tiny-listener

Why use tiny-listener

• ✔ Pure Python.
• ✔ Lightning-fast, based on native coroutine.
• ✔ 100% test coverage.

Example

Consider we want to write a program keep listening message received from somewhere (e.g. http request or message queue), when message arrived, we want to execute a series of operations in a certain order, e.g.

• step 1: save user data to database
• step 2: send email to user

In the development process, we often encounter such problems:

• the source of step 1 and step 2 may be different
• step 2 may depend on the result of step 1, that is to say, step 2 may need to wait for step 1 to complete
• there are many messages, many events, and it is not easy to maintain

Tiny-listener may help you solve these problems:

Create a file example.py with:

from tiny_listener import Event, Listener


class App(Listener):
    async def listen(self):
        ctx = self.new_ctx()
        ctx.trigger_event("step 2: send email to [e-mail]")
        ctx.trigger_event("step 1: save Alice's data to database", data={"age": 35})


app = App()


@app.on_event("step 1: save {username}'s data to database")
async def step_1(event: Event):
    username = event.params["username"]
    age = event.data["age"]
    print(f"Save data done!, {username=}, {age=}")


@app.on_event("step 2: send email to {email}")
async def step_2(event: Event):
    await event.wait_event_done("step_1")
    email = event.params["email"]
    print(f"Send email done!, {email=}")

Run it:

$ tiny-listener example:app
>>> Save data done!, username='Alice', age=35
>>> Send email done!, email='[e-mail]'

How it works

• Create your own Listener and listen something(e.g. port, queue ...):

from tiny_listener import Event, Listener


class App(Listener):
    async def listen(self):
        ctx = self.new_ctx()
        ctx.trigger_event("step 2: send email to [e-mail]")
        ctx.trigger_event("step 1: save Alice's data to database", data={"age": 35})

• Add event handler to your listener:

app = App()


@app.on_event("step 1: save {username}'s data to database")
async def step_1(event: Event):
    username = event.params["username"]
    age = event.data["age"]
    print(f"Save data done!, {username=}, {age=}")


@app.on_event("step 2: send email to {email}")
async def step_2(event: Event):
    await event.wait_event_done("step_1")
    email = event.params["email"]
    print(f"Send email done!, {email=}")

• Run listener with command:

$ tiny-listener example:app

• Tiny-listener will dispatch every event automatically:

>>> Save data done!, username='Alice', age=35
>>> Send email done!, email='[e-mail]'