Building a Flask API with APIFairy

This tutorial demonstrates how to easily create a RESTful API with Flask and APIFairy.

Objectives

By the end of this tutorial, you'll be able to:

Create API endpoints in Flask using the decorators provided by APIFairy
Utilize Flask-Marshmallow to define the schemas for inputs/outputs to the API endpoints
Generate the API documentation using APIFairy
Integrate a relational database with the API endpoints
Implement basic and token authentication using Flask-HTTPAuth

What is APIFairy?

APIFairy is an API framework written by Miguel Grinberg that allows for easily creating an API with Flask.

APIFairy provides four key components for easily creating an API in Flask:

Decorators
Schemas
Authentication
Documentation

Let's explore each one in detail...

Decorators

APIFairy provides a set of decorators for defining the inputs, outputs, and authentication for each API endpoint:

APIFairy provides five core decorators:

@arguments - specifies the input arguments in the query string of the URL
@body - specifies the input JSON body as a schema
@response - specifies the output JSON body as a schema
@other_responses - specifies additional responses (often errors) that can be returned (documentation only)
@authenticate - specifies the authentication process

Schemas

The input (using the @body decorator) and output (using the @response decorator) of an API endpoint are defined as schemas:

class EntrySchema(ma.Schema):
    """Schema defining the attributes in a journal entry."""
    id = ma.Integer()
    entry = ma.String()
    user_id = ma.Integer()

The schemas utilize marshmallow for defining the data types as classes.

Authentication

The @authenticate decorator is used to check the authentication header provided in the URL request to each API endpoint. The authentication scheme is implemented using Flask-HTTPAuth, which was also created by Miguel Grinberg.

A typical API authentication approach would be to define basic authentication for protecting the route to retrieve an authentication token:

basic_auth = HTTPBasicAuth()

@basic_auth.verify_password
def verify_password(email, password):
    user = User.query.filter_by(email=email).first()
    if user.is_password_correct(password):
        return user

And also to define token authentication for protecting the majority of routes based on a time-sensitive authentication token:

token_auth = HTTPTokenAuth()

@token_auth.verify_token
def verify_token(auth_token):
    return User.verify_auth_token(auth_token)

Documentation

One of the great features of APIFairy is the beautiful API documentation that's automatically generated:

The documentation is generated based on docstrings in the source code along with the following configuration variables:

APIFAIRY_TITLE - name of the project
APIFAIRY_VERSION - version string of the project
APIFAIRY_UI - format of the API documentation

For APIFAIRY_UI, you can generate templates from one of the following OpenAPI documentation renderers:

For a full list of configuration variables available, refer to the Configuration docs.

What Are We Building?

You'll be developing a journal API in this tutorial, allowing users to keep a daily journal of events. You can find the full source code in the flask-journal-api repository on GitLab.

Key Python packages used:

Flask: micro-framework for Python web application development
APIFairy: API framework for Flask, which uses-
- Flask-Marshmallow: Flask extension for using marshmallow (object serialization/deserialization library)
- Flask-HTTPAuth: Flask extension for HTTP authentication
- apispec - API specification generator that supports the OpenAPI specification
Flask-SQLAlchemy: ORM (Object Relational Mapper) for Flask

You'll develop the API incrementally:

Create the API endpoints for working with journal entries
Generate API documentation
Add a relational database for storing the journal entries
Add authentication to guard the API endpoints

API Endpoints

Let's jump into creating an API using Flask and APIFairy...

Project Initialization

Start by creating a new project folder and a virtual environment:

$ mkdir flask-journal-api
$ cd flask-journal-api
$ python3 -m venv venv
$ source venv/bin/activate
(venv)$

Feel free to swap out virtualenv and Pip for Poetry or Pipenv. For more, review Modern Python Environments.

Go ahead and add the following files and folders:

├── app.py
├── instance
│   └── .gitkeep
├── project
│   ├── __init__.py
│   └── journal_api
│       ├── __init__.py
│       └── routes.py
└── requirements.txt

Next, to install the necessary Python packages, add the dependencies to the requirements.txt file in the project root:

apifairy==0.9.1
Flask==2.1.2
Flask-SQLAlchemy==2.5.1
marshmallow-sqlalchemy==0.28.0

Install:

(venv)$ pip install -r requirements.txt

This Flask project will utilize two best practices for Flask applications:

Application Factory - used for creating the Flask application in a function
Blueprints - used for organizing a group of related views

Application Factory

Start by defining the Application Factory function in project/__init__.py:

from apifairy import APIFairy
from flask import Flask, json
from flask_marshmallow import Marshmallow

# -------------
# Configuration
# -------------

# Create the instances of the Flask extensions in the global scope,
# but without any arguments passed in. These instances are not
# attached to the Flask application at this point.
apifairy = APIFairy()
ma = Marshmallow()


