Skip to main content

Client-side toolkit for abstract events.

Project description

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/.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

eventoolkit-0.1.0.tar.gz (24.3 kB view hashes)

Uploaded Source

Built Distribution

eventoolkit-0.1.0-py3-none-any.whl (19.8 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page