Flask JSON Input Validation

We’ve created a simple Python microservice using Flask in Part 1. Using that as a starting point, let’s take a quick side-trip through input validation before we deploy the whole thing on Kubernetes.

Code for this can be found on GitHub. Or you can use this template as a starting point.

Input validation

We don’t have any validation yet, so let’s add that in. We want to have a 400 error of some sort, so let’s write a test for it:

# src/tests/app/test_validation.py

def test_hello_requires_greetee(client):
   request_payload = {}
   response = client.post("/hello", json=request_payload)
   result = response.get_json()

   assert response.status_code == 400
   assert result is not None
   assert "message" in result
   assert "is a required property" in result['message'][0]

This test doesn’t pass yet. In our case, we’re using json, and there are many ways input can go wrong—missing keys, bad types, weird structure. So let’s come up with a json schema to validate our greeting input. The flask extension flask_input can validate incoming json against the python jsonschema library.

This comes from the suggestions on validating API errors:

# /src/app/invalid_usage.py
class InvalidUsage(Exception):
    status_code = 400

    def __init__(self, message, status_code=None, payload=None):
        Exception.__init__(self)
        self.message = message
        if status_code is not None:
            self.status_code = status_code
        self.payload = payload

    def to_dict(self):
        rv = dict(self.payload or ())
        rv['message'] = self.message
        return rv

Then let’s register an error handler in main.py

# src/app/main.py

@app.errorhandler(InvalidUsage)
def handle_invalid_usage(error):
   response = jsonify(error.to_dict())
   response.status_code = error.status_code
   return response

Then rewrite our hello-handler to validate the input

# src/app/main.py

@app.route("/hello", methods=['POST'])
def hello() -> str:
   errors = validate_greeting(request)
   if errors is not None:
       print(errors)
       raise InvalidUsage(errors)
   greetee = request.json.get("greetee", None)
   response = {"message": say_hello_to(greetee)}
   return jsonify(response)

You’ll see that 400 errors can be created by raising an InvalidUsage exception, which is processed by our handler.

Flask_inputs is a bit behind in maintenance—I get some Deprecation warnings when I run it:

tests/app/test_main.py::test_hello_greets_greetee
  C:\Users\MikeBridge\venv\pythondemo\lib\site-packages\flask_inputs\inputs.py:37: DeprecationWarning: Using or importing the ABCs from 'collections' instead of from 'collections.abc' is deprecated, and in 3.8 it will stop working
    elif isinstance(input, collections.Iterable):

Ugh. Let’s get rid of the deprecation warning:

python -W ignore::DeprecationWarning -m pytest -s

Lastly, let’s implement the validation. Have a look at this and this for more information on how the json schema works:

# src/app/validation.py

from flask_inputs import Inputs
from flask_inputs.validators import JsonSchema

# https://pythonhosted.org/Flask-Inputs/#module-flask_inputs
# https://json-schema.org/understanding-json-schema/
# we want an object containing a required greetee  string value
greeting_schema = {
   'type': 'object',
   'properties': {
       'greetee': {
           'type': 'string',
       }
   },
   'required': ['greetee']
}


class GreetingInputs(Inputs):
   json = [JsonSchema(schema=greeting_schema)]


def validate_greeting(request):
   inputs = GreetingInputs(request)
   if inputs.validate():
       return None
   else:
       return inputs.errors

You could make this more succinct by triggering the validation as a method decorator, but I think it’s easier to unit test if it’s more explicit. Let’s test our schema out by creating a valid request via the fixture create_valid_greeting_request, then altering it in various ways to test different inputs:

# src/app/tests/conftest.py

import pytest
from app import main

# ...

@pytest.fixture()
def create_valid_greeting_request():
    """
    Helper function for creating a correctly-structured
    json request
    """
    def _create_valid_greeting_request(greetee="fixture"):
        return {
            "greetee": greetee
        }
    return _create_valid_greeting_request

# src/tests/app/test_validation.py

import pytest
from flask import Flask, request

from app.validation import validate_greeting

app = Flask(__name__)


@pytest.mark.parametrize("params", [
   {"greetee": 1},
   {"greetee": ["array"]}
])
def test_invalid_types_are_rejected(params, create_valid_greeting_request):
   json_input = create_valid_greeting_request(**params)
   with app.test_request_context('/', json=json_input):
       errors = validate_greeting(request)
       assert errors is not None


@pytest.mark.parametrize("required_parm_name", ["greetee"])
def test_missing_required_params_is_rejected(required_parm_name, create_valid_greeting_request):
   json_input = create_valid_greeting_request()
   del json_input[required_parm_name]
   with app.test_request_context('/', json=json_input):
       errors = validate_greeting(request)
       assert errors is not None


def test_valid_greetee_is_accepted(create_valid_greeting_request):
   json_input = create_valid_greeting_request(greetee="Tester")
   with app.test_request_context('/', json=json_input):
       errors = validate_greeting(request)
       assert errors is None

Our flask app is pretty much done at this point. Next, let’s dockerize it and ship it via Kubernetes.