Quick REST Client Wrapper, and an example wrapper for the blink(1)control REST API

blinky.py

"""
Wrapper client for the blink(1)control REST API.

Example Usage:
      import blinky
      b = Blinky()

      b.on()  # Let there be (white) light

      # Have the first LED fade to #336699 over 10 seconds
      b.fade_to_rgb(rgb='#336699', time=10, ledn=1)
      # Have the other LED fade to red almost instantly (0.1 secs)
      b.fade_to_rgb(rgb='red', ledn=2)

      b.play_pattern(pname='policecar')  # play a known pattern as defined in the blink(1)control app
      b.stop_pattern(pname='policecar')

      b.off()  # no more light


For details of the underlying REST API, see:
  https://github.com/todbot/blink1/blob/master/docs/app-url-api.md
"""
from .resty import SimpleRestClient, SimpleQuery


class Blinky(SimpleRestClient):

    def __init__(self, base_url='http://localhost:8934/blink1'):
        super(Blinky, self).__init__(base_url)

    on = SimpleQuery('on')   # Stop pattern playback and set blink(1) to white (#FFFFFF)
    off = SimpleQuery('off')  # Stop pattern playback and set blink(1) to black (#000000)

    play_pattern = SimpleQuery('pattern/play', params=['pname'])  # Play/test a specific color pattern
    stop_pattern = SimpleQuery('pattern/stop', params=['pname'])  # Stop playback of given pattern or all patterns

    # Send fadeToRGB command to blink(1) with hex color and fade time (defaults to 0.1 if not provided)
    # The id parameter can be used to address specific blink(1) device
    # The ledn parameted can be used to choose LED to control. 0=all, 1=LED A, 2=LED B
    fade_to_rgb = SimpleQuery('fadeToRGB', params=['rgb'], optional=['time', 'id', 'ledn'])

    id = SimpleQuery('id')  # Display blink1_id and blink1 serial numbers (if any)
    enumerate = SimpleQuery('enumerate')  # Re-enumerate and List available blink(1) devices
    last_color = SimpleQuery('lastColor')  # Return the last color command sent to blink(1)
    regenerate_id = SimpleQuery('regenerateblink1id')  # Generate, save, and return new blink1_id

resty.py

"""
resty.py -- Create simple REST clients quickly.

This is currently a POC and supports only basic GET/POST queries. It is meant to allow users to quickly define a Python
 class that interfaces with a REST API.


See blinky.py for an example.
"""

import json
import types
import urllib
import urllib2
from urlparse import urljoin


class SimpleRestClient(object):
    """
    Base class for defining a simple REST API wrapper class
    """

    def __init__(self, base_url):
        """
        :param base_url: Base URL for REST API we are wrapping.
        """
        self.base_url = base_url
        self._bind_queries()

    def _bind_queries(self):
        # for each class attribute that is an SimpleQuery instance, bind the its query function as an object method
        queries = ((key, obj) for key, obj in self.__class__.__dict__.items() if isinstance(obj, SimpleQuery))
        for key, obj in queries:
            setattr(self, key, types.MethodType(obj.get_query_func(self.base_url), self))


class SimpleQuery(object):
    """Factory class for a specific REST query.
    Instantiate this as an attribute to a :SimpleRestClient: class to define an API method.
    """

    supported_methods = ('GET', 'POST')

    def __init__(self, command, params=None, optional=None, method='GET'):
        """
        :param command: Query path, relative to base_url
        :param params: Required query parameters
        :param optional: Optional query parameters
        :param method: HTTP method to use. Defaults to GET.
        """
        assert method in self.supported_methods, 'Unsupported method. Expecting: {0}'.format(self.supported_methods)
        self.command = command
        self.required_params = frozenset(params or [])
        self.valid_params = frozenset(optional or []).union(self.required_params)
        self.method = method

    def get_query_func(self, base_url):
        """Returns a callable that will perform the actual REST call.

        For now, always assume that the service will return a JSON output.

        :param base_url: Base URL for REST API we are wrapping.
        :return: callable
        """
        def _REST_query(caller, *args, **kwargs):
            self._validate_kwargs(args, kwargs)
            return self._load(base_url, self.command, kwargs)
        return _REST_query

    @staticmethod
    def load_url(url, data=None):
        """Wrapper method for urllib2.urlopen that handles urllib2 exceptions and parses the response data as JSON.

        :param url: URL to load.
        :param data: addition data to be sent to server. HTTP POST method will be used if this is defined.
        :return: output of json.loads
        """
        try:
            response = urllib2.urlopen(url, data).read()
        except urllib2.HTTPError as e:
            raise QueryError('Could not fulfill request. Error code {0}'.format(e.code))
        except urllib2.URLError as e:
            raise ConnectionError('Could not reach server. Reason {0}'.format(e.reason))
        return json.loads(response)

    @staticmethod
    def join_url(base_url, command):
        # we play musical chairs with the slashes to make sure urljoin does the right thing when the base url contains
        # a partial path and not just a netloc
        return urljoin(base_url.rstrip('/') + '/', command.lstrip('/'))

    def _load(self, base_url, command, kwargs):
        url = self.join_url(base_url, command)
        params = urllib.urlencode(kwargs)
        if self.method == 'GET':
            url = url + '?' + params
            params = None
        return self.load_url(url, params)

    def _validate_kwargs(self, input_args, input_kwargs):
        missing_params = self.required_params.difference(input_kwargs)
        invalid_params = [x for x in input_kwargs if x not in self.valid_params]

        msg = []
        if input_args:
            msg.append(' * No positional args expected. {0} provided.'.format(len(input_args)))
        if missing_params:
            msg.append(' * Required keyword args were not provided: {0}'.format(', '.join(missing_params)))
        if invalid_params:
            msg.append(' * Unexpected keyword args provided: {0}'.format(', '.join(invalid_params)))

        if msg:
            raise UsageError('Invalid parameters for "{0}":\n{1}'.format(self.command, '\n'.join(msg)))


class ConnectionError(Exception):
    """Raised when a connection could not be made to the server.
    """
    pass


class QueryError(Exception):
    """Raised when the server returns an error code.
    """
    pass


class UsageError(Exception):
    """Raised on invalid method calls, e.g. when a required parameter is not provided.
    """
    pass