Metadata-Version: 2.4
Name: kevent
Version: 1.0.0
Summary: Typed event system for small Python applications
Project-URL: Homepage, https://github.com/kotazzz/kevent
Project-URL: Repository, https://github.com/kotazzz/kevent
Project-URL: Issues, https://github.com/kotazzz/kevent/issues
License: MIT
License-File: LICENSE
Keywords: cli,event-bus,events,middleware,typed
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.13
Description-Content-Type: text/markdown

# kevent

[Русский](README.ru.md) | 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

```python
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:

```bash
kevent demo "hello"
```

Expected output:

```text
middleware:DemoEvent
listener:hello
fallback:DemoEvent
```

## Installation

```bash
uv sync --dev
```

## Development

Run tests:

```bash
just test
```

Run linting and type checks:

```bash
just check
```

Format code:

```bash
just fmt
```

Build release artifacts:

```bash
just build
```

## Project Layout

```text
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](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:

```python
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`:

```python
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:

```python
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:

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

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

