Skip to content

tom-bartk/eventoolkit

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

1 Commits
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

Logo

Client-side toolkit for abstract events.
Documentation

Simple example

# main.py

from collections.abc import Sequence
from typing import NamedTuple

import pydepot

from eventoolkit import Event, EventPublisher, StoreEventHandler


class Chatroom(NamedTuple):
    users: tuple[str, ...]


class State(NamedTuple):
    chatroom: Chatroom


class StateSubscriber:
    def on_state(self, state: State) -> None:
        print(f"[StoreSubscriber] on_state called with {state}")


class JoinUserAction(pydepot.Action):
    def __init__(self, user: str):
        self.user: str = user


class JoinUserReducer(pydepot.Reducer[JoinUserAction, State]):
    @property
    def action_type(self) -> type[JoinUserAction]:
        return JoinUserAction

    def apply(self, action: JoinUserAction, state: State) -> State:
        return State(chatroom=Chatroom(users=(*state.chatroom.users, action.user)))


class UserJoinedEvent(Event):
    @property
    def name(self) -> str:
        return "user_joined"

    def __init__(self, user: str):
        self.user: str = user


class UserJoinedEventHandler(StoreEventHandler[UserJoinedEvent, State]):
    @property
    def event_type(self) -> type[UserJoinedEvent]:
        return UserJoinedEvent

    def actions(self, event: UserJoinedEvent) -> Sequence[pydepot.Action]:
        return [JoinUserAction(user=event.user)]


def main() -> None:
    store = pydepot.Store(State(chatroom=Chatroom(users=())))
    store.register(JoinUserReducer())

    subscriber = StateSubscriber()
    store.subscribe(subscriber)

    handler = UserJoinedEventHandler(store=store)

    publisher = EventPublisher()
    publisher.subscribe(handler)

    publisher.publish(UserJoinedEvent(user="Alice"))
    publisher.publish(UserJoinedEvent(user="Bob"))


if __name__ == "__main__":
    main()
$ python3 main.py

[StoreSubscriber] on_state called with
    State(chatroom=Chatroom(users=('Alice',)))

[StoreSubscriber] on_state called with
    State(chatroom=Chatroom(users=('Alice', 'Bob')))

Installation

Eventoolkit is available as eventoolkit on PyPI:

pip install eventoolkit

Usage

For detailed quickstart and API reference, visit the Documentation.

License

AGPLv3

Copyright (C) 2023 tombartk 

This program is free software: you can redistribute it and/or modify it under the terms
of the GNU Affero General Public License as published by the Free Software Foundation,
either version 3 of the License, or (at your option) any later version.

This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
See the GNU Affero General Public License for more details.

You should have received a copy of the GNU Affero General Public License along with this program.
If not, see https://www.gnu.org/licenses/.