# Welcome to `tomata` readme!
This is a generic state automata-ish module allowing you to build an typed event-driven stateful flow with minimal setup.
The basic flow is:
1. You define your own emitter class, inheriting from `tomata.Emitter`. There you define `get_state` and `set_state` methods to provide the state storage.
2. You create an instance of emitter
3. You register handlers for the state changes and events with `emitter.on(state_key, handler_type = "event")`
4. In your main loop, you call `emitter.emit(event, identity)` with arbitrary event and identity values (identity is passed to `get_state`, `set_state` and handler, event is passed to handler)
If you want to dive into examples, go to [examples](https://github.com/evtn/tomata/tree/lord/examples) folder.
## Concepts
tomata works around several concepts (in *italic* in text):
### Event
**Event** is an arbitrary value to process. It is passed to all the *handlers*
### State
**State** is some key to denote the current state. *Handlers* are called depending on which state is currently active
### Identity
**Identity** is a key that is used to differentiate between states of different entites. An identity could be, for example, a user id or a string hash.
### Handler
**Handler** is a function called on any event. There are three types of handlers: `enter`, `event` and `leave`.
- `enter` handlers are called when *state* switches **to** the one tied to handler
- `event` handlers are called when *emitter* emits an event. Handlers of this type can change the *state* by returning the next *state* key
- `leave` handlers are called when *state* switches **from** the one tied to handler
### Emitter
**Emitter** is the main object of the module. It consumes *events* and calls *handlers* as a result. *Handlers* can request a *state* change, *emitter* also handles that.
To work with events, you need your own `Emitter` class (and an instance). It manages calling your handlers and changing the state.
A minimal `Emitter` class has to define `get_state` and `set_state`:
```python
from tomata import Emitter
class MyEmitter(Emitter):
def set_state(self, identity, state):
...
def get_state(self, identity):
...
```
Those methods should handle how to store the state (for example, get_state could retrieve the state from database and set_state could write it into the database).
You can also define `__init__` for your emitter class:
```python
...
class MyEmitter(Emitter):
def __init__(self):
default_state = "start"
super().__init__(default_state)
...
...
```
If you love type hints, we've got you covered:
```python
from tomata import Emitter
from typing import Literal
StateKey = Literal["start", "age", "name", "finish"]
Event = str
Identity = int
class MyEmitter(Emitter[Event, StateKey, Identity]):
def __init__(self):
super().__init__("start")
def get_state(self, identity: Identity) -> StateKey:
...
def set_state(self, identity: Identity, state: StateKey) -> None:
...
```
...or, if you are on Python 3.12:
```python
from tomata import Emitter
from typing import Literal
# those types are arbitrary, you can actually use anything as your event / state / identity
type Event = str
type StateKey = Literal["start", "age", "name", "finish"]
type Identity = int
class MyEmitter(Emitter[Event, StateKey, Identity]):
def __init__(self):
super().__init__("start")
def get_state(self, identity: Identity) -> StateKey:
...
def set_state(self, identity: Identity, state: StateKey) -> None:
...
```
## Defining handlers
Let's say, you want to run some code when state `cake_shooting` is active and some event comes by:
```python
em = MyEmitter()
...
@em.on("cake_shooting")
def cake_shooting(event: Event, identity: Identity, state: StateKey):
# event and identity are provided through emit, state is the current state.
if not event:
return "no_cake" # new state
cake = catch_cake(event)
eat_cake(cake)
```
Now, if you want to run some code when state was switched to `cake_shooting`, use `enter` handler:
```python
em = MyEmitter()
...
@em.on.enter("cake_shooting")
def cake_shooting(event: Event, identity: Identity, state: StateKey):
# event and identity are provided through emit, state is the current state.
...
```
Same goes for when state was switched from state, just use `.leave` handler
## Emitting events
After you've set up your emitter and handlers, it's time to emit some events:
```python
emitter = MyEmitter()
from json import load
from typing import TypedDict
class EventData(TypedDict):
data: Event
identity: Identity
with open("events.json") as file:
events: list[EventData] = load(file)
for event in events:
emitter.emit(event["data"], event["identity"])
```
## Default handler
You can set a default (fallback) handler to handle events when no other handler is found:
```python
@em.on.default
def default_handler(event, identity, state):
...
```
At the moment, you can't define a default `enter` and `leave` handlers, because it doesn't really seem useful (but feel free to open an issue if you find a good use-case)
## Async
To use async, replace Emitter with AsyncEmitter:
```python
from tomata import AsyncEmitter
from typing import Literal
type StateKey = Literal["start", "age", "name", "finish"]
type Event = str
type Identity = int
class MyEmitter(AsyncEmitter[Event, StateKey, Identity]):
def __init__(self):
super().__init__("start")
async def get_state(self, identity: Identity) -> StateKey:
...
async def set_state(self, identity: Identity, state: StateKey) -> None:
...
em = MyEmitter()
# AsyncEmitter can call both async and sync handlers
@em.on("funny_pineapple")
async def funny_pineapple(event: Event, identity: Identity, state: StateKey):
...
@em.on("sad_pineapple")
def sad_pineapple(event: Event, identity: Identity, state: StateKey):
...
```
## Advanced behaviour
Let's say, you want to redefine handler storage.
You can easily do that by defining your own `get_handler` and `set_handler` methods.
Let's make an Emitter where state would be a dictionary, and the handler would be called on state["type"]
```python
from tomata import Emitter
from tomata.base import make_handlers_dict
Event = dict[str, str]
State = dict[str, str]
Identity = int
class AdvancedEmitter(Emitter[Event, State, Identity])
handlers: dict[HandlerType, SyncHandlerStore[str, Event, Identity]]
def get_state(self, identity: Identity) -> State:
...
def set_state(self, identity: Identity, state: State) -> None:
...
def get_handler(
self, state: SK, handler_type: HandlerType
) -> SyncHandler[Ev, Id, SK] | None:
store = self.handlers[handler_type]
key = state.get("type", self.default_state)
return store.get(key)
def set_handler(
self,
state: SK,
handler_type: HandlerType,
handler: SyncHandler[Ev, Id, SK],
):
key = state.get("type", self.default_state)
store = self.handlers[handler_type]
store[key] = handler
em = AdvancedEmitter({"type": "start"})
@em.on({"type": "start"})
def start_event(event: Event, identity: Identity, state: State):
new_state = {**state, "type": "ongoing"}
new_state["count"] = new_state.get("count", 0)
return new_state
```
To redefine setting fallback handler logic, define your own `set_default(handler)` and `get_default()` methods.
### AsyncEmitter methods
To extend AsyncEmitter, you have to remember two things:
- Some methods (`[get|set]_state`, `get_handler`, `get_default`, `reroute`) are async. This means you have to define them with `async def` even if you don't plan to use await in them.
*`emit` is obviously async, but if you don't need an async `emit`, take a look at synchronous `Emitter`*
- In `set_handler` and `set_default`, you have to async-ify the handlers. To do that, use `tomata.async_emitter.make_async` function, which takes any handler and makes it async
Otherwise, the extension process should be just as with `Emitter`.
## Reroute
*new in 1.1.0*
Sometimes you receive an event within one state and want to handle it within another instead. Use `*Emitter.reroute()` for this:
```python
from time import time
HOUR = 3600
DAY = 24 * HOUR
NOON = 12 * HOUR
@handler.on("state1")
def handle1(event: Event, identity: Identity, state: State):
if (time() % DAY) > NOON:
return handler.reroute("state2", event, identity)
print("it is morning")
@handler.on("state2")
def handle2(event: Event, identity: Identity, state: State):
print("it is state2 or afternoon")
```
`.reroute` basically does this:
```python
handler.set_state(identity, state)
handler.emit(event, identity)
return None
```
Raw data
{
"_id": null,
"home_page": "https://github.com/evtn/tomata",
"name": "tomata",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.8",
"maintainer_email": null,
"keywords": null,
"author": "Dmitry Gritsenko",
"author_email": "tomata@evtn.me",
"download_url": "https://files.pythonhosted.org/packages/80/f3/d216ac3f7de3cb65b1da71eaee9c9271db6ed23d423c7bcaad715e3b683b/tomata-1.1.0.tar.gz",
"platform": null,
"description": "# Welcome to `tomata` readme!\n\nThis is a generic state automata-ish module allowing you to build an typed event-driven stateful flow with minimal setup.\n\nThe basic flow is:\n\n1. You define your own emitter class, inheriting from `tomata.Emitter`. There you define `get_state` and `set_state` methods to provide the state storage.\n2. You create an instance of emitter\n3. You register handlers for the state changes and events with `emitter.on(state_key, handler_type = \"event\")`\n4. In your main loop, you call `emitter.emit(event, identity)` with arbitrary event and identity values (identity is passed to `get_state`, `set_state` and handler, event is passed to handler)\n\nIf you want to dive into examples, go to [examples](https://github.com/evtn/tomata/tree/lord/examples) folder.\n\n\n## Concepts\n\ntomata works around several concepts (in *italic* in text):\n\n### Event\n\n**Event** is an arbitrary value to process. It is passed to all the *handlers*\n\n### State\n\n**State** is some key to denote the current state. *Handlers* are called depending on which state is currently active\n\n### Identity\n\n**Identity** is a key that is used to differentiate between states of different entites. An identity could be, for example, a user id or a string hash.\n\n### Handler\n\n**Handler** is a function called on any event. There are three types of handlers: `enter`, `event` and `leave`.\n\n- `enter` handlers are called when *state* switches **to** the one tied to handler\n- `event` handlers are called when *emitter* emits an event. Handlers of this type can change the *state* by returning the next *state* key\n- `leave` handlers are called when *state* switches **from** the one tied to handler\n\n### Emitter\n\n**Emitter** is the main object of the module. It consumes *events* and calls *handlers* as a result. *Handlers* can request a *state* change, *emitter* also handles that.\n\nTo work with events, you need your own `Emitter` class (and an instance). It manages calling your handlers and changing the state.\n\nA minimal `Emitter` class has to define `get_state` and `set_state`:\n\n\n```python\nfrom tomata import Emitter\n\nclass MyEmitter(Emitter):\n def set_state(self, identity, state):\n ...\n\n def get_state(self, identity):\n ...\n``` \n\nThose methods should handle how to store the state (for example, get_state could retrieve the state from database and set_state could write it into the database).\n\nYou can also define `__init__` for your emitter class:\n\n```python\n...\nclass MyEmitter(Emitter):\n def __init__(self):\n default_state = \"start\"\n super().__init__(default_state)\n ...\n \n ...\n```\n\nIf you love type hints, we've got you covered:\n\n```python\nfrom tomata import Emitter\nfrom typing import Literal\n\nStateKey = Literal[\"start\", \"age\", \"name\", \"finish\"]\nEvent = str\nIdentity = int\n\nclass MyEmitter(Emitter[Event, StateKey, Identity]):\n def __init__(self):\n super().__init__(\"start\")\n \n def get_state(self, identity: Identity) -> StateKey:\n ...\n \n def set_state(self, identity: Identity, state: StateKey) -> None:\n ...\n\n```\n\n...or, if you are on Python 3.12:\n\n```python\nfrom tomata import Emitter\nfrom typing import Literal\n\n# those types are arbitrary, you can actually use anything as your event / state / identity\ntype Event = str\ntype StateKey = Literal[\"start\", \"age\", \"name\", \"finish\"]\ntype Identity = int\n\nclass MyEmitter(Emitter[Event, StateKey, Identity]):\n def __init__(self):\n super().__init__(\"start\")\n \n def get_state(self, identity: Identity) -> StateKey:\n ...\n \n def set_state(self, identity: Identity, state: StateKey) -> None:\n ...\n\n```\n\n## Defining handlers\n\nLet's say, you want to run some code when state `cake_shooting` is active and some event comes by:\n\n```python\n\nem = MyEmitter()\n\n...\n\n\n@em.on(\"cake_shooting\")\ndef cake_shooting(event: Event, identity: Identity, state: StateKey):\n # event and identity are provided through emit, state is the current state.\n if not event:\n return \"no_cake\" # new state\n \n cake = catch_cake(event)\n eat_cake(cake)\n```\n\nNow, if you want to run some code when state was switched to `cake_shooting`, use `enter` handler:\n\n```python\n\nem = MyEmitter()\n\n...\n\n\n@em.on.enter(\"cake_shooting\")\ndef cake_shooting(event: Event, identity: Identity, state: StateKey):\n # event and identity are provided through emit, state is the current state.\n ...\n```\n\nSame goes for when state was switched from state, just use `.leave` handler\n\n\n## Emitting events\n\nAfter you've set up your emitter and handlers, it's time to emit some events:\n\n```python\nemitter = MyEmitter()\n\nfrom json import load\nfrom typing import TypedDict\n\nclass EventData(TypedDict):\n data: Event\n identity: Identity\n \n\nwith open(\"events.json\") as file:\n events: list[EventData] = load(file)\n\n\nfor event in events:\n emitter.emit(event[\"data\"], event[\"identity\"])\n\n```\n\n## Default handler\n\nYou can set a default (fallback) handler to handle events when no other handler is found:\n\n```python\n\n@em.on.default\ndef default_handler(event, identity, state):\n ...\n\n```\n\nAt the moment, you can't define a default `enter` and `leave` handlers, because it doesn't really seem useful (but feel free to open an issue if you find a good use-case)\n\n## Async \n\nTo use async, replace Emitter with AsyncEmitter:\n\n\n```python\nfrom tomata import AsyncEmitter\nfrom typing import Literal\n\n\ntype StateKey = Literal[\"start\", \"age\", \"name\", \"finish\"]\ntype Event = str\ntype Identity = int\n\n\nclass MyEmitter(AsyncEmitter[Event, StateKey, Identity]):\n def __init__(self):\n super().__init__(\"start\")\n \n async def get_state(self, identity: Identity) -> StateKey:\n ...\n \n async def set_state(self, identity: Identity, state: StateKey) -> None:\n ...\n\n\nem = MyEmitter()\n\n# AsyncEmitter can call both async and sync handlers\n\n@em.on(\"funny_pineapple\")\nasync def funny_pineapple(event: Event, identity: Identity, state: StateKey):\n ...\n\n@em.on(\"sad_pineapple\")\ndef sad_pineapple(event: Event, identity: Identity, state: StateKey):\n ...\n\n```\n\n## Advanced behaviour\n\nLet's say, you want to redefine handler storage.\nYou can easily do that by defining your own `get_handler` and `set_handler` methods.\n\nLet's make an Emitter where state would be a dictionary, and the handler would be called on state[\"type\"]\n\n```python\nfrom tomata import Emitter\nfrom tomata.base import make_handlers_dict\n\nEvent = dict[str, str]\nState = dict[str, str]\nIdentity = int\n\nclass AdvancedEmitter(Emitter[Event, State, Identity])\n handlers: dict[HandlerType, SyncHandlerStore[str, Event, Identity]] \n \n def get_state(self, identity: Identity) -> State:\n ...\n \n def set_state(self, identity: Identity, state: State) -> None:\n ...\n \n def get_handler(\n self, state: SK, handler_type: HandlerType\n ) -> SyncHandler[Ev, Id, SK] | None:\n store = self.handlers[handler_type]\n \n key = state.get(\"type\", self.default_state) \n \n return store.get(key)\n\n def set_handler(\n self,\n state: SK,\n handler_type: HandlerType,\n handler: SyncHandler[Ev, Id, SK],\n ):\n key = state.get(\"type\", self.default_state)\n \n store = self.handlers[handler_type]\n \n store[key] = handler\n\n\nem = AdvancedEmitter({\"type\": \"start\"})\n\n@em.on({\"type\": \"start\"})\ndef start_event(event: Event, identity: Identity, state: State):\n new_state = {**state, \"type\": \"ongoing\"}\n new_state[\"count\"] = new_state.get(\"count\", 0)\n \n return new_state\n```\n\nTo redefine setting fallback handler logic, define your own `set_default(handler)` and `get_default()` methods.\n\n\n### AsyncEmitter methods\n\nTo extend AsyncEmitter, you have to remember two things:\n\n- Some methods (`[get|set]_state`, `get_handler`, `get_default`, `reroute`) are async. This means you have to define them with `async def` even if you don't plan to use await in them. \n *`emit` is obviously async, but if you don't need an async `emit`, take a look at synchronous `Emitter`*\n\n- In `set_handler` and `set_default`, you have to async-ify the handlers. To do that, use `tomata.async_emitter.make_async` function, which takes any handler and makes it async\n\nOtherwise, the extension process should be just as with `Emitter`.\n\n\n## Reroute \n\n*new in 1.1.0*\n\nSometimes you receive an event within one state and want to handle it within another instead. Use `*Emitter.reroute()` for this:\n\n```python\nfrom time import time\n\nHOUR = 3600\nDAY = 24 * HOUR\nNOON = 12 * HOUR\n\n@handler.on(\"state1\")\ndef handle1(event: Event, identity: Identity, state: State):\n if (time() % DAY) > NOON:\n return handler.reroute(\"state2\", event, identity)\n \n print(\"it is morning\")\n\n\n@handler.on(\"state2\")\ndef handle2(event: Event, identity: Identity, state: State):\n print(\"it is state2 or afternoon\")\n\n```\n\n`.reroute` basically does this:\n\n```python\nhandler.set_state(identity, state)\nhandler.emit(event, identity)\nreturn None\n```",
"bugtrack_url": null,
"license": "MIT",
"summary": null,
"version": "1.1.0",
"project_urls": {
"Homepage": "https://github.com/evtn/tomata",
"Repository": "https://github.com/evtn/tomata"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "6e8ebc740152b536e6e1e47e5f2a2f64006cfe5b4f5ce66513120c2b6599befd",
"md5": "883108588599863e3c6656228c139cfd",
"sha256": "06068396a4b99643a1526a6b5e90c9d68acbc0c6e55ae89dd873fce9d0ab79c1"
},
"downloads": -1,
"filename": "tomata-1.1.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "883108588599863e3c6656228c139cfd",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.8",
"size": 7579,
"upload_time": "2024-04-26T03:52:51",
"upload_time_iso_8601": "2024-04-26T03:52:51.370902Z",
"url": "https://files.pythonhosted.org/packages/6e/8e/bc740152b536e6e1e47e5f2a2f64006cfe5b4f5ce66513120c2b6599befd/tomata-1.1.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "80f3d216ac3f7de3cb65b1da71eaee9c9271db6ed23d423c7bcaad715e3b683b",
"md5": "8295434392477dd41f3634117dd575e6",
"sha256": "e071435ce25acd45d1550e5d15d65d57bd435b3ed4d822f60009667ebaa29aca"
},
"downloads": -1,
"filename": "tomata-1.1.0.tar.gz",
"has_sig": false,
"md5_digest": "8295434392477dd41f3634117dd575e6",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.8",
"size": 5989,
"upload_time": "2024-04-26T03:52:52",
"upload_time_iso_8601": "2024-04-26T03:52:52.993064Z",
"url": "https://files.pythonhosted.org/packages/80/f3/d216ac3f7de3cb65b1da71eaee9c9271db6ed23d423c7bcaad715e3b683b/tomata-1.1.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-04-26 03:52:52",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "evtn",
"github_project": "tomata",
"travis_ci": false,
"coveralls": true,
"github_actions": true,
"lcname": "tomata"
}
JSON
