Schema is a library for validating Python data structures. In this tutorial, we'll be using schema to validate the body of the requests coming to our endpoints.
Context:
Imagine, an endpoint in one of our SaaS APIs expects an email address and a username that has some constraints as input, then does some business logic before returning a response.
How do we ensure that the input conforms to the constraints we set? Of course, we can extract the email_address and the username field from the request body, run some kind of regex validation on them and figure what to do after.
I'll be sharing with you what I believe is I better way of handling this kind of scenario using the Schema library.
Project setup:
I'll be using the same setup we ended up with in this blog post.
Here's how the project tree looks like so far:
myflaskapp
venv/
app/
__init__.py
routes.py
.flaskenv
run.py
Adding a new route:
Just like in the previous post, we are going to add a new route to the routes.py module. I'll call the route handler "service".
routes.py should look like this now:
from flask import request
from app import app
@app.route('/')
def home():
return {'success': True, 'message': 'Hello Flask!'}
@app.route('/service', methods=['POST'])
def service():
return {'success': True}The request object:
Flask provides us with an object called "request", this object basically represents the incoming request from a high-level. The request object gives us direct access to information about the request: headers, cookies, body, etc...
The "service" endpoint expects a request body in JSON format, that has two fields: email_address and username.
Let's first update the "service" route handler to take into account the JSON format constraint. The "request" object implements a useful property called "is_json" which is a boolean that holds a value of True if the request has a valid JSON body and False if not.
Update routes.py as follows:
from flask import request
from app import app
@app.route('/')
def home():
return {'success': True, 'message': 'Hello Flask!'}
@app.route('/service', methods=['POST'])
def service():
if not request.is_json:
return {'success': False}
return {'success': True}line 2: we imported the request object from flask.
line 3-4: we check if the request has a valid JSON object, the 400 in the return value means that the response will have a status of Bad Request.
To send a POST request to the endpoint, I'll be using cURL:
Valid JSON:
curl -i -H "content-type: application/json" -X POST http://127.0.0.1:5000/service
Invalid JSON:
curl -i -H -X POST http://127.0.0.1:5000/service
Great, now our endpoint returns a different response based on the request content-type.
It's common that REST APIs expose multiple endpoints that accept only JSON inputs, so instead of writing the same "if check" for all the handlers that we could potentially have, It might be helpful to create a decorator that checks if the request has a valid JSON body and apply it each time we want to force the JSON constraint on an endpoint.
Create a new Python module called decorators.py in the app/ directory and add the following:
from flask import request
from functools import wraps
def is_json(f):
@wraps(f)
def wrapper(*args, **kwargs):
if not request.is_json:
return {'success': False}, 400
return f(*args, **kwargs)
return wrapperThis is_json decorator will wrap a route handler, in this example the "service" handler, and will check if the request has a valid JSON body.
Import the is_json decorator and update the service handler in routes.py :
from app.decorators import is_json
@app.route('/service', methods=['POST'])
@is_json
def service():
return {'success': True}Request body schema validation:
We'll be using the Schema library to validate the request body.
Install Schema with pip:
Before defining our scheme, let's install the Schema library
(venv) $ pip3 install schema
Defining the schema:
Since this is a really small app, I'll define the schema in the decorators.py module because we are going to write a new decorator that uses this schema to validate the request, but feel free to create a new module if you want to.
Our constraints are as follows:
email_address: Should be a valid email address.
username: Should start with a letter, and be at least 3 characters long.
The Schema library provides a wide range of options to use for validating a data structure, including regex validation, which I'll be using in this example.
Since this tutorial is more about the request body validation, I'm going to use two fairly simple regex patterns, which are good enough for this example but don't cover edge cases.
email_address:
'^[a-z0-9]+@[a-zA-Z0-9-]+\.[a-zA-Z]{2,6}$'username: '^\[a-zA-Z]{1}[a-zA-Z0-9_.-]{2,}'
Let's use those two patterns as our schema. In decorators.py add the following:
from schema import Schema, Regex, SchemaError
.
.
.
schema = Schema({
'email_address': Regex(r'^[a-z0-9]+@[a-zA-Z0-9-]+\.[a-zA-Z]{2,6}$'),
'username': Regex(r'^\[a-zA-Z]{1}[a-zA-Z0-9_.-]{2,}')
})
We import the Schema class from the schema library.
We create a new instance of the Schema class.
Schema expects the first parameter to be the structure of our data-structure, in our case, it's a dictionary because when the body request is valid JSON, Flask parses the body on the fly into a dictionary and makes it available for us in the request.json property.
Our schema is going to be a dictionary where the keys are the field names in the JSON body, and the values are the schema constraints.
Regex is a class provided by the schema library, that we instantiate by passing the regex pattern as the first parameter.
Schema will then check if each field in our parsed JSON body conforms to its constraints.
Validation decorator:
We'll create now a new decorator that uses the schema instance we just created to validate incoming requests.
Just after the schema instance, add the following:
def validate_json(f):
@wraps(f)
def wrapper(*args, **kwargs):
try:
schema.validate(request.json)
return f(*args, **kwargs)
except SchemaError as s:
return {'success': False, 'message': str(s)}, 400
return wrapperInside of our decorator, we use the validate method that the schema instance implements to validate the request.json dictionary against the schema constraints.
The validate method raises a SchemaError if the object we are checking is invalid, therefore we are using the try-except error handling to call the validate method.
Finally, let's apply this decorator to our route handler.
Update the service handler in routes.py:
@app.route('/service', methods=['POST'])
@is_json
@validate_json
def service():
return {'success': True}Let's test our endpoint again using cURL:
Valid request body:
Invalid username field in the request body:
Empty request body:
That is it, our endpoint now only accepts requests where the body is a valid JSON and also conforms to the schema we specified.