Building REST APIs With Python and Flask: Part 2

This is the second part of a three-part tutorial on creating REST APIs with Python and Flask. In this part, you will learn how to connect to a database and how to different types of HTTP requests.

Here are the links to other parts of this tutorial.

  1. Part 1
  2. Part 2(You are here)
  3. Part 3

Connecting to a Database

We will be using a single table in our application to store the details of our users. Let us define the schema for that table first.

Create a file schema.sql inside the inner profiler directory and put the following contents in it.

DROP TABLE IF EXISTS user;

CREATE TABLE user (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    first_name TEXT NOT NULL,
    last_name TEXT NOT NULL,
    email TEXT UNIQUE NOT NULL,
    password TEXT NOT NULL,
    created_on TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
  • The first line drops the table if it is already present.
  • id is our primary key. It is set to auto-increment so that we do not have to set this manually.
  • email has a unique constraint because we will be using the email to identify the user.
  • first_name, last_name and password are non nullable text fields. Their values cannot be empty.
  • created_on is a timestamp field for storing the account creation time. It has the default value of the current timestamp so the database will put in the creation time for us.

Now we need to connect to a database from python and run our schema file to create the user table.

We will be using the SQLite database. Python standard library comes with SQLite so you do not have to install anything else. In a production system, you should use a fully-featured database server like PostgreSQL or MySQL.

Open the config.py file we created in the last chapter and update it to include a database name.

SECRET_KEY = "change-this-please"

DATABASE_NAME = "db.sqlite3"

Create a new file, db.py, in the inner profiler directory and add the following code.

import sqlite3

import click
from flask import current_app, g
from flask.cli import with_appcontext


def get_db():
    if "db" not in g:
        g.db = sqlite3.connect(current_app.config["DATABASE_NAME"])
        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()

Ignore the imports click, current_app and with_appcontext for now. We will use them later in the chapter. Here is what is going on in the rest of the code.

  • The get_db is a helper function that returns a connection to a SQLite database.
  • g is a special object that is global to an app context. An object like a database connection might be used from different functions. Instead of creating this every time a function needs it, we create it the first time when a function asks for it and stores it in the g object. During subsequent calls, we take the value from g and return it directly.
  • We use sqlite.Row to get the query results in dictionary format. By default, SQLite will return tuples. Dictionaries are much easier when working with APIs.
  • In close_db function, we check whether a db object is created and if it is created, we close the connection.
  • close_db function has a keyword argument e but we are not using it. This is because flask will call the close_db function at the end of a request and flask might pass an error value with the call. We will get an error if we do not accept the error from flask.

Next, we need to load our schema into the database to create tables. We could do it manually but flask gives us a much cleaner way to do operations like this. We will add a new command that can create the schema by reading the SQL file. We can then run this command just like the command we use to run the development server and it will create the table for us.

Add the following code to the end of db.py file. Do not delete the code that is already in the file.

def init_db():
    db = get_db()

    with current_app.open_resource("schema.sql") as f:
        db.executescript(f.read().decode("utf8"))


@click.command("initialise-database")
@with_appcontext
def initialise_database_command():
    init_db()
    click.echo("Successfully created database schema.")
  • The init_db function calls our get_db function from previous section to get a database connection.
  • We use current_app.open_resource to open the schema file instead of opening it ourselves. current_app.open_resource will handle the file path differences so we do not have to worry about the correct path to the file.
  • We use the executescript method in sqlite to run the statements in our file. open_resource opens the file in binary mode so we need to convert it to a string before executing. We use decode for that.
  • Click is a package we use for creating custom commands. Click gets installed when we install flask so no extra installation is required.
  • click.command decorator registers a new command with click. The string we pass to this function will become the command name. When we run flask initialise-database, click will call the initialise_database_command function.
  • with_appcontext decorator tells flask that this function needs an app context. Without an app context, utilities like open_resource won’t work.
  • echo is similar to print. We use it to display a message if the command is successful.

Now, we need to tell flask about the command we created and also about our close_db function. We need to use the flask object to do this but we do not have a flask app object in db.py. The app object is in our __init__ file. We will create a function that takes the app as an argument and register our commands. Later we will call this function from our __init__ file and pass the app object.

Add the following code to the end of db.py file.

def init_app(app):
    app.teardown_appcontext(close_db)
    app.cli.add_command(initialise_database_command)
  • We use the teardown_appcontext tell flask to call close_db function at the of an application context. teardown_appcontext is useful for closing the resources we opened in the app such as network connections, database connections etc.
  • app.cli.add_command method adds a new command to flask app.

Next step is to call the init_app function and pass an app object. For this, we need to update our create_app function inside __init__.py. Open profiler/__init__.py and add the following code after the line app.config.from_pyfile("config.py", silent=True)

from . import db
db.init_app(app)

We can also remove the hello function. Your __init__.py file should look like the following.

from flask import Flask


def create_app():
    app = Flask(__name__)

    app.config.from_pyfile("config.py", silent=True)

    from . import db
    db.init_app(app)

    return app

It is time to test our custom command for creating a database and table. Open a terminal from the outer profiler directory and activate pipenv shell. Then run our command.

flask initialise-database

You should see the success message printed into the terminal.

Blueprints

When we added the hello-world API, we did it inside the create_app function. But if we have several APIs, putting everything inside the create_app function is not a feasible solution. Flask has a feature called blueprints which helps us to organise our APIs into multiple modules.

Our API is for managing users in our application. We will implement the following endpoints and methods for that.

Endpoint Method Functionality
/users/ POST Create new user
/users/ GET Get the details of all the users
/users/user-id/ GET Get the details of a single user matching the user-id
/users/user-id/ PUT Update the details of the users matching the user-id
/users/user-id/ DELETE Delete the user matching the user-id

Create a new file users.py inside the inner profiler directory and put the following content in it.

from flask import Blueprint

blueprint = Blueprint("users", __name__, url_prefix="/users")


@blueprint.route("/", methods=["POST"])
def create_user():
    return {"action": "create user"}, 201
  • We created an object of the Blueprint class. The first argument is the name of the blueprint. We named it as users. Just like the app object, the blueprint needs to know the module name. We passed the __name__ for that. The third argument, url_prefix is optional. Since all the APIs we define in this blueprint has their URLs starting with /users, we decided to define it as a prefix instead of repeating it with every function.
  • blueprint.route is a decorator similar to app.route. We specified the url as / since the blueprint will add /users part.
  • To blueprint.route, we passed a list of methods also. We need to restrict this to only the POST method since that is what clients use for creating a user. Flask will return an error if the method is not POST.
  • Instead of returning a dictionary, the create_user function is returning a tuple containing a dictionary and an integer. The second value in the tuple will be the status code for this API. Conventionally, APIs return the 201 status code if the creation is successful.

The next step is to tell flask app about the new blueprint we created. Open __init__.py file and add the following lines to the end of create_app function.

from . import users
app.register_blueprint(users.blueprint)

Your __init__.py file should look like the following.

from flask import Flask


def create_app():
    app = Flask(__name__)

    app.config.from_pyfile("config.py", silent=True)

    from . import db
    db.init_app(app)

    from . import users
    app.register_blueprint(users.blueprint)

    return app

It is time to test our new API. Open a terminal and run the flask server. Now go to postman and load http://127.0.0.1:5000/users/. Make sure you select the POST method from the dropdown. Otherwise, you will get a method not allowed error.

Let’s now add the remaining APIs. Open users.py and add the following code to the end of the file.

@blueprint.route("/", methods=["GET"])
def list_users():
    return {"action": "list all users"}


@blueprint.route("/<int:user_id>/", methods=["GET"])
def get_user(user_id):
    return {"action": f"get user {user_id}"}


@blueprint.route("/<int:user_id>/", methods=["PUT"])
def update_user(user_id):
    return {"action": f"update user {user_id}"}


@blueprint.route("/<int:user_id>/", methods=["DELETE"])
def delete_user(user_id):
    return "", 204
  • list_users is also using the same URL we used for create_user but the method is changed to GET.
  • For get, update and delete APIs, we added <int:user_id> to the URL pattern definition. What this means is that the URL should contain an integer and that integer should be mapped to user_id. Flask will pass the captured user_id to our function. That is why we have an extra user_id argument in these functions.
  • The list, get and update APIs does not have a status code in the return statement. We want these APIs to return the 200 status code. If we don’t provide a status code, flask will return a 200 status code. You can manually provide a 200 status code but it is not required.
  • The return value of delete_user is different. We are returning empty string as content and 204 as status code. By convention, the delete operation should not return any content and the status code should be 204.

Now open postman and load http://127.0.0.1:5000/users/ using the GET method. You should see the response from list_users function. To test get, update and delete we can use the URL http://127.0.0.1:5000/users/1/. 1 will be the user id. Try it with the GET method and you should see the get_user response. Now try changing the method PUT and then to DELETE.

User API

The blueprint we built in the last section does not do the actual user management. In this section, we will update the API with real functionality.

We will start with the create API since we need a user created before doing any other operations. Open our user.py file and update it with the following code. Note that we have made some changes to the import statements also.

from flask import Blueprint, jsonify, request
from werkzeug.security import generate_password_hash

from .db import get_db

blueprint = Blueprint("users", __name__, url_prefix="/users")


@blueprint.route("/", methods=["POST"])
def create_user():
    data = request.json

    first_name = data.get("first_name")
    last_name = data.get("last_name")
    email = data.get("email")
    password = data.get("password")

    if not all([first_name, last_name, email, password]):
        return {"error": "Required fields are missing"}, 400

    db = get_db()

    hashed_password = generate_password_hash(password)

    try:
        cursor = db.execute(
            "INSERT INTO user (first_name, last_name, email, password) VALUES (?, ?, ?, ?)",
            (first_name, last_name, email, hashed_password),
        )
        db.commit()
    except db.IntegrityError:
        return {"error": "This email is already registered"}, 400

    return jsonify({"id": cursor.lastrowid}), 201
  • We have imported the request object from flask to get the data user sends to the API. Our API will be accepting JSON data as input. To get the JSON data from the request, we need to access its json property.
  • data is just a dictionary so we can use the get method to extract values from the dictionary. We used the get method instead of the subscript notation because we don’t want python to throw an error if a key is not present.
  • We use the all function to make sure all of our input has a value. If any of them is null, all will return False. In that case, we will return an error with status code 400. Status code 400 means there is an error in the data sent from the client.
  • We are using the get_db function we created earlier to get a connection to our database.
  • We don’t want to store the plain text passwords in our database. We will use the generate_password_hash function from werkzeug to hash and salt the password. werkzeug is a dependency for flask so it got installed while you were installing flask. No extra installation is required.
  • We run our SQL statement using the execute method. We need to call the commit method for the cursor to update our changes in the database.
  • When executing the query, we used ? as a placeholder for the values and then supplied the actual values in a tuple. The sqlite driver will then perform the necessary sanitation in data to protect the API from attacks like SQL injection.
  • The insert statement will throw an IntegrityError if there is already a user with the email id. This is because we set the email column as unique while defining our schema. We catch the exception and return an error with the status code 400.
  • If everything completes without any errors, we get the inserted id using lastrowid property and return it.

Save this file and start the flask development server. Then go to postman and put http://127.0.0.1:5000/users/ in the url field and set the method to POST. To send data to the server, select Body and then select raw. From the dropdown select JSON. Now put something like this in the body text field.

{
  "first_name": "Harry",
  "last_name": "Potter",
  "email": "harry@example.com",
  "password": "1234"
}

Hit send and you should see the inserted id returned. Try with the same email again and see what happens. Also, try setting an empty value to any of the fields.

Let us update the list_users function from the last chapter to return the list of all users in our database.

@blueprint.route("/", methods=["GET"])
def list_users():
    db = get_db()

    cursor = db.execute("SELECT * FROM user")
    users = cursor.fetchall()

    user_list = []
    for user in users:
        user_list.append(
            {
                "id": user["id"],
                "first_name": user["first_name"],
                "last_name": user["last_name"],
                "email": user["email"],
                "created_on": user["created_on"],
            }
        )

    return jsonify(user_list)
  • We used the get_db function to get a database connection and the execute method to run our select query.
  • The fetchall method will retrieve all users matching the query. If there are no users in the database, it will return an empty list.
  • We then loop over the users' data returned from the cursor. users will be a list of sqlite Row objects. A Row object is not directly convertible to JSON even though it behaves like a dictionary. Inside the loop, we create a dictionary from the data and append it to our list. Note that we are not including the password in the new dictionary since it should never be returned to the client.

To test our new API, load http://127.0.0.1:5000/users/. Make sure you set the method to GET.

Now it is time to update get_user function.

@blueprint.route("/<int:user_id>/", methods=["GET"])
def get_user(user_id):
    db = get_db()

    cursor = db.execute("SELECT * FROM user WHERE id=?", (user_id,))
    user = cursor.fetchone()

    if user is None:
        return {"error": "Invalid user id"}, 404

    return {
        "id": user["id"],
        "first_name": user["first_name"],
        "last_name": user["last_name"],
        "email": user["email"],
        "created_on": user["created_on"],
    }
  • We used the select statement with a where clause since we need to fetch only the matching record. Note the comma after user_id. This is required to create a single element tuple.
  • We used the fetchone method since we are looking for a single row here.
  • If there is no matching record in the database, fetchone will return None. When this happens, we return an error message with status code 404. 404 for is generally used when a resource is not available in the server.
  • In the last step, we constructed a dictionary and returned it.

To test this, send a GET request to http://127.0.0.1:5000/users/1/. Try changing the user id in the URL.

Let us modify update_user function next.

@blueprint.route("/<int:user_id>/", methods=["PUT"])
def update_user(user_id):
    data = request.json

    first_name = data.get("first_name")
    last_name = data.get("last_name")
    email = data.get("email")

    if not all([first_name, last_name, email]):
        return {"error": "Required fields are missing"}, 400

    db = get_db()

    try:
        db.execute(
            "UPDATE user SET first_name=?, last_name=?, email=? WHERE id=?",
            (first_name, last_name, email, user_id),
        )
        db.commit()
    except db.IntegrityError:
        return {"error": "This email is already registered"}, 400

    return jsonify({"id": user_id}), 200

It is mostly similar to our create API with some key differences.

  • We removed the password from input fields since changing the password without extra verification steps is not a good idea.
  • We are using an update query for updating the details of an already existing user.

Testing this API is also pretty similar to the create API. Add user id to the URL and set the method to PUT.

delete_user is next in our list to update.

@blueprint.route("/<int:user_id>/", methods=["DELETE"])
def delete_user(user_id):
    db = get_db()

    db.execute("DELETE FROM user WHERE id=?", (user_id,))
    db.commit()

    return "", 204

We are not validating the user_id here since it does not matter whether a user exists with that id. If the id is invalid, the delete query will not remove anything from the database.

For testing this, use the same URL you used for testing the single user API but change the method DELETE. After deletion, call the single user API again. You should get an error with 404 status this time.

Continue to part 3 of this tutorial

Last updated on February 5, 2022
Tags: Python Flask Tutorial