Added detection of API Errors and fixed coverage

aiogram/client/errors_middleware.py

@@ -1,64 +0,0 @@
-from __future__ import annotations
-
-import re
-from typing import TYPE_CHECKING, List, Type
-
-from aiogram.methods import Response, TelegramMethod
-from aiogram.types import TelegramObject
-from aiogram.utils.exceptions.base import TelegramAPIError
-from aiogram.utils.exceptions.exceptions import (
-    CantParseEntitiesStartTag,
-    CantParseEntitiesUnclosed,
-    CantParseEntitiesUnmatchedTags,
-    CantParseEntitiesUnsupportedTag,
-    DetailedTelegramAPIError,
-)
-
-if TYPE_CHECKING:
-    from aiogram.client.bot import Bot
-    from aiogram.client.session.base import NextRequestMiddlewareType
-
-
-class RequestErrorMiddleware:
-    def __init__(self) -> None:
-        self._registry: List[Type[DetailedTelegramAPIError]] = [
-            CantParseEntitiesStartTag,
-            CantParseEntitiesUnmatchedTags,
-            CantParseEntitiesUnclosed,
-            CantParseEntitiesUnsupportedTag,
-        ]
-
-    def mount(self, error: Type[DetailedTelegramAPIError]) -> Type[DetailedTelegramAPIError]:
-        if error in self:
-            raise ValueError(f"{error!r} is already registered")
-        if not hasattr(error, "patterns"):
-            raise ValueError(f"{error!r} has no attribute 'patterns'")
-        self._registry.append(error)
-        return error
-
-    def detect_error(self, err: TelegramAPIError) -> TelegramAPIError:
-        message = err.message
-        for variant in self._registry:
-            for pattern in variant.patterns:
-                if match := re.match(pattern, message):
-                    return variant(
-                        method=err.method,
-                        message=err.message,
-                        match=match,
-                    )
-        return err
-
-    def __contains__(self, item: Type[DetailedTelegramAPIError]) -> bool:
-        return item in self._registry
-
-    async def __call__(
-        self,
-        bot: Bot,
-        method: TelegramMethod[TelegramObject],
-        make_request: NextRequestMiddlewareType,
-    ) -> Response[TelegramObject]:
-        try:
-            return await make_request(bot, method)
-        except TelegramAPIError as e:
-            detected_err = self.detect_error(err=e)
-            raise detected_err from e

aiogram/client/session/base.py

@@ -4,6 +4,7 @@ import abc
 import datetime
 import json
 from functools import partial
+from http import HTTPStatus
 from types import TracebackType
 from typing import (
     TYPE_CHECKING,
@@ -25,8 +26,13 @@ from aiogram.utils.helper import Default
 from ...methods import Response, TelegramMethod
 from ...methods.base import TelegramType
 from ...types import UNSET, TelegramObject
+from ...utils.exceptions.bad_request import BadRequest
+from ...utils.exceptions.conflict import ConflictError
+from ...utils.exceptions.network import EntityTooLarge
+from ...utils.exceptions.not_found import NotFound
+from ...utils.exceptions.server import RestartingTelegram, ServerError
 from ...utils.exceptions.special import MigrateToChat, RetryAfter
-from ..errors_middleware import RequestErrorMiddleware
+from ...utils.exceptions.unauthorized import UnauthorizedError
 from ..telegram import PRODUCTION, TelegramAPIServer
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -55,12 +61,8 @@ class BaseSession(abc.ABC):
     timeout: Default[float] = Default(fget=lambda self: float(self.__class__.default_timeout))
     """Session scope request timeout"""
 
-    errors_middleware: ClassVar[RequestErrorMiddleware] = RequestErrorMiddleware()
-
     def __init__(self) -> None:
-        self.middlewares: List[RequestMiddlewareType[TelegramObject]] = [
-            self.errors_middleware,
-        ]
+        self.middlewares: List[RequestMiddlewareType[TelegramObject]] = []
 
     def check_response(
         self, method: TelegramMethod[TelegramType], status_code: int, content: str
@@ -70,10 +72,11 @@ class BaseSession(abc.ABC):
         """
         json_data = self.json_loads(content)
         response = method.build_response(json_data)
-        if response.ok:
+        if HTTPStatus.OK <= status_code <= HTTPStatus.IM_USED and response.ok:
             return response
 
         description = cast(str, response.description)
+
         if parameters := response.parameters:
             if parameters.retry_after:
                 raise RetryAfter(
@@ -85,6 +88,21 @@ class BaseSession(abc.ABC):
                     message=description,
                     migrate_to_chat_id=parameters.migrate_to_chat_id,
                 )
+        if status_code == HTTPStatus.BAD_REQUEST:
+            raise BadRequest(method=method, message=description)
+        if status_code == HTTPStatus.NOT_FOUND:
+            raise NotFound(method=method, message=description)
+        if status_code == HTTPStatus.CONFLICT:
+            raise ConflictError(method=method, message=description)
+        if status_code in (HTTPStatus.UNAUTHORIZED, HTTPStatus.FORBIDDEN):
+            raise UnauthorizedError(method=method, message=description)
+        if status_code == HTTPStatus.REQUEST_ENTITY_TOO_LARGE:
+            raise EntityTooLarge(method=method, message=description)
+        if status_code >= HTTPStatus.INTERNAL_SERVER_ERROR:
+            if "restart" in description:
+                raise RestartingTelegram(method=method, message=description)
+            raise ServerError(method=method, message=description)
+
         raise TelegramAPIError(
             method=method,
             message=description,