This tutorial demonstrates how to easily create a RESTful API with Flask and APIFairy.
Contents
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 projectAPIFAIRY_VERSION
- version string of the projectAPIFAIRY_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:
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 tableentry
- 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: