Make it simpler

Author: fabaff

dingz/__init__.py

@@ -7,52 +7,44 @@
 import aiohttp
 import async_timeout
 
-from .constants import TIMEOUT, USER_AGENT
+from .constants import TIMEOUT, USER_AGENT, CONTENT_TYPE_JSON, CONTENT_TYPE, CONTENT_TYPE_TEXT_PLAIN
 from .exceptions import DingzConnectionError, DingzError
 
 
-async def _request(
+async def make_call(
     self,
     uri: str,
     method: str = "GET",
     data: Optional[Any] = None,
     json_data: Optional[dict] = None,
-    params: Optional[Mapping[str, str]] = None,
+    parameters: Optional[Mapping[str, str]] = None,
+    token: str = None,
 ) -> Any:
-    """Handle a request to the dingz unit."""
+    """Handle the requests to the dingz unit."""
+
     headers = {
         "User-Agent": USER_AGENT,
-        "Accept": "application/json, text/plain, */*",
+        "Accept": f"{CONTENT_TYPE_JSON}, {CONTENT_TYPE_TEXT_PLAIN}, */*",
     }
 
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+
     if self._session is None:
         self._session = aiohttp.ClientSession()
         self._close_session = True
 
     try:
         with async_timeout.timeout(TIMEOUT):
             response = await self._session.request(
-                method, uri, data=data, json=json_data, params=params, headers=headers,
+                method, uri, data=data, json=json_data, params=parameters, headers=headers,
             )
     except asyncio.TimeoutError as exception:
-        raise DingzConnectionError(
-            "Timeout occurred while connecting to dingz unit."
-        ) from exception
+        raise DingzConnectionError("Timeout occurred while connecting to dingz unit") from exception
     except (aiohttp.ClientError, socket.gaierror) as exception:
-        raise DingzConnectionError(
-            "Error occurred while communicating with dingz."
-        ) from exception
-
-    content_type = response.headers.get("Content-Type", "")
-    if (response.status // 100) in [4, 5]:
-        contents = await response.read()
-        response.close()
-
-        if content_type == "application/json":
-            raise DingzError(response.status, json.loads(contents.decode("utf8")))
-        raise DingzError(response.status, {"message": contents.decode("utf8")})
+        raise DingzConnectionError("Error occurred while communicating with dingz") from exception
 
-    if "application/json" in content_type:
+    if CONTENT_TYPE_JSON in response.headers.get(CONTENT_TYPE, ""):
         response_json = await response.json()
         return response_json
 

dingz/constants.py

@@ -33,3 +33,8 @@
 PIR_CONFIGURATION = "pir_config"
 THERMOSTAT_CONFIGURATION = "thermostat_config"
 INPUT_CONFIGURATION = "input_config"
+
+# Communication constants
+CONTENT_TYPE_JSON = "application/json"
+CONTENT_TYPE = 'Content-Type'
+CONTENT_TYPE_TEXT_PLAIN = "text/plain"

dingz/dingz.py

@@ -4,7 +4,7 @@
 import aiohttp
 from yarl import URL
 
-from . import _request as request
+from . import make_call
 from .constants import API, DEVICE_INFO, TEMPERATURE, LIGHT, FRONT_LED_GET, FRONT_LED_SET
 from .exceptions import DingzError
 
@@ -28,38 +28,38 @@ def __init__(self, host: str, session: aiohttp.client.ClientSession = None) -> N
     async def get_device_info(self) -> None:
         """Get the details from the dingz."""
         url = URL(self.uri).join(URL(DEVICE_INFO))
-        response = await request(self, uri=url)
+        response = await make_call(self, uri=url)
         self._device_details = response
 
     async def get_temperature(self) -> None:
         """Get the room temperature from the dingz."""
         url = URL(self.uri).join(URL(TEMPERATURE))
-        response = await request(self, uri=url)
+        response = await make_call(self, uri=url)
         self._temperature = response["temperature"]
 
     async def get_light(self) -> None:
         """Get the light details from the switch."""
         url = URL(self.uri).join(URL(LIGHT))
-        response = await request(self, uri=url)
+        response = await make_call(self, uri=url)
         self._intensity = response["intensity"]
         self._day = response["day"]
 
     async def enabled(self) -> bool:
         """Return true if front LED is on."""
         url = URL(self.uri).join(URL(FRONT_LED_GET))
-        response = await request(self, uri=url)
+        response = await make_call(self, uri=url)
         return bool(response["on"])
 
     async def turn_on(self) -> None:
         """Enable/turn on the front LED."""
         data = {"action": "on"}
         url = URL(self.uri).join(URL(FRONT_LED_SET))
-        await request(self, uri=url, method="POST", data=data)
+        await make_call(self, uri=url, method="POST", data=data)
 
     async def turn_off(self) -> None:
         """Disable/turn off the front LED."""
         url = URL(self.uri).join(URL(FRONT_LED_SET))
-        await request(self, uri=url, method="POST", data={"action": "off"})
+        await make_call(self, uri=url, method="POST", data={"action": "off"})
 
     @property
     def device_details(self) -> str:
@@ -81,6 +81,8 @@ def intensity(self) -> float:
         """Return the current light intensity in lux."""
         return round(self._intensity, 1)
 
+    # See "Using Asyncio in Python" by Caleb Hattingh for implementation
+    # details.
     async def close(self) -> None:
         """Close an open client session."""
         if self._session and self._close_session: