Refactor bot management and introduce caching mechanism

.github/workflows/tests.yml

@@ -35,5 +35,8 @@ jobs:
       - name: Run Ruff
         run: uv run ruff check --output-format=github .
 
+      - name: Run Ty (type checker)
+        run: uv run ty check --output-format=github .
+
       - name: Run pytest
         run: uv run pytest

README.md

@@ -157,7 +157,7 @@ Allows you to serve multiple Telegram bots in a single application. Useful if yo
 
 - Allows serving multiple bots via a single endpoint
 - Uses the bot token for request routing
-- Requires dispatcher, web_adapter, routing, bot_settings (optional), webhook_config (optional), and security (optional)
+- Requires dispatcher, web_adapter, routing, bot_config (optional), webhook_config (optional), and security (optional)
 
 **Example:**
 
@@ -198,7 +198,16 @@ engine = TokenEngine(
 
 #### Custom Engines
 
-You can create your own engine by inheriting from the base engine class (`BaseEngine`). This allows you to implement custom logic for webhook processing, routing, or bot management.
+You can create your own engine by inheriting from `WebhookEngine`. This allows you to implement custom logic for webhook processing, routing, or bot management.
+
+### Request processing
+
+`WebhookEngine` handles incoming updates in this order:
+
+1. Extract token from request (`_get_bot_token_for_request`)
+2. Run security checks for the token (`Security.verify(token, bound_request)`)
+3. Resolve bot (`_get_bot_by_token`)
+4. Pass update to aiogram dispatcher
 
 ---
 
@@ -237,9 +246,9 @@ routing = StaticRouting(url="https://example.com/webhook")
 
 ### TokenRouting (Multi-bot, Abstract)
 Base class for token-based routing strategies. Used with **TokenEngine** to serve multiple bots.
-- Requires a URL template with a parameter placeholder (e.g. `{bot_token}`)
+- Defines the token parameter name (default: `bot_token`)
 - Extracts bot token from incoming requests
-- Automatically formats webhook URL using the bot token
+- Automatically builds webhook URL using the bot token
 
 ### PathRouting (Multi-bot)
 Extracts bot token from the URL path parameter.
@@ -277,7 +286,7 @@ routing = QueryRouting(url="https://example.com/webhook?other=value")
 ```
 
 ### Custom Routing
-You can implement your own routing by inheriting from `BaseRouting` or `TokenRouting` and implementing the `webhook_point()` method (and `extract_token()` if using token-based routing).
+You can implement your own routing by inheriting from `BaseRouting` or `TokenRouting` and implementing the `webhook_url()` method (and `resolve_token()` if using token-based routing).
 
 See [routing examples](/src/aiogram_webhook/routing) for implementation details.
 

src/aiogram_webhook/adapters/fastapi/adapter.py

@@ -4,14 +4,14 @@
 from fastapi.responses import JSONResponse
 
 from aiogram_webhook.adapters.base_adapter import BoundRequest, WebAdapter
-from aiogram_webhook.adapters.fastapi.mapping import FastAPIHeadersMapping, FastAPIQueryMapping
+from aiogram_webhook.adapters.fastapi.mapping import FastApiHeadersMapping, FastApiQueryMapping
 
 
-class FastAPIBoundRequest(BoundRequest[Request]):
+class FastApiBoundRequest(BoundRequest[Request]):
     def __init__(self, request: Request):
         super().__init__(request)
-        self._headers = FastAPIHeadersMapping(self.request.headers)
-        self._query_params = FastAPIQueryMapping(self.request.query_params)
+        self._headers = FastApiHeadersMapping(self.request.headers)
+        self._query_params = FastApiQueryMapping(self.request.query_params)
 
     async def json(self) -> dict[str, Any]:
         return await self.request.json()
@@ -23,11 +23,11 @@ def client_ip(self):
         return None
 
     @property
-    def headers(self) -> FastAPIHeadersMapping:
+    def headers(self) -> FastApiHeadersMapping:
         return self._headers
 
     @property
-    def query_params(self) -> FastAPIQueryMapping:
+    def query_params(self) -> FastApiQueryMapping:
         return self._query_params
 
     @property
@@ -36,8 +36,8 @@ def path_params(self):
 
 
 class FastApiWebAdapter(WebAdapter):
-    def bind(self, request: Request) -> FastAPIBoundRequest:
-        return FastAPIBoundRequest(request=request)
+    def bind(self, request: Request) -> FastApiBoundRequest:
+        return FastApiBoundRequest(request=request)
 
     def register(self, app: FastAPI, path, handler, on_startup=None, on_shutdown=None) -> None:  # noqa: ARG002
         async def endpoint(request: Request):

src/aiogram_webhook/adapters/fastapi/mapping.py

@@ -5,11 +5,11 @@
 from aiogram_webhook.adapters.base_mapping import MappingABC
 
 
-class FastAPIHeadersMapping(MappingABC[Headers]):
+class FastApiHeadersMapping(MappingABC[Headers]):
     def getlist(self, name: str) -> list[Any]:
         return self._mapping.getlist(name)
 
 
-class FastAPIQueryMapping(MappingABC[QueryParams]):
+class FastApiQueryMapping(MappingABC[QueryParams]):
     def getlist(self, name: str) -> list[Any]:
         return self._mapping.getlist(name)

src/aiogram_webhook/config/bot.py

@@ -4,7 +4,7 @@
 from aiogram.client.session.base import BaseSession
 
 
-@dataclass
+@dataclass(slots=True)
 class BotConfig:
     session: BaseSession | None = None
     """HTTP Client session (For example AiohttpSession). If not specified it will be automatically created."""

src/aiogram_webhook/engines/base.py

@@ -1,10 +1,12 @@
 from __future__ import annotations
 
 import asyncio
+import warnings
 from abc import ABC, abstractmethod
 from typing import TYPE_CHECKING, Any
 
 from aiogram.methods import TelegramMethod
+from aiogram.utils.token import TokenValidationError
 
 from aiogram_webhook.config.webhook import WebhookConfig
 
@@ -45,9 +47,8 @@ def __init__(
         self.handle_in_background = handle_in_background
         self._background_feed_update_tasks: set[asyncio.Task[Any]] = set()
 
-    @abstractmethod
-    def _get_bot_from_request(self, bound_request: BoundRequest) -> Bot | None:
-        raise NotImplementedError
+        if self.security is None:
+            warnings.warn("Security is not configured, skipping verification", UserWarning, stacklevel=3)
 
     @abstractmethod
     async def set_webhook(self, *args, **kwargs) -> Bot:
@@ -71,25 +72,41 @@ def _build_workflow_data(self, app: Any, **kwargs) -> dict[str, Any]:
             **kwargs,
         }
 
+    @abstractmethod
+    async def _get_bot_token_for_request(self, bound_request: BoundRequest) -> str | None:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def _get_bot_by_token(self, token: str) -> Bot | None:
+        raise NotImplementedError
+
     async def handle_request(self, bound_request: BoundRequest):
-        bot = self._get_bot_from_request(bound_request)
-        if bot is None:
-            return self.web_adapter.create_json_response(status=400, payload={"detail": "Bot not found"})
+        token = await self._get_bot_token_for_request(bound_request)
+        if token is None:
+            return self.web_adapter.create_json_response(status=400, payload={"detail": "Bot token not found"})
 
-        if self.security is not None and not await self.security.verify(bot=bot, bound_request=bound_request):
+        if self.security is not None and not await self.security.verify(
+            bot_token=token, bound_request=bound_request, dispatcher=self.dispatcher
+        ):
             return self.web_adapter.create_json_response(status=403, payload={"detail": "Forbidden"})
 
-        update = await bound_request.json()
+        try:
+            bot = await self._get_bot_by_token(token)
+        except TokenValidationError:
+            return self.web_adapter.create_json_response(status=400, payload={"detail": "Invalid bot token"})
+
+        if bot is None:
+            return self.web_adapter.create_json_response(status=400, payload={"detail": "Bot not found"})
 
+        update = await bound_request.json()
         if self.handle_in_background:
             return await self._handle_request_background(bot=bot, update=update)
-
         return await self._handle_request(bot=bot, update=update)
 
     def register(self, app: Any) -> None:
         self.web_adapter.register(
             app=app,
-            path=self.routing.path,
+            path=self.routing.webhook_path,
             handler=self.handle_request,
             on_startup=self.on_startup,
             on_shutdown=self.on_shutdown,
@@ -98,7 +115,7 @@ def register(self, app: Any) -> None:
     async def _handle_request(self, bot: Bot, update: dict[str, Any]) -> dict[str, Any]:
         result = await self.dispatcher.feed_webhook_update(bot=bot, update=update)
 
-        if not isinstance(result, TelegramMethod):
+        if result is None:
             return self.web_adapter.create_json_response(status=200, payload={})
 
         payload = self._build_webhook_payload(bot, result)
@@ -110,14 +127,12 @@ async def _handle_request(self, bot: Bot, update: dict[str, Any]) -> dict[str, A
         return self.web_adapter.create_json_response(status=200, payload=payload)
 
     async def _background_feed_update(self, bot: Bot, update: dict[str, Any]) -> None:
-        result = await self.dispatcher.feed_raw_update(bot=bot, update=update)  # **self.data
+        result = await self.dispatcher.feed_raw_update(bot=bot, update=update)
         if isinstance(result, TelegramMethod):
             await self.dispatcher.silent_call_request(bot=bot, result=result)
 
     async def _handle_request_background(self, bot: Bot, update: dict[str, Any]):
-        feed_update_task = asyncio.create_task(
-            self._background_feed_update(bot=bot, update=update),
-        )
+        feed_update_task = asyncio.create_task(self._background_feed_update(bot=bot, update=update))
         self._background_feed_update_tasks.add(feed_update_task)
         feed_update_task.add_done_callback(self._background_feed_update_tasks.discard)
 

src/aiogram_webhook/engines/simple.py

@@ -42,13 +42,16 @@ def __init__(
             handle_in_background=handle_in_background,
         )
 
-    def _get_bot_from_request(self, bound_request: BoundRequest) -> Bot | None:  # noqa: ARG002
+    async def _get_bot_token_for_request(self, bound_request: BoundRequest) -> str:  # noqa: ARG002
         """
-        Always returns the single Bot instance for any request.
+        Always returns the single bot token for any request.
 
         :param bound_request: The incoming bound request.
-        :return: The single Bot instance
+        :return: The single bot token
         """
+        return self.bot.token
+
+    async def _get_bot_by_token(self, token: str) -> Bot:  # noqa: ARG002
         return self.bot
 
     async def set_webhook(
@@ -78,11 +81,13 @@ async def set_webhook(
         params = config.model_dump(exclude_none=True)
 
         if self.security is not None:
-            secret_token = await self.security.get_secret_token(bot=self.bot)
+            secret_token = await self.security.secret_token(bot_token=self.bot.token)
             if secret_token is not None:
                 params["secret_token"] = secret_token
 
-        await self.bot.set_webhook(url=self.routing.webhook_point(self.bot), request_timeout=request_timeout, **params)
+        await self.bot.set_webhook(
+            url=await self.routing.webhook_url(self.bot), request_timeout=request_timeout, **params
+        )
         return self.bot
 
     async def on_startup(self, app: Any, *args, **kwargs) -> None:  # noqa: ARG002

src/aiogram_webhook/engines/token.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+from types import MappingProxyType
 from typing import TYPE_CHECKING, Any
 
 from aiogram import Bot, Dispatcher
+from aiogram.client.session.aiohttp import AiohttpSession
 from aiogram.utils.token import extract_bot_id
 
 from aiogram_webhook.config.bot import BotConfig
@@ -44,36 +46,30 @@ def __init__(
         )
         self.routing: TokenRouting = routing  # for type checker
         self.bot_config = bot_config or BotConfig()
+        self._session = self.bot_config.session or AiohttpSession()
         self._bots: dict[int, Bot] = {}
 
-    def _get_bot_from_request(self, bound_request: BoundRequest) -> Bot | None:
-        """
-        Get a :class:`Bot` instance from request by token.
-        If the bot is not yet created, it will be created automatically.
+    @property
+    def bots(self) -> MappingProxyType[int, Bot]:
+        return MappingProxyType(self._bots)
 
-        :param bound_request: Incoming request
-        :return: Bot instance or None
-        """
-        token = self.routing.extract_token(bound_request)
-        if not token:
-            return None
-        return self.get_bot(token)
+    async def _get_bot_token_for_request(self, bound_request: BoundRequest) -> str | None:
+        return await self.routing.resolve_token(bound_request)
 
-    def get_bot(self, token: str) -> Bot:
-        """
-        Resolve or create a Bot instance by token and cache it.
+    async def _get_bot_by_token(self, token: str) -> Bot:
+        bot_id = extract_bot_id(token)
+        existing_bot = self._bots.get(bot_id)
 
-        :param token: The bot token
-        :return: Bot
+        if existing_bot is None or existing_bot.token != token:
+            new_bot = self._build_bot(token)
+            self._bots[bot_id] = new_bot
+            return new_bot
 
-        .. note::
-            To connect the bot to Telegram API and set up webhook, use :meth:`set_webhook`.
-        """
-        bot = self._bots.get(extract_bot_id(token))
-        if not bot:
-            bot = Bot(token=token, session=self.bot_config.session, default=self.bot_config.default)
-            self._bots[bot.id] = bot
-        return bot
+        return existing_bot
+
+    def _build_bot(self, token: str) -> Bot:
+        """Build a new Bot instance from token."""
+        return Bot(token=token, session=self._session, default=self.bot_config.default)
 
     async def set_webhook(
         self,
@@ -96,22 +92,30 @@ async def set_webhook(
         :param request_timeout: Request timeout
         :return: Bot instance
         """
-        bot = self.get_bot(token)
-        config = self._build_webhook_config(
+
+        bot = await self._get_bot_by_token(token=token)
+        params = self._build_webhook_config(
             max_connections=max_connections,
             drop_pending_updates=drop_pending_updates,
             allowed_updates=allowed_updates,
-        )
-        params = config.model_dump(exclude_none=True)
+        ).model_dump(exclude_none=True)
 
         if self.security is not None:
-            secret_token = await self.security.get_secret_token(bot=bot)
+            secret_token = await self.security.secret_token(bot_token=token)
             if secret_token is not None:
                 params["secret_token"] = secret_token
 
-        await bot.set_webhook(url=self.routing.webhook_point(bot), request_timeout=request_timeout, **params)
+        await bot.set_webhook(url=await self.routing.webhook_url(bot), request_timeout=request_timeout, **params)
         return bot
 
+    async def remove_bot(self, bot_id: int) -> bool:
+        """Remove cached bot"""
+        bot = self._bots.get(bot_id)
+        if bot is None:
+            return False
+        del self._bots[bot_id]
+        return True
+
     async def on_startup(self, app: Any, *args, bots: set[Bot] | None = None, **kwargs) -> None:  # noqa: ARG002
         all_bots = set(bots) | set(self._bots.values()) if bots else set(self._bots.values())
         workflow_data = self._build_workflow_data(app=app, bots=all_bots, **kwargs)
@@ -121,6 +125,7 @@ async def on_shutdown(self, app: Any, *args, **kwargs) -> None:  # noqa: ARG002
         workflow_data = self._build_workflow_data(app=app, bots=set(self._bots.values()), **kwargs)
         await self.dispatcher.emit_shutdown(**workflow_data)
 
-        for bot in self._bots.values():
-            await bot.session.close()
+        if self.bot_config.session is None:
+            await self._session.close()
+
         self._bots.clear()