| Name | aiostate JSON |
| Version |
0.1.0
JSON |
| download |
| home_page | None |
| Summary | A flexible, async-friendly state machine implementation for Python with thread-safe operations and decorator-based transitions. |
| upload_time | 2025-07-13 19:08:00 |
| maintainer | None |
| docs_url | None |
| author | BoChen SHEN |
| requires_python | <3.13,>=3.11 |
| license | MIT License
Copyright (c) 2025 BoChen SHEN
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE. |
| keywords |
asyncio
state-machine
fsm
python
async
state
|
| VCS |
|
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# aiostate
A flexible, async-friendly state machine implementation for Python with thread-safe operations and decorator-based
transitions.
## Features
- **Thread-safe**: Uses asyncio locks for concurrent access
- **Decorator-based**: Clean, intuitive API for defining transitions
- **Guard conditions**: Conditional transitions with custom logic
- **Entry/Exit handlers**: Execute code when entering or leaving states
- **Wildcard transitions**: Define transitions from any state
- **Type-safe**: Full typing support with generics
- **Flexible**: Works with any hashable type for states and events
## Installation
```bash
pip install aiostate
```
Or using Poetry:
```bash
poetry add aiostate
```
## Quick Start
```python
import asyncio
from aiostate import AsyncStateMachine
# Create a state machine for a simple traffic light
fsm = AsyncStateMachine('red')
@fsm.transition('red', 'timer', 'green')
async def red_to_green():
print("Light turns green")
@fsm.transition('green', 'timer', 'yellow')
async def green_to_yellow():
print("Light turns yellow")
@fsm.transition('yellow', 'timer', 'red')
async def yellow_to_red():
print("Light turns red")
async def main():
print(f"Current state: {fsm.state}") # red
await fsm.trigger('timer') # red -> green
print(f"Current state: {fsm.state}") # green
await fsm.trigger('timer') # green -> yellow
print(f"Current state: {fsm.state}") # yellow
await fsm.trigger('timer') # yellow -> red
print(f"Current state: {fsm.state}") # red
asyncio.run(main())
```
## Advanced Usage
### Entry and Exit Handlers
```python
fsm = AsyncStateMachine('idle')
@fsm.on_enter('running')
async def on_enter_running():
print("System is now running")
# Initialize resources, start monitoring, etc.
@fsm.on_exit('running')
async def on_exit_running():
print("System is stopping")
# Cleanup resources, save state, etc.
@fsm.transition('idle', 'start', 'running')
async def start_system():
print("Starting system...")
# Perform startup logic
```
### Guard Conditions
```python
fsm = AsyncStateMachine('locked')
def has_valid_key(key):
return key == "secret123"
@fsm.transition('locked', 'unlock', 'unlocked', guard=has_valid_key)
async def unlock_door(key):
print(f"Door unlocked with key: {key}")
# Usage
success = await fsm.trigger('unlock', key="wrong_key")
print(success) # False - guard condition failed
success = await fsm.trigger('unlock', key="secret123")
print(success) # True - transition successful
```
### Multiple Source States
```python
fsm = AsyncStateMachine('idle')
# Transition from either 'running' or 'paused' to 'stopped'
@fsm.transition({'running', 'paused'}, 'stop', 'stopped')
async def stop_process():
print("Process stopped")
# Wildcard transition - from any state to 'error'
@fsm.transition('*', 'error', 'error')
async def handle_error():
print("Error occurred, transitioning to error state")
```
### Async Guard Conditions
```python
async def async_guard(user_id):
# Simulate async database check
await asyncio.sleep(0.1)
return user_id in ['admin', 'user123']
@fsm.transition('pending', 'approve', 'approved', guard=async_guard)
async def approve_request(user_id):
print(f"Request approved by {user_id}")
```
## API Reference
### AsyncStateMachine
#### Constructor
```python
AsyncStateMachine(initial_state: T)
```
Creates a new state machine with the specified initial state.
#### Properties
- `state: T` - Current state (read-only)
- `all_states: Set[T]` - All registered states (read-only)
#### Methods
- `is_state(state: T) -> bool` - Check if currently in specified state
- `can_trigger(evt: T) -> bool` - Check if event can be triggered
- `add_state(state: T) -> None` - Add state without transitions
- `get_valid_events() -> Set[T]` - Get valid events for current state
- `get_transition_graph() -> Dict[T, Dict[T, T]]` - Get complete transition graph
- `trigger(evt: T, **kwargs) -> bool` - Trigger an event
#### Decorators
- `@transition(from_states, evt, to_state, guard=None)` - Define a transition
- `@on_enter(state)` - Register enter handler
- `@on_exit(state)` - Register exit handler
## Error Handling
The library raises `StateTransitionError` when:
- No transition is defined for the current state and event
- A guard condition fails during execution
- An exit handler fails
- A transition action fails
```python
from aiostate import StateTransitionError
try:
await fsm.trigger('invalid_event')
except StateTransitionError as e:
print(f"Transition failed: {e}")
```
## License
This project is licensed under the MIT License.
## Development
### Setup
```bash
poetry install
```
### Running Tests
```bash
poetry run pytest
```
### Building
```bash
poetry build
```
Raw data
{
"_id": null,
"home_page": null,
"name": "aiostate",
"maintainer": null,
"docs_url": null,
"requires_python": "<3.13,>=3.11",
"maintainer_email": null,
"keywords": "asyncio, state-machine, fsm, python, async, state",
"author": "BoChen SHEN",
"author_email": "6goddddddd@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/af/32/bd9917f1b5a741942179de60e880d129765d68271e0502130b51fdb08254/aiostate-0.1.0.tar.gz",
"platform": null,
"description": "# aiostate\n\nA flexible, async-friendly state machine implementation for Python with thread-safe operations and decorator-based\ntransitions.\n\n## Features\n\n- **Thread-safe**: Uses asyncio locks for concurrent access\n- **Decorator-based**: Clean, intuitive API for defining transitions\n- **Guard conditions**: Conditional transitions with custom logic\n- **Entry/Exit handlers**: Execute code when entering or leaving states\n- **Wildcard transitions**: Define transitions from any state\n- **Type-safe**: Full typing support with generics\n- **Flexible**: Works with any hashable type for states and events\n\n## Installation\n\n```bash\npip install aiostate\n```\n\nOr using Poetry:\n\n```bash\npoetry add aiostate\n```\n\n## Quick Start\n\n```python\nimport asyncio\nfrom aiostate import AsyncStateMachine\n\n# Create a state machine for a simple traffic light\nfsm = AsyncStateMachine('red')\n\n\n@fsm.transition('red', 'timer', 'green')\nasync def red_to_green():\n print(\"Light turns green\")\n\n\n@fsm.transition('green', 'timer', 'yellow')\nasync def green_to_yellow():\n print(\"Light turns yellow\")\n\n\n@fsm.transition('yellow', 'timer', 'red')\nasync def yellow_to_red():\n print(\"Light turns red\")\n\n\nasync def main():\n print(f\"Current state: {fsm.state}\") # red\n\n await fsm.trigger('timer') # red -> green\n print(f\"Current state: {fsm.state}\") # green\n\n await fsm.trigger('timer') # green -> yellow\n print(f\"Current state: {fsm.state}\") # yellow\n\n await fsm.trigger('timer') # yellow -> red\n print(f\"Current state: {fsm.state}\") # red\n\n\nasyncio.run(main())\n```\n\n## Advanced Usage\n\n### Entry and Exit Handlers\n\n```python\nfsm = AsyncStateMachine('idle')\n\n\n@fsm.on_enter('running')\nasync def on_enter_running():\n print(\"System is now running\")\n # Initialize resources, start monitoring, etc.\n\n\n@fsm.on_exit('running')\nasync def on_exit_running():\n print(\"System is stopping\")\n # Cleanup resources, save state, etc.\n\n\n@fsm.transition('idle', 'start', 'running')\nasync def start_system():\n print(\"Starting system...\")\n # Perform startup logic\n```\n\n### Guard Conditions\n\n```python\nfsm = AsyncStateMachine('locked')\n\n\ndef has_valid_key(key):\n return key == \"secret123\"\n\n\n@fsm.transition('locked', 'unlock', 'unlocked', guard=has_valid_key)\nasync def unlock_door(key):\n print(f\"Door unlocked with key: {key}\")\n\n\n# Usage\nsuccess = await fsm.trigger('unlock', key=\"wrong_key\")\nprint(success) # False - guard condition failed\n\nsuccess = await fsm.trigger('unlock', key=\"secret123\")\nprint(success) # True - transition successful\n```\n\n### Multiple Source States\n\n```python\nfsm = AsyncStateMachine('idle')\n\n\n# Transition from either 'running' or 'paused' to 'stopped'\n@fsm.transition({'running', 'paused'}, 'stop', 'stopped')\nasync def stop_process():\n print(\"Process stopped\")\n\n\n# Wildcard transition - from any state to 'error'\n@fsm.transition('*', 'error', 'error')\nasync def handle_error():\n print(\"Error occurred, transitioning to error state\")\n```\n\n### Async Guard Conditions\n\n```python\nasync def async_guard(user_id):\n # Simulate async database check\n await asyncio.sleep(0.1)\n return user_id in ['admin', 'user123']\n\n\n@fsm.transition('pending', 'approve', 'approved', guard=async_guard)\nasync def approve_request(user_id):\n print(f\"Request approved by {user_id}\")\n```\n\n## API Reference\n\n### AsyncStateMachine\n\n#### Constructor\n\n```python\nAsyncStateMachine(initial_state: T)\n```\n\nCreates a new state machine with the specified initial state.\n\n#### Properties\n\n- `state: T` - Current state (read-only)\n- `all_states: Set[T]` - All registered states (read-only)\n\n#### Methods\n\n- `is_state(state: T) -> bool` - Check if currently in specified state\n- `can_trigger(evt: T) -> bool` - Check if event can be triggered\n- `add_state(state: T) -> None` - Add state without transitions\n- `get_valid_events() -> Set[T]` - Get valid events for current state\n- `get_transition_graph() -> Dict[T, Dict[T, T]]` - Get complete transition graph\n- `trigger(evt: T, **kwargs) -> bool` - Trigger an event\n\n#### Decorators\n\n- `@transition(from_states, evt, to_state, guard=None)` - Define a transition\n- `@on_enter(state)` - Register enter handler\n- `@on_exit(state)` - Register exit handler\n\n## Error Handling\n\nThe library raises `StateTransitionError` when:\n\n- No transition is defined for the current state and event\n- A guard condition fails during execution\n- An exit handler fails\n- A transition action fails\n\n```python\nfrom aiostate import StateTransitionError\n\ntry:\n await fsm.trigger('invalid_event')\nexcept StateTransitionError as e:\n print(f\"Transition failed: {e}\")\n```\n\n## License\n\nThis project is licensed under the MIT License.\n\n## Development\n\n### Setup\n\n```bash\npoetry install\n```\n\n### Running Tests\n\n```bash\npoetry run pytest\n```\n\n### Building\n\n```bash\npoetry build\n```",
"bugtrack_url": null,
"license": "MIT License\n\nCopyright (c) 2025 BoChen SHEN\n\nPermission is hereby granted, free of charge, to any person obtaining a copy\nof this software and associated documentation files (the \"Software\"), to deal\nin the Software without restriction, including without limitation the rights\nto use, copy, modify, merge, publish, distribute, sublicense, and/or sell\ncopies of the Software, and to permit persons to whom the Software is\nfurnished to do so, subject to the following conditions:\n\nThe above copyright notice and this permission notice shall be included in all\ncopies or substantial portions of the Software.\n\nTHE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR\nIMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,\nFITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE\nAUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER\nLIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,\nOUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE\nSOFTWARE.",
"summary": "A flexible, async-friendly state machine implementation for Python with thread-safe operations and decorator-based transitions.",
"version": "0.1.0",
"project_urls": null,
"split_keywords": [
"asyncio",
" state-machine",
" fsm",
" python",
" async",
" state"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "e76278fdd414e0a839b30131b83ffcde4e61c9f0c89f2a0e57e464dfd724af32",
"md5": "9b5f22565ae042b09398d07a4fed5fe5",
"sha256": "d8f9295610d84ed42225a0ab92535eda1d28f259a383b4e7af43800139e019f3"
},
"downloads": -1,
"filename": "aiostate-0.1.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "9b5f22565ae042b09398d07a4fed5fe5",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<3.13,>=3.11",
"size": 8364,
"upload_time": "2025-07-13T19:07:59",
"upload_time_iso_8601": "2025-07-13T19:07:59.442407Z",
"url": "https://files.pythonhosted.org/packages/e7/62/78fdd414e0a839b30131b83ffcde4e61c9f0c89f2a0e57e464dfd724af32/aiostate-0.1.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "af32bd9917f1b5a741942179de60e880d129765d68271e0502130b51fdb08254",
"md5": "29fc51036dd7f5a2787153a084f584cd",
"sha256": "67efa35fef7d162704a32599b7a4f42f4156727d73b85c96d64be21ac66e21e6"
},
"downloads": -1,
"filename": "aiostate-0.1.0.tar.gz",
"has_sig": false,
"md5_digest": "29fc51036dd7f5a2787153a084f584cd",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<3.13,>=3.11",
"size": 6360,
"upload_time": "2025-07-13T19:08:00",
"upload_time_iso_8601": "2025-07-13T19:08:00.686947Z",
"url": "https://files.pythonhosted.org/packages/af/32/bd9917f1b5a741942179de60e880d129765d68271e0502130b51fdb08254/aiostate-0.1.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-13 19:08:00",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "aiostate"
}