# ----------------------------
# Application Factory Function
# ----------------------------

def create_app():
    # Create the Flask application
    app = Flask(__name__)

    initialize_extensions(app)
    register_blueprints(app)
    return app


# ----------------
# Helper Functions
# ----------------

def initialize_extensions(app):
    # Since the application instance is now created, pass it to each Flask
    # extension instance to bind it to the Flask application instance (app)
    apifairy.init_app(app)
    ma.init_app(app)


def register_blueprints(app):
    # Import the blueprints
    from project.journal_api import journal_api_blueprint

    # Since the application instance is now created, register each Blueprint
    # with the Flask application instance (app)
    app.register_blueprint(journal_api_blueprint, url_prefix='/journal')

With the Application Factory function defined, it can be called in app.py in the top-level folder of the project:

from project import create_app


# Call the application factory function to construct a Flask application
# instance using the development configuration
app = create_app()

Blueprint

Let's define the journal_api blueprint. Start by defining the journal_api blueprint in project/journal_api/__init__.py:

"""
The 'journal_api' blueprint handles the API for managing journal entries.
Specifically, this blueprint allows for journal entries to be added, edited,
and deleted.
"""
from flask import Blueprint

journal_api_blueprint = Blueprint('journal_api', __name__, template_folder='templates')

from . import routes

Now it's time to define the API endpoints for the journal in project/journal_api/routes.py.

Start with the necessary imports:

from apifairy import body, other_responses, response
from flask import abort

from project import ma
from . import journal_api_blueprint

For this initial version of the Flask Journal API, the database will be a list of journal entries:

# --------
# Database
# --------

messages = [
    dict(id=1, entry='The sun was shining when I woke up this morning.'),
    dict(id=2, entry='I tried a new fruit mixture in my oatmeal for breakfast.'),
    dict(id=3, entry='Today I ate a great sandwich for lunch.')
]

Next, define the schemas for creating a new journal entry and for returning the journal entries:

# -------
# Schemas
# -------

class NewEntrySchema(ma.Schema):
    """Schema defining the attributes when creating a new journal entry."""
    entry = ma.String(required=True)


class EntrySchema(ma.Schema):
    """Schema defining the attributes in a journal entry."""
    id = ma.Integer()
    entry = ma.String()


new_entry_schema = NewEntrySchema()
entry_schema = EntrySchema()
entries_schema = EntrySchema(many=True)

Both of these schema classes inherit from ma.Schema, which is provided by Flask-Marshmallow. It's also a good idea to create objects of these schemas, as this allows you to define a schema that can return multiple entries (using the many=True argument).

Now we're ready to define the API endpoints!

Routes

Start with retrieving all the journal entries:

@journal_api_blueprint.route('/', methods=['GET'])
@response(entries_schema)
def journal():
    """Return all journal entries"""
    return messages

This view function uses the @response decorator to define that multiple entries are returned. The view function returns the full list of journal entries (return messages).

Next, create the API endpoint for adding a new journal entry:

@journal_api_blueprint.route('/', methods=['POST'])
@body(new_entry_schema)
@response(entry_schema, 201)
def add_journal_entry(kwargs):
    """Add a new journal entry"""
    new_message = dict(**kwargs, id=messages[-1]['id']+1)
    messages.append(new_message)
    return new_message

This view function uses the @body decorator to define the input to the API endpoint and the @response decorator to define the output from the API endpoint.

The input data that is parsed from the @body decorator is passed into the add_journal_entry() view function as the kwargs (key word arguments) argument. This data is then used to create a new journal entry and add it to the database:

new_message = dict(**kwargs, id=messages[-1]['id']+1)
messages.append(new_message)

The newly created journal entry is then returned (return new_message). Notice how the @response decorator defines the return code as 201 (Created) to indicate that the journal entry was added to the database.

Create the API endpoint for retrieving a specific journal entry:

@journal_api_blueprint.route('/<int:index>', methods=['GET'])
@response(entry_schema)
@other_responses({404: 'Entry not found'})
def get_journal_entry(index):
    """Return a journal entry"""
    if index >= len(messages):
        abort(404)

    return messages[index]

This view function uses the @other_responses decorator to specify non-standard responses.

The @other_responses decorator is only used for documentation purposes! It does not provide any functionality in terms of returning error codes.

Create the API endpoint for updating a journal entry:

@journal_api_blueprint.route('/<int:index>', methods=['PUT'])
@body(new_entry_schema)
@response(entry_schema)
@other_responses({404: 'Entry not found'})
def update_journal_entry(data, index):
    """Update a journal entry"""
    if index >= len(messages):
        abort(404)

    messages[index] = dict(data, id=index+1)
    return messages[index]

This view function uses the @body and @response decorators to define the inputs and outputs (respectively) for this API endpoint. Additionally, the @other_responses decorator defines the non-standard response if the journal entry is not found.

