Building a Python client for web APIs.

Imagine we are responsible for testing our company’s new home irrigation system product. Conviently, it is built from a Raspberry Pi and Arduino, which should be familiar to hobbyists. Along with those two computing platforms there are a collection of sensors and a pump connected to a water resevoir. When the soil moisture gets too dry the system should actuate the pump. Thus, flowing water to the plants and/or your basement floor

In this system the Arduino acts as the input/output interface. Its duty is to interface with the peripherals and communicate with the Raspberry Pi. It will accept ASCII command strings, parse them, and perform the respective action. Those actions are very primitie, such as retrieving sensor data or actuating the pump. For example sending the string get pressure would prompt the Arduino to record the current value from the pressure sensor and provide the measurement back to the Raspberry Pi. Similarly for the other types of sensors. Sending set pump 1 would activate the pump. That value latches until something sets it back (either a human or the control software). The Raspberry Pi is executing the control software.

The control software performs a cycle where it reads the sensor values, executes the control logic, and produce output commands. Since this system is pretty simple, the only output commands are for pump control. Whether or not to turn it on. On every cycle those steps will be performed. This system wouldn’t be very fun to test or study if it didn’t provide an API. Let’s say a REST API provided at the Raspberry Pi’s local IP address. Each of the accessor/command functions of the Arduino will be exposed through this API. Maybe even some higher level capabilities like retrieving all of the sensor data values , collecting X seconds of data, or returning some statistics.

 import requests
 response = requests.get('http://10.1.2.3/sensors/pressure')
 data = response.json()
 print(data)
 # {'timestamp': 1231231231.0, 'value': 14.7}

 post = {'value': 1}
 response = requests.post('http://10.1.2.3/pump/', data=post)
 data = response.json()
 print(data['feedback'])
 # 1

Reflecting the API

With only our embedded system and a workstation we can start to build up a testing infrastructure. First, we’ll need a way to interact with the system in an automation-friendly way. After we figure out how to talk to our system we can then build test cases and an automa- tion framework. Only after some goo-goo’s and ga-ga’s can we build a deeper vocabulary.

Our interface with the Raspberry Pi is through a REST API. If our test cases are full of GET and POST calls then it really obfuscates the intention of the test case. Building an interface that showcases the intention the test case will be worth its weight in gold. Code will be read infinitely more than it is written. System-level test cases typically have larger audiences than most code. If you are the poor soul responsible for testing a safety-critical system then you can bet on producing system verification reports for certification boards, customers, or grandmaster wizards. Wherever the amount of acceptable risk is extremely low (ideally zero) then there will many more eyes on the testing and verification of the system. Creating un- necessarily complex test cases or API only hinders the ability for these extra eyes to successfully scrutinize the test cases and system behavior. Building an API that allows the intention of the test to “be the star of the dish” (any Beat Bobby Flay fans out there?) fosters transparency by making the test so simple that anyone can understand it.

We’ll start off with the natural, Pythonic way to develop the API. As the irrigation system is subjected to feature requests and hardware changes we’ll try to adapt our implementation.

Property-based Interface

On our first attempt of developing an API for our system, we’d make a class with a bunch of getter/setter methods for each of the different devices. Something like get_pressure, get_light, get_moisture, send_command, etc. Each one of those functions would be extremely similar though. The only content that would change between the requests is the name of the sensor. Essentially each function is passing in a constant to some helper function that performs the actual request.

 import socket
 
 class IrrigationSystem(object):
     def __init__(self, host):
         self.host = host
         self.connection = None
 
     def connect(self):
         self.connection = socket.socket(socket.AF_INET, 
                                         socket.SOCK_STREAM)
         self.connection.connect((self.host, 8080))
 
     def close(self):
         self.connection.close()
 
     def send_command(self, device, value):
         command = 'set {device} {value}'
                   .format(device=device, value=value)

         self.connection.send(command)
         response = json.loads(self.connection.recv(1024))
 
         assert response['accepted'],
             'Command receipt indicates "{}" was not accepted'
             .format(response['command'])
 
         return response
 
     def get_value(self, sensor):
         request = 'get {device}'.format(device=sensor)
         self.connection.send(request)
         data = irrigation.recv(1024)
 
         return json.loads(data)
 
     @property
     def light(self):
         return self.get_value('light')
 
     @property
     def moisture(self):
         return self.get_value('moisture')
 
     @property
     def pressure(self):
         return self.get_value('pressure')
 
     @property
     def humidity(self):
         return self.get_value('humidity')
 
     @property
     def pump(self):
         return self.get_value('pump.feedback')
 
     @pump.setter
     def pump(self, value):
         self.send_command('pump', value)

The class looks pleasant. It represents the Arduino part of the system pretty well with a Pythonic implementation. The difficulty comes from scaling. Imagine our hardware development team added twenty more sensors. We essentially have to copy/paste one of the properties twenty more times and find/replace along the way. That’s quite annoying. Unit testing this class suffers from the same scalability issue. We would be copy/pasting some basic structure around and changing names, say light to moisture .

Descriptor-based Interface

