Merge pull request #28 from Aidenable/dev

v1.3.6: Conversations

Author: Aidenable

docs/source/conf.py

@@ -14,7 +14,7 @@
 project = "Raito"
 copyright = "2025, Aiden"
 author = "Aiden"
-release = "1.3.4"
+release = "1.3.6"
 
 # -- General configuration ---------------------------------------------------
 # https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration

docs/source/plugins/conversations.rst

@@ -0,0 +1,51 @@
+💬 Conversations
+================
+
+Building multi-step dialogs in Telegram bots is often clunky.
+
+Raito provides a lightweight way to wait for the **next user message** in a clean, linear style.
+
+--------
+
+Example
+-------
+
+.. code-block:: python
+
+    from aiogram import F, Router, filters
+    from aiogram.fsm.context import FSMContext
+    from aiogram.types import Message
+
+    from raito import Raito
+
+    router = Router(name="mute")
+
+
+    @router.message(filters.Command("mute"))
+    async def mute(message: Message, raito: Raito, state: FSMContext) -> None:
+        await message.answer("Enter username:")
+        user = await raito.wait_for(state, F.text.regexp(r"@[\w]+"))
+
+        await message.answer("Enter duration (in minutes):")
+        duration = await raito.wait_for(state, F.text.isdigit())
+
+        while not duration.number or duration.number < 0:
+            await message.answer("⚠️ Duration cannot be negative")
+            duration = await duration.retry()
+
+        await message.answer(f"✅ {user.text} will be muted for {duration.number} minutes")
+
+--------
+
+How it works
+------------
+
+- Each ``wait_for`` call registers a pending conversation in Raito’s internal registry
+- A conversation entry stores:
+
+  - A ``Future`` (to resume your handler later)
+  - The filters to check incoming messages
+- When a message arrives:
+
+  - Raito checks for an active conversation bound to that ``FSMContext.key``
+  - If filters match, the future is completed and returned to your handler

docs/source/plugins/index.rst

@@ -12,3 +12,4 @@ Plugins
     throttling
     pagination
     lifespan
+    conversations

examples/12-conversations/__main__.py

@@ -0,0 +1,24 @@
+import asyncio
+from pathlib import Path
+
+from aiogram import Bot, Dispatcher
+
+from raito import Raito
+
+TOKEN = "TOKEN"
+HANDLERS_DIR = Path(__file__).parent / "handlers"
+DEBUG = False
+
+bot = Bot(token=TOKEN)
+dispatcher = Dispatcher()
+raito = Raito(dispatcher, HANDLERS_DIR, developers=[], locales=["en"], production=not DEBUG)
+raito.init_logging()
+
+
+async def main() -> None:
+    await raito.setup()
+    await dispatcher.start_polling(bot)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/12-conversations/handlers/mute.py

@@ -0,0 +1,23 @@
+from aiogram import F, Router, filters
+from aiogram.fsm.context import FSMContext
+from aiogram.types import Message
+
+from raito import Raito
+from raito.plugins.roles import ADMINISTRATOR, DEVELOPER, MODERATOR
+
+router = Router(name="mute")
+
+
+@router.message(filters.Command("mute"), DEVELOPER | ADMINISTRATOR | MODERATOR)
+async def mute(message: Message, raito: Raito, state: FSMContext) -> None:
+    await message.answer("Enter username:")
+    user = await raito.wait_for(state, F.text.regexp(r"@[\w]+"))
+
+    await message.answer("Enter duration (in minutes):")
+    duration = await raito.wait_for(state, F.text.isdigit())
+
+    while not duration.number or duration.number < 0:
+        await message.answer("⚠️ Duration cannot be negative")
+        duration = await duration.retry()
+
+    await message.answer(f"✅ {user.text} will be muted for {duration.number} minutes")

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "raito"
-version = "1.3.5"
+version = "1.3.6"
 description = "REPL, hot-reload, keyboards, pagination, and internal dev tools — all in one. That's Raito."
 authors = [{ name = "Aiden", email = "aidenthedev@gmail.com" }]
 license = "MIT"

raito/core/raito.py

@@ -6,11 +6,19 @@
 from typing import TYPE_CHECKING
 
 from aiogram.dispatcher.event.event import EventObserver
+from aiogram.dispatcher.event.handler import CallbackType
+from aiogram.fsm.context import FSMContext
 from aiogram.fsm.storage.memory import MemoryStorage
 
 from raito.plugins.album.middleware import AlbumMiddleware
 from raito.plugins.commands.middleware import CommandMiddleware
 from raito.plugins.commands.registration import register_bot_commands
