kevent 1.0.0

Typed event system for small Python applications

kevent

Русский | English

kevent is a small, typed event system for Python applications. It provides a base Event type, declarative listeners, middleware hooks, and a lightweight handler that dispatches events by concrete type and by generic fallback.

What It Does

• Event is the base class for application events
• Listener binds one event class to one callback
• listener() is a decorator for declarative listener registration
• Middleware runs before listeners receive an event
• EventHandler stores listeners, runs middleware, and dispatches events

Quick Start

from kevent import Event, EventHandler, Middleware, listener


class UserLoggedIn(Event):
    def __init__(self, username: str) -> None:
        self.username = username


class TraceMiddleware(Middleware):
    def __init__(self) -> None:
        self.trace: list[str] = []

    def process(self, event: Event) -> Event | None:
        self.trace.append(type(event).__name__)
        return event  # Continue propagation


def on_login(event: UserLoggedIn) -> None:
    print(f"welcome:{event.username}")


def on_any(event: Event) -> None:
    print(f"received:{type(event).__name__}")


handler = EventHandler()
handler.add_middleware(TraceMiddleware())
handler.subscribe(listener(UserLoggedIn)(on_login))
handler.subscribe(listener(Event)(on_any))

handler.publish(UserLoggedIn("aria"))

CLI

Run a built-in demo that shows middleware and listener dispatch:

kevent demo "hello"

Expected output:

middleware:DemoEvent
listener:hello
fallback:DemoEvent

Installation

uv sync --dev

Development

Run tests:

just test

Run linting and type checks:

just check

Format code:

just fmt

Build release artifacts:

just build

Project Layout

kevent/
  __init__.py
  cli.py
  event.py
  handler.py
  listener.py
  middleware.py
examples/
  quickstart.py
tests/
  test_event_system.py

Notes

• The package has no runtime dependencies
• The public API is exported from kevent/init.py
• The CLI entry point is kevent = "kevent.cli:main"
• MRO-based dispatch is O(n) where n is the depth of the event class hierarchy; for deeply nested events, consider flattening or using explicit event types
• By default, exceptions in listeners propagate; use on_error to implement custom error strategies (logging, recovery, etc.)

Event Hierarchy and Dispatch

EventHandler.publish() uses method resolution order (MRO) to dispatch events. If you register a listener for a base event class, it will receive all subclass instances:

class UserEvent(Event):
    pass

class UserLoggedIn(UserEvent):
    pass

def on_user_event(event: UserEvent) -> None:
    print(f"user event: {type(event).__name__}")

handler = EventHandler()
handler.subscribe(listener(UserEvent)(on_user_event))
handler.publish(UserLoggedIn("alice"))  # Matches UserEvent via MRO

Middleware and Flow Control

Middleware can observe events and optionally cancel propagation by returning None:

class ValidationMiddleware(Middleware):
    def process(self, event: Event) -> Event | None:
        if not is_valid(event):
            return None  # Prevent listeners from receiving this event
        return event

handler = EventHandler()
handler.add_middleware(ValidationMiddleware())
handler.publish(event)  # Listeners won't run if validation fails

Error Handling

Catch exceptions in listeners without losing other events:

def on_error(exc: Exception, event: Event, listener_obj) -> None:
    print(f"Error in {listener_obj}: {exc}")

handler = EventHandler(on_error=on_error)
handler.subscribe(listener(Event)(on_event))  # If this raises, on_error is called
handler.publish(event)  # Other listeners still execute

Monitoring Listeners

Check how many listeners are registered:

handler = EventHandler()
print(handler.handlers_count)  # 0

handler.subscribe(listener(Event)(lambda e: None))
print(handler.handlers_count)  # 1