Getting the weather — a full stack ReactJS/Python Flask API tutorial — Part 2

Recap

In part 1 we setup the Python Flask server, containerised it with Docker, and configured the build process. In this article we will flesh out the server to get some weather data for us to render in a ReactJS app.

Before we start, a quick addendum to the implementation of Part 1, where we hardcoded the host to start our Flask server on to ‘0.0.0.0’ to get it to run properly in a Docker container. But this has the side effect of not being very useful for debugging during development (i.e. we can not run $ python3 main.py from a shell, and hit http://0.0.0.0:5000 in the browser). To address this, we can use an environment variable to set the host to run on when in the Docker container.

So we add the following to our Dockerfile:

...
ENV FLASK_HOST=0.0.0.0
...

And the following to our main.py:

...
def main():
    hostname = os.getenv('FLASK_HOST')
    hostname = '127.0.0.1' if hostname is None else hostname
    print ("Running on %s" %(hostname))
    app.run(host= hostname)
...

Voila ! We have the best of both worlds now.

More housekeeping…

While we are at it, we should introduce some code quality measures into our build script. The first we will look at is using Pylint to get some quality metrics around our code — in a commercial setting we can parse the output and feed the results into an information radiator.

To install and run pylint …

$ pip install pylint
# From the /script/ folder...
$ pylint --output-format=parseable ../app/main.py

And add to the build.sh …

#!/bin/bash
# Validate codebase
pylint --output-format=parseable ../app/main.py
...

For testing we will use unittest — lets just copy the first example from the unittest site and include it into our code base as a placeholder for now. The test class will be in the file app/test_main.py — using the convention of prefixing test_ to the name of the source file being tested. This allows for the auto-discovery of tests throughout the project folders.

class TestStringMethods(unittest.TestCase):

    def test_upper(self):
        self.assertEqual('foo'.upper(), 'FOO')

    def test_isupper(self):
        self.assertTrue('FOO'.isupper())
        self.assertFalse('Foo'.isupper())

    def test_split(self):
        s = 'hello world'
        self.assertEqual(s.split(), ['hello', 'world'])
        # check that s.split fails when the separator is not a string
        with self.assertRaises(TypeError):
            s.split(2)

if __name__ == '__main__':
    unittest.main()

We can run this from the command line to verify our tests are run successfully ( $ python3 -m unittest -v ), and then add it to the build script (after the pylint) — with the discover subcommand, so we can specify the folder to look for tests. We also add an additional line that checks the status code returned from running our tests, so if we fail, the build process aborts.

...
# Unit Test
python3 -m unittest discover -s ../app -v
[ $? -eq 0 ]  || exit 1
...

Lets get some weather already…

The base structure of the code to retrieve the weather data from OpenWeatherMap and WeatherStack will implement a function to:

1. Request the API endpoint with the API Key
2. Map the result to our custom JSON format.
3. Cache the request — if the request fails return that data.

Let’s look at the implementation of the OpenWeatherMap service — essentially identical to the WeatherStack implementation, apart from an additional bit of processing to map the wind speed from m/s to km/hr.

'''' Functions to interact with the OpenWeatherMap service to get the weather'''
import os
import urllib.request
import json
from datetime import datetime
BASE_URL = 'http://api.openweathermap.org/data/2.5/weather?q='
last_result = None
last_ts = None
is_up = False
def fetch_weather(location=None):
    """Perform the retrieval of the weather - default location is Melbourne, AUS"""
    global is_up, last_result, last_ts
    if location is None: 
        location = "Melbourne,Victoria,Australia"
    api_key = os.getenv('OPENWEATHERMAP_API_KEY')
    if api_key is None:
        print("OpenWeatherMap API key not specified")
        return None
url = BASE_URL + location + '&units=metric&appid=' + api_key
    response = urllib.request.urlopen(url).read() 
    src = json.loads(response) 
    #Some simple validation
    if not "main" in src:
        is_up = False
        return None
    result = {
        'location': {
            'name': src['name'],
            'country' : src['sys']['country'],
            'lat': src['coord']['lat'],
            'lon': src['coord']['lon'],
            },
        'temperature' : src['main']['temp'],
        'humidity': src['main']['humidity'],
        'wind' : {
            'speed' : float(src['wind']['speed']) * 3.6,
            'deg' : src['wind']['deg'] 
        },
        'cloud' : src['clouds']['all'],
        'pressure' : src['main']['pressure'],
    }
    last_result = result
    last_ts = datetime.now()
    return result

And complementing this code is a corresponding test case, which doesn't do much just now — just the one rough validation of the structure of the data being returned matching our generic format.

''' Test cases for the OpenWeatherMap web service '''
import unittest
from . import openWeatherMap
class TestOpenWeatherMap(unittest.TestCase):
def test_fetch(self):
        ''' Simple working test '''
        res = openWeatherMap.fetch_weather()
        self.assertIsNotNone(res)
def test_format(self):
        ''' Test result format '''
        res = openWeatherMap.fetch_weather()
        print(res)
        self.assertIsNotNone(res['location'])
        self.assertIsNotNone(res['temperature'])
        self.assertIsNotNone(res['humidity'])
        self.assertIsNotNone(res['wind'])
        self.assertIsNotNone(res['cloud'])
        self.assertIsNotNone(res['pressure'])
if __name__ == '__main__':
    unittest.main()

Our app folder now looks like: (the __init__.py is empty at this stage)

├── main.py
├── requirements.txt
├── test_main.py
└── weather_services
    ├── __init__.py
    ├── openWeatherMap.py
    ├── test_openweathermap.py
    ├── test_weatherstack.py
    └── weatherStack.py

We will need to add the weather_services module to our pylint command in the build.sh script, to ensure it covers the code in there. The good news is the “unittest discover” process will automatically pickup our test cases in the module.

So we now have a module for each web service that will get the weather in our defined JSON format, but what happens if that fails ? We are already “caching” the result in the last_result variable and capturing the timestamp of the last successful result. We now need to implement a mechanism to :

• gracefully failover to the alternate service
• serve stale content (preferably the latest from whatever service went offline last)

So we wrap our urlopen call in a try/except block as follows:

try:
    response = urllib.request.urlopen(url).read() 
    is_up = True
except:
    #Dont care about the type of error - at this point
    is_up = False
    return None

Then in the main.py we need to implement a function that retrieves the weather, and inspects the state of each weather service.

def lookupWeather(location):
    ''' Perform the lookup of weather information across our two services '''
    weather = weatherStack.fetch_weather(location)
    if (not weatherStack.is_up):
        weather = openWeatherMap.fetch_weather(location)
        if (not openWeatherMap.is_up):
            if (weatherStack.last_ts is None):
                weather = None
            else:    
                if (openWeatherMap.last_ts is None):
                    weather = weatherStack.last_result
                else:     
                    #Get the latest cached
                    if (openWeatherMap.last_ts > weatherStack.last_ts):
                        weather = openWeatherMap.last_result
                    else:
                        weather = weatherStack.last_result  
     if (weather is None): weather = {'status':'Offline'}
     return weather

Which we then create a webservice endpoint to access…

@app.route('/weather')
def getWeather():
    """ Weather web service """
    location = request.args.get("location", None)
    return lookupWeather(location)

We can test this out by running our code python3 main.py and then opening a browser to (http://localhost:5000/weather) — if you want to get fancy, you can also send the location (http://localhost:5000/weather?location=Perth) , and you should see something like…

Weatherstack service

To test out the failover mechanism, try adding a dummy entry in your hosts file to map the Weatherstack URL to your localhost…

127.0.0.1 api.weatherstack.com

After which you should see the source change to OpenWeatherMap…

OpenWeatherMap service

And as a final test, disconnect your internet connection to see the old data served. The eagle eyed readers would have noticed that the lat/lon details are formatted differently between the 2 services — I’ll leave that up to the reader to address.

So we reach the end of Part 2 — looks like a got spot to draw to a close, and save up for a Part 3 where we consume the service with a ReactJS app.

Hope you have stuck with me, and the article has the right level of detail for you — if not let me know in the comments.