+from raito.plugins.conversations import (
+    ConversationMiddleware,
+    ConversationRegistry,
+    Waiter,
+    wait_for,
+)
 from raito.plugins.pagination import PaginationMode, PaginatorMiddleware, get_paginator
 from raito.plugins.roles import (
     BaseRoleProvider,
@@ -99,6 +107,7 @@ def __init__(
         )
 
         self.command_parameters_error = EventObserver()
+        self.registry = ConversationRegistry()
 
     async def setup(self) -> None:
         """Set up the Raito by loading routers and starting watchdog.
@@ -122,6 +131,7 @@ async def setup(self) -> None:
         self.dispatcher.callback_query.middleware(PaginatorMiddleware("raito__is_pagination"))
         self.dispatcher.message.middleware(CommandMiddleware())
         self.dispatcher.message.middleware(AlbumMiddleware())
+        self.dispatcher.message.outer_middleware(ConversationMiddleware(self.registry))
 
         await self.router_manager.load_routers(self.routers_dir)
         await self.router_manager.load_routers(ROOT_DIR / "handlers")
@@ -231,3 +241,18 @@ def init_logging(self, *mute_loggers: str) -> None:
         root_logger.handlers.clear()
         root_logger.addHandler(handler)
         root_logger.setLevel(logging.DEBUG if not self.production else logging.INFO)
+
+    async def wait_for(self, context: FSMContext, *filters: CallbackType) -> Waiter:
+        """Wait for the next message from user that matches given filters.
+
+        This function sets special state ``raito__conversation`` in FSM and
+        suspends coroutine execution until user sends a message that passes
+        all provided filters. Result is wrapped into :class:`Waiter`.
+
+        :param context: FSM context for current chat
+        :param filters: Sequence of aiogram filters
+        :return: Conversation result with text, parsed number and original message
+        :raises RuntimeError: If handler object not found during filter execution
+        :raises asyncio.CancelledError: If conversation was cancelled
+        """
+        return await wait_for(self, context, *filters)

raito/plugins/conversations/__init__.py

@@ -0,0 +1,10 @@
+from .middleware import ConversationMiddleware
+from .registry import ConversationRegistry
+from .waiter import Waiter, wait_for
+
+__all__ = (
+    "ConversationMiddleware",
+    "ConversationRegistry",
+    "Waiter",
+    "wait_for",
+)

raito/plugins/conversations/middleware.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+from typing import TYPE_CHECKING, Any, TypeVar
+
+from aiogram.dispatcher.middlewares.base import BaseMiddleware
+from aiogram.fsm.context import FSMContext
+from aiogram.types import Message
+from typing_extensions import override
+
+from raito.utils.helpers.filters import call_filters
+
+from .registry import ConversationRegistry
+
+if TYPE_CHECKING:
+    from aiogram.types import TelegramObject
+
+R = TypeVar("R")
+
+
+__all__ = ("ConversationMiddleware",)
+
+
+class ConversationMiddleware(BaseMiddleware):
+    """Middleware for conversation handling."""
+
+    def __init__(self, registry: ConversationRegistry) -> None:
+        """Initialize ConversationMiddleware.
+
+        :param registry: conversation registry
+        """
+        self.registry = registry
+
+    @override
+    async def __call__(
+        self,
+        handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[R]],
+        event: TelegramObject,
+        data: dict[str, Any],
+    ) -> R | None:
+        """Process message with conversation support.
+
+        :param handler: Next handler in the middleware chain
+        :type handler: Callable
+        :param event: Telegram event (Message)
+        :type event: TelegramObject
+        :param data: Additional data passed through the middleware chain
+        :type data: dict[str, Any]
+        :return: Handler result if not throttled, None if throttled
+        """
+        if not isinstance(event, Message):
+            return await handler(event, data)
+
+        context: FSMContext | None = data.get("state")
+        if context is not None:
+            state = await context.get_state()
+            filters = self.registry.get_filters(context.key)
+
+            if state is None or filters is None:
+                return await handler(event, data)
+
+            check = await call_filters(event, data, *filters)
+            if not check:
+                return await handler(event, data)
+
+            self.registry.resolve(context.key, event)
+
+        return await handler(event, data)

raito/plugins/conversations/registry.py

@@ -0,0 +1,78 @@
+import asyncio
+from collections.abc import Sequence
+
+from aiogram.dispatcher.event.handler import CallbackType
+from aiogram.fsm.storage.base import StorageKey
+from aiogram.types import Message
+from typing_extensions import NamedTuple
+
+__all__ = ("ConversationRegistry",)
+
+
+class ConversationData(NamedTuple):
+    """Container for an active conversation.
+
+    Stores the Future object awaiting a message and the filters to apply.
+
+    :param future: asyncio.Future that will hold the incoming Message
+    :param filters: Sequence of CallbackType filters to validate the message
+    """
+
+    future: asyncio.Future[Message]
+    filters: Sequence[CallbackType]
+
+
+class ConversationRegistry:
+    """Registry for managing active conversations with users.
+
+    This class allows setting up a "wait for message" scenario where
+    a handler can pause and wait for a specific message from a user,
+    optionally filtered by aiogram filters.
+    """
+
+    STATE = "raito__conversation"
+
+    def __init__(self) -> None:
+        """Initialize the conversation registry."""
+        self._conversations: dict[StorageKey, ConversationData] = {}
+
+    def listen(self, key: StorageKey, *filters: CallbackType) -> asyncio.Future[Message]:
+        """Start listening for a message with a specific StorageKey.
+
+        :param key: StorageKey identifying the conversation (user/chat/bot)
+        :param filters: Optional filters to apply when the message arrives
+        :return: Future that will resolve with the Message when received
+        """
+        future = asyncio.get_running_loop().create_future()
+        self._conversations[key] = ConversationData(future, filters)
+        return future
+
+    def get_filters(self, key: StorageKey) -> Sequence[CallbackType] | None:
+        """Get the filters associated with an active conversation.
+
+        :param key: StorageKey identifying the conversation
+        :return: Sequence of CallbackType filters or None if no conversation exists
+        """
+        data = self._conversations.get(key)
+        return data.filters if data else None
+
+    def resolve(self, key: StorageKey, message: Message) -> None:
+        """Complete the conversation with a received message.
+
+        :param key: StorageKey identifying the conversation
+        :param message: Message object that satisfies the filters
+        """
+        data = self._conversations.pop(key, None)
+        if data and not data.future.done():
+            data.future.set_result(message)
+
+    def cancel(self, key: StorageKey) -> None:
+        """Cancel an active conversation.
+
+        Cancels the Future and removes the conversation from the registry.
+
+        :param key: StorageKey identifying the conversation
+        """
+        data = self._conversations.pop(key, None)
+        if data and not data.future.done():
+            data.future.cancel()