Finally, create the API endpoint for deleting a journal entry:

@journal_api_blueprint.route('/<int:index>', methods=['DELETE'])
@other_responses({404: 'Entry not found'})
def delete_journal_entry(index):
    """Delete a journal entry"""
    if index >= len(messages):
        abort(404)

    messages.pop(index)
    return '', 204

This view function does not use the @body and @response decorators, as there are no inputs or outputs for this API endpoint. If the journal entry is successfully deleted, then a 204 (No Content) status code is returned with no data.

Running the Flask Application

To test things out, within one terminal window, configure the Flask application and run the development server:

(venv) $ export FLASK_APP=app.py
(venv) $ export FLASK_ENV=development
(venv) $ flask run

Then, in a different terminal window, you can interact with the API. Feel free to use your tool of choice here, like cURL, HTTPie, Requests, or Postman.

Requests example:

$ python3

>>> import requests
>>>
>>> r = requests.get('http://127.0.0.1:5000/journal/')
>>> print(r.text)
>>>
>>> post_data = {'entry': "some message"}
>>> r = requests.post('http://127.0.0.1:5000/journal/', json=post_data)
>>> print(r.text)

Want to test out the API endpoints more easily? Check out this script, which adds CLI commands for interacting with the API endpoints for retrieving, creating, updating, and deleting journal entries.

Documentation

An incredible feature of APIFairy is the automatic API documentation creation!

There are three key aspects to configuring the API documentation:

Docstrings for the API endpoints (i.e., view functions)
Docstring for the overall API project
Configuration variables to specify the look of the API documentation

We already covered the first item in the previous section since we included the docstrings for each view function. For example, the journal() view function has a short description of the purpose of this API endpoint:

@journal_api_blueprint.route('/', methods=['GET'])
@response(entries_schema)
def journal():
    """Return all journal entries"""
    return messages

Next, we need to include the docstring to describe the overall project at the very top of the project/__init__.py file:

"""
Welcome to the documentation for the Flask Journal API!

## Introduction

The Flask Journal API is an API (Application Programming Interface) for creating a **daily journal** that documents events that happen each day.

## Key Functionality

The Flask Journal API has the following functionality:

1. Work with journal entries:
  * Create a new journal entry
  * Update a journal entry
  * Delete a journal entry
  * View all journal entries
2. <More to come!>

## Key Modules

This project is written using Python 3.10.1.

The project utilizes the following modules:

* **Flask**: micro-framework for web application development which includes the following dependencies:
  * **click**: package for creating command-line interfaces (CLI)
  * **itsdangerous**: cryptographically sign data
  * **Jinja2**: templating engine
  * **MarkupSafe**: escapes characters so text is safe to use in HTML and XML
  * **Werkzeug**: set of utilities for creating a Python application that can talk to a WSGI server
* **APIFairy**: API framework for Flask which includes the following dependencies:
  * **Flask-Marshmallow** - Flask extension for using Marshmallow (object serialization/deserialization library)
  * **Flask-HTTPAuth** - Flask extension for HTTP authentication
  * **apispec** - API specification generator that supports the OpenAPI specification
* **pytest**: framework for testing Python projects
"""
...

This docstring is used to describe the overall project, including the key functionality provided and the key Python packages used by the project.

Finally, some configuration variables need to be defined to specify the look of the API documentation. Update the create_app() function in project/__init__.py:

def create_app():
    # Create the Flask application
    app = Flask(__name__)

    # Configure the API documentation
    app.config['APIFAIRY_TITLE'] = 'Flask Journal API'
    app.config['APIFAIRY_VERSION'] = '0.1'
    app.config['APIFAIRY_UI'] = 'elements'

    initialize_extensions(app)
    register_blueprints(app)

    return app

Ready to see the documentation for the project? Start the Flask development server via flask run, and then navigate to http://127.0.0.1:5000/docs to see the API documentation created by APIFairy:

On the left-hand pane, there's a list of API endpoints for the journal_api blueprint. Clicking on one of the endpoints shows all the details about that endpoint:

What's amazing about this API documentation is the ability to see how the API endpoints work (assuming the Flask development server is running). On the right-hand pane of the documentation, enter a journal entry index and click "Send API request". The API response is then displayed:

This interactive documentation makes it easy for users to understand the API!

Database

For demonstration purposes, a SQLite database will be used in this tutorial.

Configuration

Since Flask-SQLAlchemy was already installed at the beginning of this tutorial, we need to configure it in the project/__init__.py file.

Start by creating a SQLAlchemy() object in the 'Configuration' section:

...

from apifairy import APIFairy
from flask import Flask, json
from flask_marshmallow import Marshmallow
from flask_sqlalchemy import SQLAlchemy  # <-- NEW!!


