[](https://t.me/aiostep_chat)
# Aiostep - Simple and Flexible State Management
Aiostep is a lightweight and flexible state management tool designed for Telegram bots and similar applications. It allows developers to track user states and manage transitions between them with ease. Whether you're building a multi-step form, handling complex user interactions, or simply need to store temporary user data, Aiostep makes it straightforward.
_We have a vibrant community of developers helping each other in our_ **[Telegram group](https://t.me/aiostep_chat)**. _Join us!_
_Stay tuned for library updates and new releases on our_ **[Telegram Channel](https://t.me/aiostep)**.
---
## Features
- **Simple API**: Intuitive methods for setting, getting, and deleting user states and associated data.
- **Customizable Storage**: Use in-memory storage or integrate with persistent options like Redis or file-based storage.
- **Direct User Interaction**: Easily ask questions and receive user responses directly within handlers, reducing boilerplate code.
- **Extendable**: Designed to integrate with existing frameworks such as aiogram.
---
## How It Works
### User Interaction Methods
Aiostep provides three primary methods for interacting with users and managing multi-step processes:
1. **`wait_for`**:
- Use this method to wait for a specific user response within the current handler.
- Simplifies user interaction by reducing the need for separate handlers.
2. **`register_next_step`**:
- Allows registering the next handler explicitly for a user.
- Useful for chaining steps in a process.
3. **States**:
- Define user states to manage stages in a multi-step workflow.
- States can include optional callbacks for seamless navigation between steps.
---
## Installation
To start using Aiostep, simply install it via pip.
```bash
pip install --upgrade aiostep
```
If you want use `RedisStateStorage`, you should install aiostep with redis support:
```bash
pip install --upgrade aiostep[redis]
```
---
## Usage
### Using `wait_for` and `register_next_step`
**Aiostep offers two primary methods for managing direct user interactions:**
#### 1. `wait_for`:
- This method allows you to wait for a user response directly within the current handler.
- Requires the `Listen` middleware to be set up for intercepting subsequent user messages.
**Example:**
```python
from aiostep import Listen, wait_for
from aiogram import Dispatcher
dp = Dispatcher()
dp.message.outer_middleware(Listen())
@dp.message_handler(commands=["start"])
async def ask_question(message: Message):
await message.reply("Please type something:")
try:
response = await wait_for(message.from_user.id, timeout=25) # timeout is optional
except TimeoutError:
await message.reply("You took too long to answer.")
else:
await message.reply(f"You typed: {response.text}")
```
> [!NOTE]\
> The `timeout` parameter is optional; if not provided, the bot will wait indefinitely for a response.
#### 2. `register_next_step`
- Use this method to explicitly register the next handler for the user's response.
- Also requires the `Listen` middleware for processing follow-up messages.
**Example:**
```python
@dp.message_handler(commands=["start"])
async def ask_question(message: Message):
await aiostep.register_next_step(message.chat.id, handle_answer)
await message.reply("What's your name?")
async def handle_answer(message: Message):
await message.reply(f"Hello, {message.text}!")
```
### Using States
**Aiostep supports managing user states to handle multi-step workflows. Unlike the previous methods, managing states does not require the `Listen` middleware.**
#### 1. Memory State Storage:
- This is an in-memory implementation suitable for temporary state storage.
**Example:**
```python
from aiostep import MemoryStateStorage
state_manager = MemoryStateStorage()
@dp.message_handler(commands=["start"])
async def start_process(message: Message):
state_manager.set_state(
user_id=message.from_user.id,
state="STEP_ONE"
)
await message.reply("State set to STEP_ONE!")
@dp.message_handler(lambda message: state_manager.get_state(message.from_user.id).current_state == "STEP_ONE")
async def handle_step_one(message: Message):
await message.reply("You're in STEP_ONE.")
```
**Returning to Previous State:**
```python
@dp.message_handler(lambda message: message.text == "Back")
async def go_back(message: Message):
step = state_manager.get_state(message.from_user.id)
if step and step.callback:
await step.callback(message)
else:
await message.reply("No previous state found.")
```
> [!NOTE]\
> You should manually use getattr to find and call the back step handler if you use `RedisStateStorage` or `FileStateStorage`, because callbacks are saved as strings (function name)
>```python
>@dp.message_handler(lambda message: message.text == "Back")
>async def go_back(message: Message):
> step = await state_manager.get_state(message.from_user.id)
> if step and step.callback:
> callback = getattr(step.callback)
> await callback(message)
> else:
> await message.reply("No previous state found.")
>```
#### 2. Other Storage Options:
- File-based and Redis storage implementations are also available, providing similar functionality with persistent data storage.
- Simply replace MemoryStateStorage with FileStateStorage or RedisStateStorage when initializing the state manager.
> [!NOTE]\
> Methods in `MemoryStateStorage`, `FileStateStorage` and `RedisStateStorage` are synchronous.
> **If you want use asynchronous versions, use `aiostep.asyncio`:**
> ```python
> from aiostep.asyncio import AsyncMemoryStateStorage
> from aiostep.asyncio import AsyncFileStateStorage
> from aiostep.asyncio import AsyncRedisStateStorage
> ```
#### 3. Timeout States
To set a timeout (expiry) for the state storage, you can use the `ex` argument for both `RedisStateStorage` and `FileStateStorage`.
But for `MemoryStateStorage` you need to pass a TTLCache if you want set timeout.
Here's how you can set it up:
- **For `MemoryStateStorage`** (using `cachebox.TTLCache`):
```python
from aiostep import MemoryStateStorage
from cachebox import TTLCache # cachetools.TTLCache also works
# Create a TTLCache with a timeout of 200 seconds
storage = MemoryStateStorage(TTLCache(0, 200)) # Timeout is 200 seconds
```
- **For `RedisStateStorage`** (using the `ex` argument for expiry time):
```python
from aiostep import RedisStateStorage
from aiostep import FileStateStorage
# Create RedisStateStorage with a timeout of 200 seconds
storage = RedisStateStorage(db=0, ex=200) # Timeout (expiry) is 200 seconds
storage = FileStateStorage("path.txt", ex=200) # Same as RedisStateStorage
```
In both cases, the state will automatically expire after the specified time, and the data will be removed from the storage.
---
### Using Data
#### Setting Data
```python
state_manager.set_data(
user_id=message.from_user.id,
data={"key": "value"}
)
```
#### Getting Data
```python
data = state_manager.get_data(user_id=message.from_user.id)
await message.reply(f"Your data: {data}")
```
---
## Important Notes
1. **Callbacks**:
- Callbacks can be any callable object, such as functions.
- In `FileStateStorage` and `RedisStateStorage` they are stored as strings (e.g. function name).
3. **Storage Flexibility**:
- The memory-based implementation is ideal for development and testing.
- Persistent storage like Redis is recommended for production.
---
## Future Plans
- **Better Library Compatibility**: Enhanced support for other Telegram bot libraries such as `pyTelegramBotAPI` and `python-telegram-bot`, in addition to `aiogram`.
- **Improved Documentation**: Detailed guides and best practices.
---
## License
This project is licensed under the MIT License. See the [LICENSE](LICENSE) file for more details.
---
For more information or to contribute, visit our [GitHub repository](#).
Raw data
{
"_id": null,
"home_page": null,
"name": "aiostep",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.9",
"maintainer_email": null,
"keywords": "aiogram, steps, states, Telegram bot, asynchronous, aiostep",
"author": "Nasrollah Yusefi",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/2f/7d/847f5eafe02bce39a1236845ccc29d7d12ccaaaccc028892fb274c493c90/aiostep-0.3.4.tar.gz",
"platform": null,
"description": "\n\n\n\n\n[](https://t.me/aiostep_chat)\n\n\n# Aiostep - Simple and Flexible State Management\n\nAiostep is a lightweight and flexible state management tool designed for Telegram bots and similar applications. It allows developers to track user states and manage transitions between them with ease. Whether you're building a multi-step form, handling complex user interactions, or simply need to store temporary user data, Aiostep makes it straightforward.\n\n_We have a vibrant community of developers helping each other in our_ **[Telegram group](https://t.me/aiostep_chat)**. _Join us!_\n_Stay tuned for library updates and new releases on our_ **[Telegram Channel](https://t.me/aiostep)**.\n\n---\n\n## Features\n\n- **Simple API**: Intuitive methods for setting, getting, and deleting user states and associated data.\n- **Customizable Storage**: Use in-memory storage or integrate with persistent options like Redis or file-based storage.\n- **Direct User Interaction**: Easily ask questions and receive user responses directly within handlers, reducing boilerplate code.\n- **Extendable**: Designed to integrate with existing frameworks such as aiogram.\n\n---\n\n## How It Works\n\n### User Interaction Methods\n\nAiostep provides three primary methods for interacting with users and managing multi-step processes:\n\n1. **`wait_for`**:\n - Use this method to wait for a specific user response within the current handler.\n - Simplifies user interaction by reducing the need for separate handlers.\n\n2. **`register_next_step`**:\n - Allows registering the next handler explicitly for a user.\n - Useful for chaining steps in a process.\n\n3. **States**:\n - Define user states to manage stages in a multi-step workflow.\n - States can include optional callbacks for seamless navigation between steps.\n\n---\n\n## Installation\n\nTo start using Aiostep, simply install it via pip.\n\n```bash\npip install --upgrade aiostep\n```\n\nIf you want use `RedisStateStorage`, you should install aiostep with redis support:\n```bash\npip install --upgrade aiostep[redis]\n```\n\n---\n\n## Usage\n\n### Using `wait_for` and `register_next_step`\n**Aiostep offers two primary methods for managing direct user interactions:**\n\n#### 1. `wait_for`:\n- This method allows you to wait for a user response directly within the current handler.\n- Requires the `Listen` middleware to be set up for intercepting subsequent user messages.\n\n**Example:**\n```python\nfrom aiostep import Listen, wait_for\nfrom aiogram import Dispatcher\n\ndp = Dispatcher()\ndp.message.outer_middleware(Listen())\n\n@dp.message_handler(commands=[\"start\"])\nasync def ask_question(message: Message):\n await message.reply(\"Please type something:\")\n try:\n response = await wait_for(message.from_user.id, timeout=25) # timeout is optional\n except TimeoutError:\n await message.reply(\"You took too long to answer.\")\n else:\n await message.reply(f\"You typed: {response.text}\")\n```\n> [!NOTE]\\\n> The `timeout` parameter is optional; if not provided, the bot will wait indefinitely for a response.\n\n#### 2. `register_next_step`\n- Use this method to explicitly register the next handler for the user's response.\n- Also requires the `Listen` middleware for processing follow-up messages.\n\n**Example:**\n\n```python\n@dp.message_handler(commands=[\"start\"])\nasync def ask_question(message: Message):\n await aiostep.register_next_step(message.chat.id, handle_answer)\n await message.reply(\"What's your name?\")\n\nasync def handle_answer(message: Message):\n await message.reply(f\"Hello, {message.text}!\")\n```\n\n### Using States\n**Aiostep supports managing user states to handle multi-step workflows. Unlike the previous methods, managing states does not require the `Listen` middleware.**\n\n\n#### 1. Memory State Storage:\n- This is an in-memory implementation suitable for temporary state storage.\n\n**Example:**\n\n```python\nfrom aiostep import MemoryStateStorage\n\nstate_manager = MemoryStateStorage()\n\n@dp.message_handler(commands=[\"start\"])\nasync def start_process(message: Message):\n state_manager.set_state(\n user_id=message.from_user.id,\n state=\"STEP_ONE\"\n )\n await message.reply(\"State set to STEP_ONE!\")\n\n@dp.message_handler(lambda message: state_manager.get_state(message.from_user.id).current_state == \"STEP_ONE\")\nasync def handle_step_one(message: Message):\n await message.reply(\"You're in STEP_ONE.\")\n```\n\n**Returning to Previous State:**\n```python\n@dp.message_handler(lambda message: message.text == \"Back\")\nasync def go_back(message: Message):\n step = state_manager.get_state(message.from_user.id)\n if step and step.callback:\n await step.callback(message)\n else:\n await message.reply(\"No previous state found.\")\n```\n> [!NOTE]\\\n> You should manually use getattr to find and call the back step handler if you use `RedisStateStorage` or `FileStateStorage`, because callbacks are saved as strings (function name)\n>```python\n>@dp.message_handler(lambda message: message.text == \"Back\")\n>async def go_back(message: Message):\n> step = await state_manager.get_state(message.from_user.id)\n> if step and step.callback:\n> callback = getattr(step.callback)\n> await callback(message)\n> else:\n> await message.reply(\"No previous state found.\")\n>```\n\n#### 2. Other Storage Options:\n- File-based and Redis storage implementations are also available, providing similar functionality with persistent data storage.\n- Simply replace MemoryStateStorage with FileStateStorage or RedisStateStorage when initializing the state manager.\n> [!NOTE]\\\n> Methods in `MemoryStateStorage`, `FileStateStorage` and `RedisStateStorage` are synchronous.\n> **If you want use asynchronous versions, use `aiostep.asyncio`:**\n> ```python\n> from aiostep.asyncio import AsyncMemoryStateStorage\n> from aiostep.asyncio import AsyncFileStateStorage\n> from aiostep.asyncio import AsyncRedisStateStorage\n> ```\n\n#### 3. Timeout States\n\nTo set a timeout (expiry) for the state storage, you can use the `ex` argument for both `RedisStateStorage` and `FileStateStorage`.\nBut for `MemoryStateStorage` you need to pass a TTLCache if you want set timeout.\n\nHere's how you can set it up:\n\n- **For `MemoryStateStorage`** (using `cachebox.TTLCache`):\n\n ```python\n from aiostep import MemoryStateStorage\n from cachebox import TTLCache # cachetools.TTLCache also works\n\n # Create a TTLCache with a timeout of 200 seconds\n storage = MemoryStateStorage(TTLCache(0, 200)) # Timeout is 200 seconds\n ```\n\n- **For `RedisStateStorage`** (using the `ex` argument for expiry time):\n\n ```python\n from aiostep import RedisStateStorage\n from aiostep import FileStateStorage\n\n # Create RedisStateStorage with a timeout of 200 seconds\n storage = RedisStateStorage(db=0, ex=200) # Timeout (expiry) is 200 seconds\n storage = FileStateStorage(\"path.txt\", ex=200) # Same as RedisStateStorage\n ```\nIn both cases, the state will automatically expire after the specified time, and the data will be removed from the storage.\n\n---\n\n### Using Data\n\n#### Setting Data\n\n```python\nstate_manager.set_data(\n user_id=message.from_user.id,\n data={\"key\": \"value\"}\n)\n```\n\n#### Getting Data\n\n```python\ndata = state_manager.get_data(user_id=message.from_user.id)\nawait message.reply(f\"Your data: {data}\")\n```\n\n---\n\n## Important Notes\n\n1. **Callbacks**:\n - Callbacks can be any callable object, such as functions.\n - In `FileStateStorage` and `RedisStateStorage` they are stored as strings (e.g. function name).\n\n3. **Storage Flexibility**:\n - The memory-based implementation is ideal for development and testing.\n - Persistent storage like Redis is recommended for production.\n\n---\n\n## Future Plans\n\n- **Better Library Compatibility**: Enhanced support for other Telegram bot libraries such as `pyTelegramBotAPI` and `python-telegram-bot`, in addition to `aiogram`.\n- **Improved Documentation**: Detailed guides and best practices.\n\n---\n\n## License\n\nThis project is licensed under the MIT License. See the [LICENSE](LICENSE) file for more details.\n\n---\n\nFor more information or to contribute, visit our [GitHub repository](#).\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A Python library to handle steps in aiogram framework.",
"version": "0.3.4",
"project_urls": {
"Homepage": "https://github.com/NasrollahYusefi/aiostep"
},
"split_keywords": [
"aiogram",
" steps",
" states",
" telegram bot",
" asynchronous",
" aiostep"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "ce5e7b6f0cf22b41a73e7fcc572573cf33bf564568c8b7f64241ccfff724d9f6",
"md5": "c91032fd0bdda6a0c75974d847a50273",
"sha256": "3f2f12e092a4431fcd70e51c3118b9c65412f972d7492cd473d706009b23318d"
},
"downloads": -1,
"filename": "aiostep-0.3.4-py3-none-any.whl",
"has_sig": false,
"md5_digest": "c91032fd0bdda6a0c75974d847a50273",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.9",
"size": 47330,
"upload_time": "2025-01-26T17:29:23",
"upload_time_iso_8601": "2025-01-26T17:29:23.652680Z",
"url": "https://files.pythonhosted.org/packages/ce/5e/7b6f0cf22b41a73e7fcc572573cf33bf564568c8b7f64241ccfff724d9f6/aiostep-0.3.4-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "2f7d847f5eafe02bce39a1236845ccc29d7d12ccaaaccc028892fb274c493c90",
"md5": "a8b129b4fc57fbee317aebf2a2682d00",
"sha256": "cfb74c8639f567dc1066783d1c02fa8969f97ee330f716e29168d60f835bdefb"
},
"downloads": -1,
"filename": "aiostep-0.3.4.tar.gz",
"has_sig": false,
"md5_digest": "a8b129b4fc57fbee317aebf2a2682d00",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.9",
"size": 29334,
"upload_time": "2025-01-26T17:29:26",
"upload_time_iso_8601": "2025-01-26T17:29:26.922166Z",
"url": "https://files.pythonhosted.org/packages/2f/7d/847f5eafe02bce39a1236845ccc29d7d12ccaaaccc028892fb274c493c90/aiostep-0.3.4.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-01-26 17:29:26",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "NasrollahYusefi",
"github_project": "aiostep",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "aiostep"
}
JSON
