Add custom Adafruit IO error types and raise them when requests fail.

Author: tdicola

Adafruit_IO/__init__.py

@@ -1 +1 @@
-from .client import Client
+from .client import Client, AdafruitIOError, RequestError, ThrottlingError

Adafruit_IO/client.py

@@ -3,6 +3,26 @@
 from urllib3 import connection_from_url
 from urllib import urlencode, quote
 
+
+class AdafruitIOError(Exception):
+  """Base class for all Adafruit IO request failures."""
+  pass
+
+class RequestError(Exception):
+  """General error for a failed Adafruit IO request."""
+  def __init__(self, response):
+    super(RequestError, self).__init__("Adafruit IO request failed: {0} {1}".format(
+      response.status, response.reason))
+
+class ThrottlingError(AdafruitIOError):
+  """Too many requests have been made to Adafruit IO in a short period of time.
+  Reduce the rate of requests and try again later.
+  """
+  def __init__(self):
+    super(ThrottlingError, self).__init__("Exceeded the limit of Adafruit IO "  \
+      "requests in a short period of time. Please reduce the rate of requests " \
+      "and try again later.")
+
 #fork of ApiClient Class: https://github.com/shazow/apiclient
 class Client(object):
   BASE_URL = 'https://io.adafruit.com/'
@@ -21,7 +41,17 @@ def _compose_url(self, path):
   def _compose_get_url(self, path, params=None):
     return self.BASE_URL + path + '?' + urlencode(params)
 
+  def _handle_error(sefl, response):
+    # Handle explicit errors.
+    if response.status == 429:
+      raise ThrottlingError()
+    # Handle all other errors (400 & 500 level HTTP responses)
+    elif response.status >= 400:
+      raise RequestError(response)
+    # Else do nothing if there was no error.
+
   def _handle_response(self, response):
+    self._handle_error(response)
     return json.loads(response.data)
 
   def _request(self, method, path, params=None):

tests/README.txt

@@ -0,0 +1,11 @@
+Adafruit IO Python Client Test README
+
+To run these tests you must have the pytest module installed.  You can install
+this (assuming you have pip installed) by executing:
+  sudo pip install pytest
+
+Some tests require a valid Adafruit IO account to run, and they key for this
+account is provided in the ADAFRUIT_IO_KEY environment variable.  Make sure to
+set this envirionment variable before running the tests, for example to run all
+the tests with a key execute in this directory:
+  ADAFRUIT_IO_KEY=my_io_key_value py.test

tests/test_errors.py

@@ -0,0 +1,31 @@
+import os
+import time
+
+import pytest
+
+import Adafruit_IO
+
+
+def _get_client():
+  """Return an Adafruit IO client instance configured with the key specified in
+  the ADAFRUIT_IO_KEY environment variable.
+  """
+  key = os.environ.get('ADAFRUIT_IO_KEY', None)
+  if key is None:
+    raise RuntimeError("ADAFRUIT_IO_KEY environment variable must be set with " \
+      "valid Adafruit IO key to run this test!")
+  return Adafruit_IO.Client(key)
+
+
+class TestErrors:    
+  def test_request_error_from_bad_key(self):
+    io = Adafruit_IO.Client("this is a bad key from a test")
+    with pytest.raises(Adafruit_IO.RequestError):
+      io.send("TestStream", 42)
+
+  def test_throttling_error_after_6_requests_in_short_period(self):
+    io = _get_client()
+    with pytest.raises(Adafruit_IO.ThrottlingError):
+      for i in range(6):
+        io.send("TestStream", 42)
+        time.sleep(0.1)  # Small delay to keep from hammering network.