# -------------
# Configuration
# -------------

# Create the instances of the Flask extensions in the global scope,
# but without any arguments passed in. These instances are not
# attached to the Flask application at this point.
apifairy = APIFairy()
ma = Marshmallow()
database = SQLAlchemy()  # <-- NEW!!

...

Next, update the create_app() function to specify the necessary configuration variables:

def create_app():
    # Create the Flask application
    app = Flask(__name__)

    # Configure the API documentation
    app.config['APIFAIRY_TITLE'] = 'Flask Journal API'
    app.config['APIFAIRY_VERSION'] = '0.1'
    app.config['APIFAIRY_UI'] = 'elements'

    # NEW!
    # Configure the SQLite database (intended for development only!)
    app.config['SQLALCHEMY_DATABASE_URI'] = f"sqlite:///{os.path.join(os.getcwd(), 'instance', 'app.db')}"
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

    initialize_extensions(app)
    register_blueprints(app)

    return app

Add the import to the top:

import os

The SQLALCHEMY_DATABASE_URI configuration variable is critical to identifying the location of the SQLite database. For this tutorial, the database is stored in instance/app.db.

Finally, update the initialize_extensions() function to initialize the Flask-SQLAlchemy object:

def initialize_extensions(app):
    # Since the application instance is now created, pass it to each Flask
    # extension instance to bind it to the Flask application instance (app)
    apifairy.init_app(app)
    ma.init_app(app)
    database.init_app(app)  # <-- NEW!!

Want to learn more about how this Flask app is wired together? Check out my course on how to build, test, and deploy a Flask application:

Developing Web Applications with Python and Flask

Database Model

Create a new project/models.py file to define the database table to represent the journal entries:

from project import database


class Entry(database.Model):
    """Class that represents a journal entry."""
    __tablename__ = 'entries'

    id = database.Column(database.Integer, primary_key=True)
    entry = database.Column(database.String, nullable=False)

    def __init__(self, entry: str):
        self.entry = entry

    def update(self, entry: str):
        self.entry = entry

    def __repr__(self):
        return f'<Entry: {self.entry}>'

This new class, Entry, specifies that the entries database table will contain two elements (for now!) to represent a journal entry:

id - the primary key (primary_key=True) for the table, which means that it's a unique identifier for each element (row) in the table
entry - string for storing the journal entry text

While models.py defines the database table, it doesn't create the tables in the SQLite database. To create the tables, start the Flask shell in a terminal window:

(venv)$ flask shell

>>> from project import database
>>> database.drop_all()
>>> database.create_all()
>>> quit()

(venv)$

Journal API Updates

Since we're progressing on to use a SQLite database, start by deleting the temporary database (Python list) that was defined in project/journal_api/routes.py:

# --------
# Database
# --------

messages = [
    dict(id=1, entry='The sun was shining when I woke up this morning.'),
    dict(id=2, entry='I tried a new fruit mixture in my oatmeal for breakfast.'),
    dict(id=3, entry='Today I ate a great sandwich for lunch.')
]

Next, we need to update each API endpoint (i.e., the view functions) to utilize the SQLite database.

Start by updating the journal() view function:

@journal_api_blueprint.route('/', methods=['GET'])
@response(entries_schema)
def journal():
    """Return all journal entries"""
    return Entry.query.all()

The complete list of journal entries is now retrieved from the SQLite database. Notice how the schemas or decorators for this view function did not need to change... only the underlying process for getting the users changed!

Add the import:

from project.models import Entry

Next, update the add_journal_entry() view function:

@journal_api_blueprint.route('/', methods=['POST'])
@body(new_entry_schema)
@response(entry_schema, 201)
def add_journal_entry(kwargs):
    """Add a new journal entry"""
    new_message = Entry(**kwargs)
    database.session.add(new_message)
    database.session.commit()
    return new_message

The inputs to this view function are specified by new_entry_schema:

class NewEntrySchema(ma.Schema):
    """Schema defining the attributes when creating a new journal entry."""
    entry = ma.String(required=True)

new_entry_schema = NewEntrySchema()

The entry string is used to create a new instance of the Entry class (defined in models.py) and this journal entry is then added to the database.

Add the import:

from project import database

Next, update get_journal_entry():

@journal_api_blueprint.route('/<int:index>', methods=['GET'])
@response(entry_schema)
@other_responses({404: 'Entry not found'})
def get_journal_entry(index):
    """Return a journal entry"""
    entry = Entry.query.filter_by(id=index).first_or_404()
    return entry

This function now attempts to look up the specified journal entry (based on the index):

entry = Entry.query.filter_by(id=index).first_or_404()

If the entry exists, it's returned to the user. If the entry does not exist, a 404 (Not Found) error is returned.

Next, update update_journal_entry():

