Avoid calling orjson to check for api status response (#17)

aiotankerkoenig/__init__.py

@@ -2,12 +2,13 @@
 from __future__ import annotations
 
 from .aiotankerkoenig import Tankerkoenig
+from .const import GasType, Sort, Status
 from .exceptions import (
     TankerkoenigConnectionError,
     TankerkoenigError,
     TankerkoenigInvalidKeyError,
 )
-from .models import GasType, PriceInfo, Sort, Station, Status
+from .models import PriceInfo, Station
 
 __all__ = [
     "Tankerkoenig",

aiotankerkoenig/aiotankerkoenig.py

@@ -7,15 +7,17 @@
 from typing import Any, Self
 
 from aiohttp import ClientSession
-import orjson
 from yarl import URL
 
-from .exceptions import (
-    TankerkoenigConnectionError,
-    TankerkoenigError,
-    TankerkoenigInvalidKeyError,
+from .const import GasType, Sort
+from .exceptions import TankerkoenigConnectionError, TankerkoenigError
+from .models import (
+    PriceInfo,
+    PriceInfoResponse,
+    Station,
+    StationDetailResponse,
+    StationListResponse,
 )
-from .models import GasType, PriceInfo, Sort, Station
 
 VERSION = metadata.version(__package__)
 
@@ -27,9 +29,11 @@ class Tankerkoenig:
     api_key: str
     session: ClientSession | None = None
     request_timeout: int = 10
+
     _close_session: bool = False
 
-    async def _request(self, path: str, params: dict[str, Any]) -> Any:
+    async def _request(self, path: str, params: dict[str, Any]) -> str:
+        """Handle request to tankerkoenig.de API."""
         url = URL.build(
             scheme="https",
             host="creativecommons.tankerkoenig.de",
@@ -38,19 +42,23 @@ async def _request(self, path: str, params: dict[str, Any]) -> Any:
         )
 
         headers = {
-            "User-Agent": f"aiotankerkoenig/{VERSION}",
             "Accept": "application/json",
         }
 
         if self.session is None:
             self.session = ClientSession()
             self._close_session = True
-            self.session.headers.update(headers)
+            headers.update(
+                {
+                    "User-Agent": f"aiotankerkoenig/{VERSION}",
+                },
+            )
 
         try:
             async with asyncio.timeout(self.request_timeout):
                 response = await self.session.get(
                     url,
+                    headers=headers,
                 )
         except asyncio.TimeoutError as exception:
             msg = "Timeout occurred while connecting to tankerkoenig.de API"
@@ -59,32 +67,15 @@ async def _request(self, path: str, params: dict[str, Any]) -> Any:
             ) from exception
 
         content_type = response.headers.get("Content-Type", "")
-
+        text = await response.text()
         if "application/json" not in content_type:
-            text = await response.text()
-            msg = "Unexpected response from tankerkoenig.de API"
+            msg = "Unexpected content type from tankerkoenig.de API"
             raise TankerkoenigError(
                 msg,
                 {"Content-Type": content_type, "response": text},
             )
 
-        obj = orjson.loads(await response.text())  # pylint: disable=maybe-no-member
-        if not obj["ok"]:
-            message = obj["message"].lower()
-            if any(x in message.lower() for x in ("api-key", "apikey")):
-                msg = "tankerkoenig.de API responded with an invalid key error"
-                raise TankerkoenigInvalidKeyError(
-                    msg,
-                    {"response": obj},
-                )
-
-            msg = "tankerkoenig.de API responded with an error"
-            raise TankerkoenigError(
-                msg,
-                {"response": obj},
-            )
-
-        return obj
+        return text
 
     async def nearby_stations(
         self,
@@ -104,7 +95,7 @@ async def nearby_stations(
                 "sort": sort,
             },
         )
-        return [Station.from_dict(station) for station in result["stations"]]
+        return StationListResponse.from_json(result).stations
 
     async def station_details(
         self,
@@ -117,7 +108,7 @@ async def station_details(
                 "id": station_id,
             },
         )
-        return Station.from_dict(result["station"])
+        return StationDetailResponse.from_json(result).station
 
     async def prices(
         self,
@@ -130,10 +121,7 @@ async def prices(
                 "ids": ",".join(station_ids),
             },
         )
-        return {
-            station_id: PriceInfo.from_dict(price_info)
-            for station_id, price_info in result["prices"].items()
-        }
+        return PriceInfoResponse.from_json(result).prices
 
     async def close(self) -> None:
         """Close open client session."""

aiotankerkoenig/const.py

@@ -0,0 +1,27 @@
+"""Constants for the aiotankerkoenig."""
+from enum import StrEnum
+
+
+class GasType(StrEnum):
+    """Gas type."""
+
+    ALL = "all"
+    DIESEL = "diesel"
+    E5 = "e5"
+    E10 = "e10"
+
+
+class Sort(StrEnum):
+    """Sort type."""
+
+    DISTANCE = "dist"
+    PRICE = "price"
+    TIME = "time"
+
+
+class Status(StrEnum):
+    """Status type."""
+
+    OPEN = "open"
+    CLOSED = "closed"
+    UNKNOWN = "unknown"

aiotankerkoenig/models.py

@@ -2,48 +2,75 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from enum import StrEnum
-from typing import Any
+from typing import Any, Self
 
-from mashumaro import DataClassDictMixin, field_options
+from mashumaro import field_options
+from mashumaro.mixins.orjson import DataClassORJSONMixin
 
+from .const import Status
+from .exceptions import TankerkoenigError, TankerkoenigInvalidKeyError
 
-class GasType(StrEnum):
-    """Gas type."""
 
-    ALL = "all"
-    DIESEL = "diesel"
-    E5 = "e5"
-    E10 = "e10"
+@dataclass(frozen=True, slots=True, kw_only=True)
+class TankerkoenigResponse(DataClassORJSONMixin):
+    """Base class for all responses."""
 
+    @classmethod
+    def __pre_deserialize__(
+        cls: type[Self],
+        d: dict[Any, Any],
+    ) -> dict[Any, Any]:
+        """Raise when response was unexpected."""
+        if d.get("ok"):
+            return d
 
-class Sort(StrEnum):
-    """Sort type."""
+        message = d.get("message", "")
+        if any(x in message.lower() for x in ("api-key", "apikey")):
+            msg = "tankerkoenig.de API responded with an invalid key error"
+            raise TankerkoenigInvalidKeyError(
+                msg,
+                {"response": d},
+            )
 
-    DISTANCE = "dist"
-    PRICE = "price"
-    TIME = "time"
+        msg = f"tankerkoenig.de API responded with an error: {message}"
+        raise TankerkoenigError(
+            msg,
+            {"response": d.pop("message")},
+        )
 
 
-class Status(StrEnum):
-    """Status type."""
+@dataclass(frozen=True, slots=True, kw_only=True)
+class StationListResponse(TankerkoenigResponse):
+    """Class representing a station list response."""
 
-    OPEN = "open"
-    CLOSED = "closed"
-    UNKNOWN = "unknown"
+    stations: list[Station]
 
 
-@dataclass(frozen=True, slots=True)
-class OpeningTime(DataClassDictMixin):
+@dataclass(frozen=True, slots=True, kw_only=True)
+class StationDetailResponse(TankerkoenigResponse):
+    """Class representing a station response."""
+
+    station: Station
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class PriceInfoResponse(TankerkoenigResponse):
+    """Class representing a station price info response."""
+
+    prices: dict[str, PriceInfo]
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class OpeningTime:
     """Class representing a station opening time."""
 
     end: str
     start: str
     text: str
 
 
-@dataclass(frozen=True, slots=True)
-class Station(DataClassDictMixin):
+@dataclass(frozen=True, slots=True, kw_only=True)
+class Station:
     """Class representing a station."""
 
     id: str
@@ -74,8 +101,8 @@ class Station(DataClassDictMixin):
     distance: float | None = field(metadata=field_options(alias="dist"), default=None)
 
 
-@dataclass(frozen=True, slots=True)
-class PriceInfo(DataClassDictMixin):
+@dataclass(frozen=True, slots=True, kw_only=True)
+class PriceInfo:
     """Class representing a station price info."""
 
     status: Status = Status.UNKNOWN