[errors] prepare new errors model

Author: capcom6

android_sms_gateway/ahttp.py

@@ -1,6 +1,10 @@
 import abc
 import typing as t
 
+from .errors import (
+    error_from_status,
+)
+
 
 class AsyncHttpClient(t.Protocol):
     @abc.abstractmethod
@@ -70,15 +74,31 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
             await self._session.close()
             self._session = None
 
+        async def _process_response(self, response: aiohttp.ClientResponse) -> dict:
+            try:
+                response.raise_for_status()
+                return await response.json()
+            except aiohttp.ClientResponseError as e:
+                # Extract error message from response if available
+                error_data = {}
+                try:
+                    error_data = await response.json()
+                except ValueError:
+                    # Response is not JSON
+                    pass
+
+                # Use the error mapping to create appropriate exception
+                error_message = str(e) or "HTTP request failed"
+                raise error_from_status(error_message, response.status, error_data)
+
         async def get(
             self, url: str, *, headers: t.Optional[t.Dict[str, str]] = None
         ) -> dict:
             if self._session is None:
                 raise ValueError("Session not initialized")
 
             async with self._session.get(url, headers=headers) as response:
-                response.raise_for_status()
-                return await response.json()
+                return await self._process_response(response)
 
         async def post(
             self,
@@ -93,8 +113,7 @@ async def post(
             async with self._session.post(
                 url, headers=headers, json=payload
             ) as response:
-                response.raise_for_status()
-                return await response.json()
+                return await self._process_response(response)
 
         async def put(
             self,
@@ -109,8 +128,7 @@ async def put(
             async with self._session.put(
                 url, headers=headers, json=payload
             ) as response:
-                response.raise_for_status()
-                return await response.json()
+                return await self._process_response(response)
 
         async def patch(
             self,
@@ -125,8 +143,7 @@ async def patch(
             async with self._session.patch(
                 url, headers=headers, json=payload
             ) as response:
-                response.raise_for_status()
-                return await response.json()
+                return await self._process_response(response)
 
         async def delete(
             self, url: str, *, headers: t.Optional[t.Dict[str, str]] = None
@@ -135,7 +152,7 @@ async def delete(
                 raise ValueError("Session not initialized")
 
             async with self._session.delete(url, headers=headers) as response:
-                response.raise_for_status()
+                await self._process_response(response)
 
     DEFAULT_CLIENT = AiohttpAsyncHttpClient
 except ImportError:
@@ -163,15 +180,31 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
             await self._client.aclose()
             self._client = None
 
+        async def _process_response(self, response: httpx.Response) -> dict:
+            try:
+                response.raise_for_status()
+                return response.json()
+            except httpx.HTTPStatusError as e:
+                # Extract error message from response if available
+                error_data = {}
+                try:
+                    error_data = response.json()
+                except ValueError:
+                    # Response is not JSON
+                    pass
+
+                # Use the error mapping to create appropriate exception
+                error_message = str(e) or "HTTP request failed"
+                raise error_from_status(error_message, response.status_code, error_data)
+
         async def get(
             self, url: str, *, headers: t.Optional[t.Dict[str, str]] = None
         ) -> dict:
             if self._client is None:
                 raise ValueError("Client not initialized")
 
             response = await self._client.get(url, headers=headers)
-
-            return response.raise_for_status().json()
+            return await self._process_response(response)
 
         async def post(
             self,
@@ -184,8 +217,7 @@ async def post(
                 raise ValueError("Client not initialized")
 
             response = await self._client.post(url, headers=headers, json=payload)
-
-            return response.raise_for_status().json()
+            return await self._process_response(response)
 
         async def put(
             self,
@@ -198,8 +230,7 @@ async def put(
                 raise ValueError("Client not initialized")
 
             response = await self._client.put(url, headers=headers, json=payload)
-
-            return response.raise_for_status().json()
+            return await self._process_response(response)
 
         async def patch(
             self,
@@ -212,8 +243,7 @@ async def patch(
                 raise ValueError("Client not initialized")
 
             response = await self._client.patch(url, headers=headers, json=payload)
-
-            return response.raise_for_status().json()
+            return await self._process_response(response)
 
         async def delete(
             self, url: str, *, headers: t.Optional[t.Dict[str, str]] = None
@@ -222,7 +252,7 @@ async def delete(
                 raise ValueError("Client not initialized")
 
             response = await self._client.delete(url, headers=headers)
-            response.raise_for_status()
+            await self._process_response(response)
 
     DEFAULT_CLIENT = HttpxAsyncHttpClient
 except ImportError:

android_sms_gateway/errors.py

@@ -0,0 +1,72 @@
+import typing as t
+
+
+class APIError(Exception):
+    """Base class for API exceptions."""
+
+    def __init__(
+        self,
+        message: str,
+        status_code: t.Optional[int] = None,
+        response: t.Optional[dict] = None,
+    ):
+        super().__init__(message)
+        self.status_code = status_code
+        self.response = response
+
+
+class BadRequestError(APIError):
+    """400 - Bad Request"""
+
+
+class UnauthorizedError(APIError):
+    """401 - Unauthorized"""
+
+
+class ForbiddenError(APIError):
+    """403 - Forbidden"""
+
+
+class NotFoundError(APIError):
+    """404 - Not Found"""
+
+
+class InternalServerError(APIError):
+    """500 - Internal Server Error"""
+
+
+class ServiceUnavailableError(APIError):
+    """503 - Service Unavailable"""
+
+
+class GatewayTimeoutError(APIError):
+    """504 - Gateway Timeout"""
+
+
+_ERROR_MAP = {
+    400: BadRequestError,
+    401: UnauthorizedError,
+    403: ForbiddenError,
+    404: NotFoundError,
+    500: InternalServerError,
+    503: ServiceUnavailableError,
+    504: GatewayTimeoutError,
+}
+
+
+def error_from_status(
+    message: str, status: int, response: t.Optional[dict] = None
+) -> APIError:
+    """Factory function to map HTTP status codes to appropriate APIError subclasses.
+
+    Args:
+        message: Error message
+        status: HTTP status code
+        response: Optional response data
+
+    Returns:
+        Appropriate APIError subclass or APIError as fallback
+    """
+    return _ERROR_MAP.get(status, APIError)(
+        message, status_code=status, response=response
+    )

android_sms_gateway/http.py

@@ -4,6 +4,10 @@
 
 from . import client, domain
 
+from .errors import (
+    error_from_status,
+)
+
 
 class HttpClient(t.Protocol):
     @abc.abstractmethod
@@ -130,8 +134,21 @@ def patch(
             )
 
         def _process_response(self, response: requests.Response) -> dict:
-            response.raise_for_status()
-            return response.json()
+            try:
+                response.raise_for_status()
+                return response.json()
+            except requests.exceptions.HTTPError as e:
+                # Extract error message from response if available
+                error_data = {}
+                try:
+                    error_data = response.json()
+                except ValueError:
+                    # Response is not JSON
+                    pass
+
+                # Use the error mapping to create appropriate exception
+                error_message = str(e) or "HTTP request failed"
+                raise error_from_status(error_message, response.status_code, error_data)
 
     DEFAULT_CLIENT = RequestsHttpClient
 except ImportError:
@@ -159,13 +176,30 @@ def __exit__(self, exc_type, exc_val, exc_tb):
             self._client.close()
             self._client = None
 
+        def _process_response(self, response: httpx.Response) -> dict:
+            try:
+                response.raise_for_status()
+                return response.json()
+            except httpx.HTTPStatusError as e:
+                # Extract error message from response if available
+                error_data = {}
+                try:
+                    error_data = response.json()
+                except ValueError:
+                    # Response is not JSON
+                    pass
+
+                # Use the error mapping to create appropriate exception
+                error_message = str(e) or "HTTP request failed"
+                raise error_from_status(error_message, response.status_code, error_data)
+
         def get(
             self, url: str, *, headers: t.Optional[t.Dict[str, str]] = None
         ) -> dict:
             if self._client is None:
                 raise ValueError("Client not initialized")
 
-            return self._client.get(url, headers=headers).raise_for_status().json()
+            return self._process_response(self._client.get(url, headers=headers))
 
         def post(
             self,
@@ -177,10 +211,8 @@ def post(
             if self._client is None:
                 raise ValueError("Client not initialized")
 
-            return (
+            return self._process_response(
                 self._client.post(url, headers=headers, json=payload)
-                .raise_for_status()
-                .json()
             )
 
         def delete(
@@ -189,7 +221,7 @@ def delete(
             if self._client is None:
                 raise ValueError("Client not initialized")
 
-            self._client.delete(url, headers=headers).raise_for_status()
+            self._process_response(self._client.delete(url, headers=headers))
 
         def put(
             self,
@@ -201,10 +233,8 @@ def put(
             if self._client is None:
                 raise ValueError("Client not initialized")
 
-            return (
+            return self._process_response(
                 self._client.put(url, headers=headers, json=payload)
-                .raise_for_status()
-                .json()
             )
 
         def patch(
@@ -217,10 +247,8 @@ def patch(
             if self._client is None:
                 raise ValueError("Client not initialized")
 
-            return (
+            return self._process_response(
                 self._client.patch(url, headers=headers, json=payload)
-                .raise_for_status()
-                .json()
             )
 
     DEFAULT_CLIENT = HttpxHttpClient

tests/test_client.py

@@ -1,12 +1,12 @@
 import os
 import pytest
-from requests import HTTPError
 
 from android_sms_gateway.client import APIClient
 from android_sms_gateway.constants import DEFAULT_URL
 from android_sms_gateway.domain import Webhook
 from android_sms_gateway.enums import WebhookEvent
 from android_sms_gateway.http import RequestsHttpClient
+from android_sms_gateway import errors
 
 
 @pytest.fixture
@@ -25,12 +25,15 @@ def client():
 
     :yields: An instance of `APIClient`.
     """
-    with RequestsHttpClient() as h, APIClient(
-        os.environ.get("API_LOGIN") or "test",
-        os.environ.get("API_PASSWORD") or "test",
-        base_url=os.environ.get("API_BASE_URL") or DEFAULT_URL,
-        http=h,
-    ) as c:
+    with (
+        RequestsHttpClient() as h,
+        APIClient(
+            os.environ.get("API_LOGIN") or "test",
+            os.environ.get("API_PASSWORD") or "test",
+            base_url=os.environ.get("API_BASE_URL") or DEFAULT_URL,
+            http=h,
+        ) as c,
+    ):
         yield c
 
 
@@ -75,7 +78,7 @@ def test_webhook_create_invalid_url(self, client: APIClient):
 
         :param client: An instance of `APIClient`.
         """
-        with pytest.raises(HTTPError):
+        with pytest.raises(errors.APIError):
             client.create_webhook(
                 Webhook(None, url="not_a_url", event=WebhookEvent.SMS_RECEIVED)
             )