@journal_api_blueprint.route('/<int:index>', methods=['PUT'])
@body(new_entry_schema)
@response(entry_schema)
@other_responses({404: 'Entry not found'})
def update_journal_entry(data, index):
    """Update a journal entry"""
    entry = Entry.query.filter_by(id=index).first_or_404()
    entry.update(data['entry'])
    database.session.add(entry)
    database.session.commit()
    return entry

The update_journal_entry() view function now attempts to retrieve the specified journal entry:

entry = Entry.query.filter_by(id=index).first_or_404()

If the journal entry exists, the entry is updated with the new text and then saved to the database.

Finally, update delete_journal_entry():

@journal_api_blueprint.route('/<int:index>', methods=['DELETE'])
@other_responses({404: 'Entry not found'})
def delete_journal_entry(index):
    """Delete a journal entry"""
    entry = Entry.query.filter_by(id=index).first_or_404()
    database.session.delete(entry)
    database.session.commit()
    return '', 204

If the specified journal entry is found, then it's deleted from the database.

Run the development server. Test out each of the endpoints to ensure they still work.

Error Handling

Since this Flask project is an API, error codes should be returned in JSON format instead of the typical HTML format.

In the Flask project, this can be accomplished by using a custom error handler. In project/__init__.py, define a new function (register_error_handlers()) at the bottom of the file:

def register_error_handlers(app):
    @app.errorhandler(HTTPException)
    def handle_http_exception(e):
        """Return JSON instead of HTML for HTTP errors."""
        # Start with the correct headers and status code from the error
        response = e.get_response()
        # Replace the body with JSON
        response.data = json.dumps({
            'code': e.code,
            'name': e.name,
            'description': e.description,
        })
        response.content_type = 'application/json'
        return response

This function registers a new error handler for when an HTTPException is raised to convert the output into JSON format.

Add the import:

from werkzeug.exceptions import HTTPException

Also, update the Application Factory function, create_app(), to call this new function:

def create_app():
    # Create the Flask application
    app = Flask(__name__)

    # Configure the API documentation
    app.config['APIFAIRY_TITLE'] = 'Flask Journal API'
    app.config['APIFAIRY_VERSION'] = '0.1'
    app.config['APIFAIRY_UI'] = 'elements'

    # Configure the SQLite database (intended for development only!)
    app.config['SQLALCHEMY_DATABASE_URI'] = f"sqlite:///{os.path.join(os.getcwd(), 'instance', 'app.db')}"
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

    initialize_extensions(app)
    register_blueprints(app)
    register_error_handlers(app)  # NEW!!

    return app

Authentication

Authentication is the process of validating the identity of a user attempting to access a system, which in this case is the API.

Authorization, on the other hand, is the process of verifying what specific resources a specific user should have access to.

APIFairy utilizes Flask-HTTPAuth for authentication support. In this tutorial, we'll be using Flask-HTTPAuth in two manners:

Basic Authentication - used to generate a token based on the user's email/password
Token Authentication - used to authenticate the user on all other API endpoints

The token authentication used via Flask-HTTPAuth is often referred to as Bearer Authentication, as the process invokes granting access to the "bearer" of the token. The token must be included in the HTTP headers in the Authorization header, such as "Authorization: Bearer ".

The following diagram illustrates a typical flow of how a new user interacts with the application to retrieve an authentication token:

Configuration

Since Flask-HTTPAuth was already installed when APIFairy was installed at the beginning of this tutorial, we just need to configure it in the project/__init__.py file.

Start by creating separate objects for the basic and token authentication:

...

import os

from apifairy import APIFairy
from flask import Flask, json
from flask_httpauth import HTTPBasicAuth, HTTPTokenAuth  # NEW!!
from flask_marshmallow import Marshmallow
from flask_sqlalchemy import SQLAlchemy
from werkzeug.exceptions import HTTPException


# -------------
# Configuration
# -------------

# Create the instances of the Flask extensions in the global scope,
# but without any arguments passed in. These instances are not
# attached to the Flask application at this point.
apifairy = APIFairy()
ma = Marshmallow()
database = SQLAlchemy()
basic_auth = HTTPBasicAuth()  # NEW!!
token_auth = HTTPTokenAuth()  # NEW!!

...

No further updates are needed in project/__init__.py.

Database Model

In project/models.py, a new User model needs to be created to represent a user:

