How to develop and test a minimal API in Python with Flask

Python Software engineering Flask

11 April, 2023

tl;dr

This post describes how to create a minimal (albeit tested) API in Python using Flask. Inspiration for this project comes from a Udacity course on the same subject. The key difference is that this project is more minimal, with fewer dependencies: There is no React front-end, and we use SQLite instead of PostgreSQL for the database.

The API serves trivia questions.

The full code for this project can be found here. The Flask documentation is definitely worth a look, too.

Project setup

There are no OS dependencies outside of Python 3.x. Create the project folder; create and activate the virtual environment; and install Flask:

mkdir flask-api
cd flask-api
python3 -m venv .env
source .env/bin/activate
pip install flask

Project layout

By convention, our top-level flask-api folder will need flaskr and tests subdirectories. flaskr is where our application code will live, inside a classic __init__.py file:

mkdir flaskr
mkdir tests
cd flaskr
touch __init__.py

Let's use the example from the official documentation to verify that everything is installed and working correctly. Add the following code to __init__.py and save it:

# flask-api/flaskr/__init__.py
from flask import Flask

app = Flask(__name__)

@app.route("/")
def hello():
    return "Hello, World!"

To run the app, make sure you're in the root flask-api directory (within the activated virtual environment), and then run the following from the terminal:

flask --app flaskr run --debug