Descriptors in Python are essentially re-usable properties. Instead of writing getter/setter functions for each of our different sensor/actuator types, we can build Sensor and Actuator descriptors. Python recognizes any class with __get__ and __set__ functions as a type of descriptor and handles them specially. When a descriptor is accessed, the __get__ function is called. When a value is assigned to the descriptor the __set__ function is called. First let’s see how the descriptors are used and look like. It’s easier to make sense of their implementation with after seeing an example of their usage. See Descriptors How To for a more thorough and explanatory introduction on the descriptor protocol.

 import socket

 from peripherals import Sensor, Actuator


 class IrrigationSystem:
     pressure = Sensor('pressure')
     temperature = Sensor('temperature')
     light = Sensor('light')
     moisture = Sensor('moisture')
     pump = Actuator('pump', 'pump.feedback')

     def __init__(self, host):
         self.host = host
         self.connection = None

     def connect(self):
         self.connection = socket.socket(socket.AF_INET, 
                                         socket.SOCK_STREAM)
         self.connection.connect((self.host, 8080))

     def close(self):
         self.connection.close()

     def send_command(self, device, value):
         command = 'set {device} {value}'
                   .format(device=device, value=value)

         self.connection.send(command)
         response = json.loads(self.connection.recv(1024))

         assert response['accepted'],
             'Command receipt indicates "{}" was not accepted' \
             .format(response['command'])

         return response

     def get_value(self, sensor):
         request = 'get {device}'.format(device=sensor)
         self.connection.send(request)
         data = irrigation.recv(1024)

         return json.loads(data)

Okay, the Sensor and Actuator classes are descriptors and they’ll perform the same functions as their associative properties in the first attempt at creating a IrrigationSystem class. They behave like an attribute.

 from interface import IrrigationSystem
 irrigation = IrrigationSystem(10.1.2.3)
 irrigation.light
 # 12.3
 irrigation.pressure
 # 14.7
 irrigation.pump = 1
 # woops, water on the floor now.

The first thing that jumps out is that the descriptors are class attributes rather than instance attributes. This means all instances of the IrrigationSystem class will share the descriptor. In most usages of descriptors this means we have to add track values for difference instances of the class.

This approach is essentially an extension of the Borg design pattern. Instead of each instance (for a specific host) sharing state we are instead retrieving the state from a single physical object. In essence, each instance is sharing the same state.

Fortunatley our API is stateless. All of the current values for the different sensors are coming from the irrigation system itself. It doesn’t matter if we have multiple instances of the IrrigationSystem class. They can even be connected to different hosts. To understand why, we’ll have to inspect the definition of our descriptors now.

 class Sensor(object):
     """Represents a device that is sensing some physical phenomena."""
     def __init__(self, name=None):
         self.name = name
 
     def __get__(self, instance, owner):
         return instance.get_value(self.name)
 
     def __set__(self, instance, value):
         raise NotImplemented
 
 
 class Actuator(object):
     """Represents a device that actuates (e.g. pump)."""
     def __init__(self, name=None, feedback=None):
         self.name = name
         self.feedback = feedback
 
     def __get__(self, instance, owner):
         return instance.get_value(self.feedback)
 
     def __set__(self, instance, value):
         instance.send_command(self.name, value)
 
 
 class PopulateDeviceNames(type):
     def __new__(meta, name, bases, class_dict):
         for key, value in class_dict.items():
             if isinstance(value, Sensor):
                 value.name = key \
                              if value.name is None \
                              else value.name
             if isinstance(value, Actuator):
                 value.name = key \
                              if value.name is None \
                              else value.name
                 value.feedback = key + '.feedback' \
                    if value.feedback is None  \
                    else value.name
 
         cls = type.__new__(meta, name, bases, class_dict)
         return cls

When the interpreter executes one of the descriptor methods it will pass in the class instance (and the class in the case of __set__). In our case, an instance of the IrrigationSystem class. The descriptor itself doesn’t need to know what instance it has, what host to talk to, or even how to retrieve the value. All it does is call one of the methods from IrrigationSystem with the name of the device the descriptor represents. Pretty neat, eh? Now those twenty new sensors only require creating twenty new attributes on the IrrigationSystem class. You can even argue that you don’t need any additional unit tests as long as the descriptor classes them self are unit tested. They only marshal some device name to another function. Test cases using this interface will be unencumbered by unnecessary details. Even without any comments the intention of the test case below is clear.

 import pytest
 from interface import IrrigationSystem

 @pytest.fixture
 def irrigation():
    return IrrigationSystem('10.1.2.3')

 def test_ambient_sensor_values(irrigation):
     pressure = irrigation.pressure
     assert pressure > 14.5 and pressure < 14.8

     temperaure = irrigation.temperature
     assert temperaure > 70.0 and temperature < 71.0

     humidity = irrigation.humidity
     assert humidity > 0.10 and humidity < 0.60

Descriptors may not always be suitable. When you encounter difficulty in expressing parts of the system with descriptors, you can always rely on falling back to using properties. The outward interface will be equivalent. One of the beautiful things of Python is that we can use simple attributes, properties, or descriptors without changing the interface.