class User(database.Model):
    __tablename__ = 'users'

    id = database.Column(database.Integer, primary_key=True)
    email = database.Column(database.String, unique=True, nullable=False)
    password_hashed = database.Column(database.String(128), nullable=False)
    entries = database.relationship('Entry', backref='user', lazy='dynamic')
    auth_token = database.Column(database.String(64), index=True)
    auth_token_expiration = database.Column(database.DateTime)

    def __init__(self, email: str, password_plaintext: str):
        """Create a new User object."""
        self.email = email
        self.password_hashed = self._generate_password_hash(password_plaintext)

    def is_password_correct(self, password_plaintext: str):
        return check_password_hash(self.password_hashed, password_plaintext)

    def set_password(self, password_plaintext: str):
        self.password_hashed = self._generate_password_hash(password_plaintext)

    @staticmethod
    def _generate_password_hash(password_plaintext):
        return generate_password_hash(password_plaintext)

    def generate_auth_token(self):
        self.auth_token = secrets.token_urlsafe()
        self.auth_token_expiration = datetime.utcnow() + timedelta(minutes=60)
        return self.auth_token

    @staticmethod
    def verify_auth_token(auth_token):
        user = User.query.filter_by(auth_token=auth_token).first()
        if user and user.auth_token_expiration > datetime.utcnow():
            return user

    def revoke_auth_token(self):
        self.auth_token_expiration = datetime.utcnow()

    def __repr__(self):
        return f'<User: {self.email}>'

Add the imports:

import secrets
from datetime import datetime, timedelta

from werkzeug.security import check_password_hash, generate_password_hash

The User model uses werkzeug.security for hashing the user's password before storing it in the database.

Remember: Never store the plaintext password in a database!

The User model uses secrets to generate an authentication token for a specific user. This token is created in the generate_auth_token() method and includes an expiration date/time of 60 minutes in the future:

def generate_auth_token(self):
    self.auth_token = secrets.token_urlsafe()
    self.auth_token_expiration = datetime.utcnow() + timedelta(minutes=60)
    return self.auth_token

There's a static method, verify_auth_token(), which is used to verify the authentication token (while considering the expiration time) and return the user from a valid token:

@staticmethod
def verify_auth_token(auth_token):
    user = User.query.filter_by(auth_token=auth_token).first()
    if user and user.auth_token_expiration > datetime.utcnow():
        return user

One more method of interest is revoke_auth_token(), which is used to revoke the authentication token for a specific user:

def revoke_auth_token(self):
    self.auth_token_expiration = datetime.utcnow()

Entry Model

To establish the one-to-many relationship between the user ("one") and their entries ("many"), the Entry model needs to be updated to link the entries and users tables together:

class Entry(database.Model):
    """Class that represents a journal entry."""
    __tablename__ = 'entries'

    id = database.Column(database.Integer, primary_key=True)
    entry = database.Column(database.String, nullable=False)
    user_id = database.Column(database.Integer, database.ForeignKey('users.id'))  # <-- NEW!!

    def __init__(self, entry: str):
        self.entry = entry

    def update(self, entry: str):
        self.entry = entry

    def __repr__(self):
        return f'<Entry: {self.entry}>'

The User model already contains the link back to the entries table:

entries = database.relationship('Entry', backref='user', lazy='dynamic')

Users API Blueprint

The user management functionality of the Flask project will be defined in a separate Blueprint called users_api_blueprint.

Start by creating a new directory in "project" called "users_api". Within that directory create an __init__.py file:

from flask import Blueprint


users_api_blueprint = Blueprint('users_api', __name__)

from . import authentication, routes

This new Blueprint needs to be registered with the Flask app in projects/__init__.py within the register_blueprints() function:

def register_blueprints(app):
    # Import the blueprints
    from project.journal_api import journal_api_blueprint
    from project.users_api import users_api_blueprint  # NEW!!

    # Since the application instance is now created, register each Blueprint
    # with the Flask application instance (app)
    app.register_blueprint(journal_api_blueprint, url_prefix='/journal')
    app.register_blueprint(users_api_blueprint, url_prefix='/users')  # NEW!!

Authentication Functions

To use Flask-HTTPAuth, several functions need to be defined to handle checking user credentials.

Create a new project/users_api/authentication.py file to handle the basic authentication and token authentication.