This tells Flask to run the flaskr app (which is what we've named out API), in debug mode. You should see something like the following:

In your browser, navigate to 127.0.0.1:5000/ and verify that you see the "Hello, World!" string returned on the page.

Test setup

We'll be using the unittest framework for testing, which is included in the Python standard library. In the tests folder, create a file test_api.py, and add the following:

# flask-api/tests/test_api.py
import unittest
from flaskr import HelloWorldTest

class TestAddition(unittest.TestCase):
    def test_add_two(self):
        self.assertEqual(HelloWorldTest.add_two(1), 3)

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

This is a hello-world-style test—we're testing that addition works, and that we can import a function from the flaskr module. Save the file.

We can use unittest's test discovery feature to find all tests from the root flask-api directory. This will find all tests associated with a Python module or package. So, we need to make the tests directory a module, and we do so simply by creating an empty init file: touch tests/__init__.py.

We can then run all tests with python3 -m unittest discover. If we do so now, we should get the following error:

ImportError: cannot import name 'HelloWorldTest' from 'flaskr'

Of course, that's because we haven't created the HelloWorldTest class or the associated add_two function yet! Let's create those now. In flaskr/__init__.py, add the following code:

class HelloWorldTest:
    def add_two(x):
        return x + 2

Simple enough! Now, when we run the tests, it should return something like:

Ran 1 test in 0.000s

OK

This indicates that the tests have all passed successfully, and that 1 + 2 does indeed equal 3.

We've done things this way for a reason: to demonstrate a test-driven-development workflow. This jist is that you write a failing test first, describing what you expect to happen, and then you write the actual code that causes the tests to pass. This ensures that you're maintaining good test coverage of your code base, and also helps to focus your code into specific, testable units.

Create the application factory

Now we're going to refactor flaskr/__init__.py to make use of a Flask application factory. Instead of creating the Flask instance globally, we create it inside a factory function. We also create a database parameter, which we will eventually use to specify a test database in our tests. Replace the __init__.py code with the following:

# flask-api/flaskr/__init__.py
import os
from flask import Flask

def create_app(test_config = None, prod = True):
    app = Flask(__name__)

    # Add configuration keys
    app.config.from_mapping(
        PROD_DATABASE = os.path.join(app.instance_path, "flaskr.sqlite"),
        TEST_DATABASE = os.path.join(app.instance_path, "test-flaskr.sqlite"),
    )

    # Ensure the instance folder exists
    try:
        os.makedirs(app.instance_path)
    except OSError:
        pass

    @app.route("/")
    def hello():
        return "Hello, World!"

    return app

Now, when we run the app with flask --app flaskr run --debug, we should see the same thing we did before.

Database setup

We'll use SQLite (via the sqlite3 library) for the database—it's lightweight and doesn't require us to install any OS dependencies. It's also included in the Python standard library so we don't need to install it separately.

Create a file flaskr/db.py, and add the following code:

# flask-api/flaskr/db.py
import sqlite3
import click
from flask import current_app, g

def get_db(prod = False):
    """
    Return a connection to a sqlite database. Set prod = True for the production
    database, or prod = False (the default) for the test database.
    """
    if prod:
        db_name = current_app.config["PROD_DATABASE"]
    else:
        db_name = current_app.config["TEST_DATABASE"]
    if "db" not in g:
        g.db = sqlite3.connect(
            db_name,
            detect_types = sqlite3.PARSE_DECLTYPES
        )
        g.db.row_factory = sqlite3.Row
    return g.db

def close_db(e = None):
    db = g.pop("db", None)
    if db is not None:
        db.close()

def init_db(prod = False):
    """
    Initialise the production or test database. Set prod = True for the 
    production database, or prod = False (the default) for the test database.
    """
    db = get_db(prod = prod)
    with current_app.open_resource("schema.sql") as f:
        db.executescript(f.read().decode("utf8"))

@click.command("init-test-db")
def init_test_db_command():
    """
    Drop existing and create new tables in the test database.
    """
    init_db()
    click.echo("Initialized the test database.")

@click.command("init-prod-db")
def init_prod_db_command():
    """
    Drop existing and create new tables in the production database.
    """
    init_db(prod = True)
    click.echo("Initialized the production database.")

def init_app(app):
    app.teardown_appcontext(close_db)
    app.cli.add_command(init_test_db_command)
    app.cli.add_command(init_prod_db_command)      

Have a look at the documentation for an explanation of what's going on here. We've slightly modified the original code to allow us to initialise and connect to a test and a production database separately. By default, we'll use the production database when running the application, and the test database when running the test client.

Next, create a file flaskr/schema.sql, and add the following:

# flask-api/flaskr/schema.sql
DROP TABLE IF EXISTS category;
DROP TABLE IF EXISTS question;

CREATE TABLE category (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    type TEXT UNIQUE NOT NULL
);

CREATE TABLE question (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    category_id INTEGER NOT NULL,
    question TEXT UNIQUE NOT NULL,
    answer TEXT NOT NULL,
    difficulty INTEGER NOT NULL,
    FOREIGN KEY (category_id) REFERENCES category (id)
);

INSERT INTO category (type)
VALUES
    ("Science"),
    ("Art"),
    ("Geography"),
    ("History"),
    ("Entertainment"),
    ("Sports");

INSERT INTO question (category_id, question, answer, difficulty)
VALUES
    (1, "What is the heaviest organ in the human body?", "The liver", 4),
    (1, "Who discovered penicillin?", "Alexander Fleming", 4),
    (1, "Hematology is a branch of medicine involving the study of what?", "Blood", 2),
    (2, "Which Dutch graphic artist, initials M C, was a creator of optical illusions?", "Escher", 1),
    (2, "La Giaconda is better known as what?", "Mona Lisa", 3),
    (2, "How many paintings did Van Gogh sell in his lifetime?", "One", 4),
    (2, "Which American artist was a pioneer of Abstract Expressionism, and a leading exponent of action painting?", "Jackson Pollock", 2),
    (3, "What is the largest lake in Africa?", "Lake Victoria", 2),
    (3, "In which royal palace would you find the Hall of Mirrors?", "The Palace of Versailles", 3),
    (3, "The Taj Mahal is located in which Indian city?", "Agra", 2),
    (4, "Whose autobiography is entitled 'I Know Why the Caged Bird Sings'?", "Maya Angelou", 2),
    (4, "What boxer's original name is Cassius Clay?", "Muhammad Ali", 1),
    (4, "Who invented peanut butter?", "George Washington Carver", 2),
    (4, "Which dung beetle was worshipped by the ancient Egyptians?", "Scarab", 4),
    (5, "What movie earned Tom Hanks his third straight Oscar nomination, in 1996?", "Apollo 13", 4),
    (5, "What actor did author Anne Rice first denounce, then praise in the role of her beloved Lestat?", "Tom Cruise", 4),
    (5, "What was the title of the 1990 fantasy directed by Tim Burton about a young man with multi-bladed appendages?", "Edward Scissorhands", 3),
    (6, "Which is the only team to play in every soccer World Cup tournament?", "Brazil", 3),
    (6, "Which country won the first ever soccer World Cup in 1930?", "Uruguay", 4);

This will be the basic database schema for our trivia API: a question table where the questions and answers will live, and with a foreign key to a category table that describes the type of question (e.g., sports, history, etc.).

Finally, add the following to the create_app function in flaskr/__init__.py:

from . import db
db.init_app(app)

Okay, let's initialise our databases. In the terminal, run the following:

flask --app flaskr init-test-db
flask --app flaskr init-prod-db

If everything has worked, we should see Initialized the <production|test> database. returned. There should also now be the database files at flask-api/flaskr/instance/flaskr.sqlite and flask-api/flaskr/instance/test-flaskr.sqlite.

Test the database

At this point, we should verify that we can access the database from within the Flask app. In db.py, we've defined a function get_db, which returns a database connection object. We use that to connect to the database and run SQL queries. In the create_app function, add the following route function:

# flask-api/flaskr/__init__.py
def create_app(test_config = None, prod = True):
    # ...

    @app.route("/questions")
    def get_questions():
        conn = db.get_db(prod = prod)
        cur = conn.cursor()
        res = cur.execute("SELECT question FROM question").fetchone()
        conn.close()
        return f"Question: {res[0]}"

    # ...

    return app

This defines an endpoint at /questions, that returns a single question from the question table. SQLite returns objects in rows, but because we only ask for one row (with fetchone), we can index by res[0], returning the first element—in this case, the only element—which is the question text. Verify that it works by navigating to 127.0.0.1:5000/questions; you should see the text of a question being returned.

JSON response

As is often the case, we'd like our API to return JSON. Flask includes a jsonify function that we can use for this purpose. In __init__.py, import that function, and modify our questions endpoint to fetch all questions, and return a JSON response.

# /flask-api/flaskr/__init__.py
from flask import Flask, jsonify
# ...
def create_app(test_config = None, prod = True):
    # ...

    @app.route("/questions")
    def get_questions():
        conn = db.get_db(prod = prod)
        cur = conn.cursor()
        res = cur.execute("SELECT * FROM question").fetchall()
        conn.close()

        return jsonify({
            "success": True,
            "number_of_questions": len(res)
        })

    # ...

    return app

Navigate to the 127.0.0.1:5000/questions endpoint, and you should see something like this (depending on your browser):

Great—we're returning JSON now.

At this point, we should also verify that we can return the same response via curl. We should see the same sort of response after running the following:

curl 127.0.0.1:5000/questions

Let's modify our response to actually include the question data. To do so, we're going to add a couple of helper functions. The first, format_question, is going to convert the SQL representation of a question object into a Python dictionary (so we can easily encode it as JSON). The second function, paginate_questions, will take a list of question objects, and return a formatted, specific subset of those questions. We place these functions in __init__.py, before the create_app definition:

def format_question(question):
    """
    Take a SQL representation of a question, and return a dictionary.
    """
    return {
        "id": question[0],
        "category_id": question[1],
        "question": question[2],
        "answer": question[3],
        "difficulty": question[4]
    }

def paginate_questions(questions, page = 1):
    """
    Return a formatted list of 5 questions, based on the requested page.
    """
    QUESTIONS_PER_PAGE = 5
    # Define the indices of the first and last questions to return
    start = (page - 1) * QUESTIONS_PER_PAGE
    end = start + QUESTIONS_PER_PAGE
    # Format the selected questions
    formatted_questions = []
    for question in questions[start:end]:
        formatted_questions.append(format_question(question))
    return formatted_questions

Now, update our /questions route (noting that we need to add an extra import, request):

from flask import Flask, jsonify, request
# ...
@app.route("/questions")
def get_questions():
    # Get all questions
    conn = db.get_db(prod = prod)
    cur = conn.cursor()
    res = cur.execute("SELECT * FROM question").fetchall()
    conn.close()

    # Get the requested page (defaults to 1)
    page = request.args.get("page", 1, type = int)

    # Subset the questions 
    questions = paginate_questions(res, page)

    return jsonify({
        "success": True,
        "number_of_questions": len(res),
        "questions": questions,
    })

We now return the full detail of five questions when we hit our /questions endpoint. You'll notice that we also have an optional parameter, page. This defaults to 1, i.e., return the first page (or the first five questions). We can also specify this in the request. For example, you'll get different questions returned with the following two requests:

curl 127.0.0.1:5000/questions
curl 127.0.0.1:5000/questions?page=2

Back to the tests

At this point, let's start writing some tests—for the existing endpoint, and for endpoints that we'd like to create next. Replace the code in test_api.py with the following:

import unittest
from flaskr import create_app
import json

class QuestionsTestCase(unittest.TestCase):
    def setUp(self):
        """
        Set up the test client, using the test database.
        """
        self.app = create_app(prod = False)
        with self.app.app_context():
            db.init_db()
        self.client = self.app.test_client

    def tearDown(self):
        pass

    def test_get_questions(self):
        """
        Test GET /questions
        """
        res = self.client().get("/questions")
        data = json.loads(res.data)
        self.assertEqual(res.status_code, 200)
        self.assertTrue(data["success"])
        self.assertEqual(data["number_of_questions"], 19)
        self.assertEqual(len(data["questions"]), 5)

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

The setUp and tearDown functions are run at the beginning and end of the class's test cases; in this case, we initialise a test client for our tests to run against. Our test_get_questions function makes a GET request to the /questions endpoint, and we are testing that a HTTP 200 status code is returned; that the success key has a value of True; that the number_of_questions key has a value of 19 (which is the number of questions we originally added in schema.sql); and that the number of elements in the questions key is five (which is the number we have set to return per page).

Run python -m unittest discover to verify that everything's working as expected.

More endpoints and tests

Let's add a test case for an endpoint that allows us to look at a specific question:

def test_get_question(self):
    """
    Test GET /questions/. We expect a HTTP 200 response, with JSON data
    including {
        "success": True,
        "question": {
            "id": ,
            "question": , 
            "answer": ,
            "category_id": ,
            "difficulty":     
        }
    }.
    """
    res = self.client().get("/questions/1")
    data = json.loads(res.data)
    self.assertEqual(res.status_code, 200)
    self.assertTrue(data["success"])
    self.assertEqual(data["question"]["id"], 1)
    self.assertEqual(data["question"]["question"], "What is the heaviest organ in the human body?")
    self.assertEqual(data["question"]["answer"], "The liver")
    self.assertEqual(data["question"]["category_id"], 1)
    self.assertEqual(data["question"]["difficulty"], 4)

When we hit 127.0.0.1:5000/questions/<id>, where <id> is an integer corresponding to the ID of a question, we want to return the entire question as JSON (as well as the usual "success" response). We implement this in the following way:

@app.route("/questions/")
def get_question(id):
    # Get the specific question by ID
    conn = db.get_db(prod = prod)
    cur = conn.cursor()
    res = cur.execute(
        f"SELECT * FROM question WHERE id == {id}"
    ).fetchall()
    conn.close()
    
    # If there is no question at the requested ID,
    # abort with a 404 resource not found
    if len(res) == 0:
        abort(404)

    # Format the question
    formatted_question = format_question(res[0])

    # Return the response
    return jsonify({
        "success": True,
        "question": formatted_question,
    })

Note that we need to add another Flask import, abort. This allows us to raise a specific HTTP error code, rather than just an internal server error. Test it out at 127.0.0.1:5000/questions/1 (which should look good), and 127.0.0.1:5000/questions/100 (which should raise a HTTP 404 error, because that question doesn't exist.)

Return errors as JSON

The final thing we want to do in this tutorial is to make sure that any errors we raise are also being returned as JSON. This makes the API much friendlier for clients to use, since they can easily parse error messages if they're in the same format as the expected response. To do so, we use the @app.errorhandler() decorator. In create_app, add the following:

@app.errorhandler(404)
def resource_not_found(error):
    return jsonify({
        "success": False,
        "error": 404,
        "message": "Resource not found"
    })

Now when we navigate to 127.0.0.1:5000/questions/100, we should see a JSON-formatted response:

That's it!

The full code for this project is here; that repo has more endpoints and tests than those that I've described here, as well as some other small differences (such as logging messages); but I'm sure you get the idea by now. Any questions or comments, feel free to add below.

---