For the basic authentication (checking a user's email and password):

from werkzeug.exceptions import Forbidden, Unauthorized

from project import basic_auth, token_auth
from project.models import User


@basic_auth.verify_password
def verify_password(email, password):
    user = User.query.filter_by(email=email).first()
    if user is None:
        return None

    if user.is_password_correct(password):
        return user


@basic_auth.error_handler
def basic_auth_error(status=401):
    error = (Forbidden if status == 403 else Unauthorized)()
    return {
        'code': error.code,
        'message': error.name,
        'description': error.description,
    }, error.code, {'WWW-Authenticate': 'Form'}

The verify_password() function is used to check that a user exists and that their password is correct. This function will be used by Flask-HTTPAuth for verifying the password when basic authentication is needed (thanks to the @basic_auth.verify_password decorator.)

Additionally, an error handler is defined for the basic authentication that returns information about the error in JSON format.

For the token authentication (processing a token to determine if the user is valid):

@token_auth.verify_token
def verify_token(auth_token):
    return User.verify_auth_token(auth_token)


@token_auth.error_handler
def token_auth_error(status=401):
    error = (Forbidden if status == 403 else Unauthorized)()
    return {
        'code': error.code,
        'message': error.name,
        'description': error.description,
    }, error.code

The verify_token() function is used to check if an authentication token is valid. This function will be used by Flask-HTTPAuth for verifying the token when token authentication is needed (thanks to the @token_auth.verify_token decorator.)

Additionally, an error handler is defined for the token authentication that returns information about the error in JSON format.

Users Routes

In the users_api_blueprint, there will be two routes:

Registering a new user
Retrieving an authentication token

To start, a new set of schemas (using marshmallow) need to be defined in projects/users_api/routes.py:

from project import ma

from . import users_api_blueprint

# -------
# Schemas
# -------

class NewUserSchema(ma.Schema):
    """Schema defining the attributes when creating a new user."""
    email = ma.String()
    password_plaintext = ma.String()


class UserSchema(ma.Schema):
    """Schema defining the attributes of a user."""
    id = ma.Integer()
    email = ma.String()


class TokenSchema(ma.Schema):
    """Schema defining the attributes of a token."""
    token = ma.String()


new_user_schema = NewUserSchema()
user_schema = UserSchema()
token_schema = TokenSchema()

These schemas will be used for defining the inputs and outputs to the view functions defined in this file.

Registering a new user

Next, define the view function for registering a new user:

@users_api_blueprint.route('/', methods=['POST'])
@body(new_user_schema)
@response(user_schema, 201)
def register(kwargs):
    """Create a new user"""
    new_user = User(**kwargs)
    database.session.add(new_user)
    database.session.commit()
    return new_user

Add the imports:

from apifairy import authenticate, body, other_responses, response

from project import basic_auth, database, ma
from project.models import User

This API endpoint uses the new_user_schema to specify that the email and password are the inputs.

NOTE: Since the email and password are sent to this API endpoint, it's a good time to remember that using HTTP is acceptable during development testing, but HTTPS (secure) should always be used in production.

The email and password (defined as the kwargs - keyword arguments) are then unpacked to create a new User object, which is saved to the database:

new_user = User(**kwargs)
database.session.add(new_user)
database.session.commit()

The output from the API endpoint is defined by user_schema, which is the ID and email for the new user.

Retrieving an authentication token

The other view function to define in projects/users_api/routes.py is for retrieving the authentication token:

@users_api_blueprint.route('/get-auth-token', methods=['POST'])
@authenticate(basic_auth)
@response(token_schema)
@other_responses({401: 'Invalid username or password'})
def get_auth_token():
    """Get authentication token"""
    user = basic_auth.current_user()
    token = user.generate_auth_token()
    database.session.add(user)
    database.session.commit()
    return dict(token=token)

The @authenticate decorator is used for the first time in this tutorial and it specifies that the basic authentication should be used to guard this route:

@authenticate(basic_auth)

When the user wants to retrieve their authentication token, they need to send a POST request to this API endpoint with the email and password embedded in the 'Authorization' header. As an example, the following Python command using the Requests package could be made to this API endpoint:

>>> import requests
>>> r = requests.post(
    'http://127.0.0.1:5000/users/get-auth-token',
    auth=('[email protected]', 'FlaskIsAwesome123')
)

If the basic authentication is successful, the view function retrieves the current user using the current_user() method provided by Flask-HTTPAuth:

user = basic_auth.current_user()

A new authentication token is created for that user:

token = user.generate_auth_token()

And that token is saved to the database so that it can be used to authenticate the user in the future (at least for the next 60 minutes!).

Finally, the new authentication token is returned for the user to save for all subsequent API calls.

API Endpoint Updates

With an authentication process in place, it's time to add some guards to the existing API endpoints to make sure that only valid users can access the application.

These updates are for the view functions defined in projects/journal_api/routes.py.

First, update journal() to only return the journal entries for the current user:

@journal_api_blueprint.route('/', methods=['GET'])
@authenticate(token_auth)
@response(entries_schema)
def journal():
    """Return journal entries"""
    user = token_auth.current_user()
    return Entry.query.filter_by(user_id=user.id).all()

Update the imports at the top like so:

from apifairy import authenticate, body, other_responses, response
from flask import abort

from project import database, ma, token_auth
from project.models import Entry

from . import journal_api_blueprint

The @authenticate decorator specifies that token authentication needs to be used when accessing this API endpoint. As an example, the following GET request could be made using Requests (after the authentication token has been retrieved):

>>> import requests
>>> headers = {'Authorization': f'Bearer {auth_token}'}
>>> r = requests.get('http://127.0.0.1:5000/journal/', headers=headers)

Once the user is authenticated, the complete list of journal entries is retrieved from the database based on the user's ID:

user = token_auth.current_user()
return Entry.query.filter_by(user_id=user.id).all()

The output from this API endpoint is defined by the @response decorator, which is a list of journal entries (ID, entry, user ID).

Next, update add_journal_entry():

@journal_api_blueprint.route('/', methods=['POST'])
@authenticate(token_auth)
@body(new_entry_schema)
@response(entry_schema, 201)
def add_journal_entry(kwargs):
    """Add a new journal entry"""
    user = token_auth.current_user()
    new_message = Entry(user_id=user.id, **kwargs)
    database.session.add(new_message)
    database.session.commit()
    return new_message

As with the previous view function, the @authenticate decorator is used to specify that token authentication needs to be used when accessing this API endpoint. Additionally, the journal entry is now added by specifying the user ID that should be associated with the journal entry:

user = token_auth.current_user()
new_message = Entry(user_id=user.id, **kwargs)

The new journal entry is saved to the database and the journal entry is returned (as defined by the @response decorator).

Next, update get_journal_entry():

@journal_api_blueprint.route('/<int:index>', methods=['GET'])
@authenticate(token_auth)
@response(entry_schema)
@other_responses({403: 'Forbidden', 404: 'Entry not found'})
def get_journal_entry(index):
    """Return a journal entry"""
    user = token_auth.current_user()
    entry = Entry.query.filter_by(id=index).first_or_404()

    if entry.user_id != user.id:
        abort(403)
    return entry

The @authenticate decorator is added to specify that token authentication is needed to access this API endpoint.

When attempting to retrieve a journal entry, an additional check is added to make sure that the user attempting to access the journal entry is the actual "owner" of the entry. If not, then a 403 (Forbidden) error code is returned via the abort() function from Flask:

if entry.user_id != user.id:
        abort(403)

Take note that this API endpoint has two off-nominal responses specified by the @other_responses decorator:

@other_responses({403: 'Forbidden', 404: 'Entry not found'})

Reminder: The @other_responses decorator is only for documentation; it's the responsibility of the view function to raise these errors.

Next, update update_journal_entry():

@journal_api_blueprint.route('/<int:index>', methods=['PUT'])
@authenticate(token_auth)
@body(new_entry_schema)
@response(entry_schema)
@other_responses({403: 'Forbidden', 404: 'Entry not found'})
def update_journal_entry(data, index):
    """Update a journal entry"""
    user = token_auth.current_user()
    entry = Entry.query.filter_by(id=index).first_or_404()

    if entry.user_id != user.id:
        abort(403)

    entry.update(data['entry'])
    database.session.add(entry)
    database.session.commit()
    return entry

The updates to this view function are similar to the other view functions in this section:

@authenticate decorator specifies that token authentication is needed to access this API endpoint
Only the user that "owns" the journal entry is allowed to update the entry (otherwise, 403 (Forbidden))

Finally, update delete_journal_entry():

@journal_api_blueprint.route('/<int:index>', methods=['DELETE'])
@authenticate(token_auth)
@other_responses({403: 'Forbidden', 404: 'Entry not found'})
def delete_journal_entry(index):
    """Delete a journal entry"""
    user = token_auth.current_user()
    entry = Entry.query.filter_by(id=index).first_or_404()

    if entry.user_id != user.id:
        abort(403)

    database.session.delete(entry)
    database.session.commit()
    return '', 204

Conclusion

This tutorial provided a walk-through of how to easily and quickly build an API in Flask using APIFairy.

Decorators are the key to define the API endpoints:

Inputs:
- @arguments - input arguments from the query string of the URL
- @body - structure of the JSON request
Outputs:
- @response - structure of the JSON response
Authentication:
- @authenticate - authentication approach using Flask-HTTPAuth
Errors:
- @other_responses - off-nominal responses, such as HTTP error codes

Plus, the API documentation that is generated by APIFairy is excellent and provides key information for users of the application.

If you're interested in learning more about Flask, check out my course on how to build, test, and deploy a Flask application:

Developing Web Applications with Python and Flask

Featured Course

Developing Web Applications with Python and Flask

This course focuses on teaching the fundamentals of Flask by building and testing a web application using Test-Driven Development (TDD).

Buy Now $40 View Course

Share this tutorial

Contents

Objectives

What is APIFairy?

Decorators

Schemas

Authentication

Documentation

What Are We Building?

API Endpoints

Project Initialization

Application Factory

Blueprint

Routes

Running the Flask Application

Documentation

Database

Configuration

Database Model

Journal API Updates

Error Handling

Authentication

Configuration

Database Model

Entry Model

Users API Blueprint

Authentication Functions

Users Routes

Registering a new user

Retrieving an authentication token

API Endpoint Updates

Conclusion

Developing Web Applications with Python and Flask

Developing Web Applications with Python and Flask

Recommended Tutorials