December 24th, 2021

Building an API with Flask and SQLAlchemy






This is part of a series of articles on SaintsXCTF Version 2.0. The first article in the series provides an overview of the application. You DO NOT need to read prior articles in the series to fully understand this article.

While researching API frameworks and libraries, I grew interested in the Flask framework. Flask is a very lightweight framework, allowing engineers to quickly create APIs in Python without much opinionated tooling. For example, Flask does not come installed with a database access layer or ORM, allowing engineers to pick whichever database access library they prefer. This flexibility is appealing to me because it allows me to configure and design the API to my liking. Also, working in Python is easy and allows me to quickly write API code.

In this article, I begin by describing how I structured my SaintsXCTF API, which is written in Flask. Next, I provide an overview of Flask and SQLAlchemy, an object relational mapper (ORM). Finally, I dive into my API code. You can view the code discussed in this article in my saints-xctf-api repository.

The SaintsXCTF API is a REST API that returns JSON structured data. One of the main design principles I used for the API is to include links in the JSON response bodies. For example, take the entrypoint of the API.

{ "api_name":"saints-xctf-api", "self_link":"/", "versions_link":"/versions" }

Two links are specified in the JSON response body. The fields containing links are self_link and versions_link. self_link specifies the current API endpoint, and versions_link specifies another endpoint that a user can navigate to. This allows users with no knowledge of the API structure to navigate the API without needing to reference external documentation. Following the /versions endpoint specified in the versions_link field gives the following response.

{ "self":"/versions", "version_1":null, "version_2":"/v2", "version_latest":"/v2" }

Once again, this API response provides more links to follow. I'm currently using the second version of my SaintsXCTF API, so the remainder of my endpoints exist under the /v2 route.

{ "latest":true, "links":"/v2/links", "self":"/v2", "version":2 }

The root /v2 route gives a bit of metadata about the API, and also provides a link to /v2/links. If a user follows this route, they receive a list of all the top-level application endpoints in the API.

{ "activation_code":"/v2/activation_code/links", "comment":"/v2/comments/links", "flair":"/v2/flair/links", "forgot_password":"/v2/forgot_password/links", "group":"/v2/groups/links", "log":"/v2/logs/links", "log_feed":"/v2/log_feed/links", "notification":"/v2/notifications/links", "range_view":"/v2/range_view/links", "self":"/v2/links", "team":"/v2/teams/links", "user":"/v2/users/links" }

Let's say you are navigating through the API and are interested in viewing user routes. Following the /v2/users/links route returns all the user endpoints that are available, and what each endpoint is used for.

{ "endpoints":[ { "description":"Get all the users in the database.", "link":"/v2/users", "verb":"GET" }, { "description":"Create a new user.", "link":"/v2/users", "verb":"POST" }, { "description":"Retrieve a single user with a given username.", "link":"/v2/users/<username>", "verb":"GET" }, { "description":"Update a user with a given username.", "link":"/v2/users/<username>", "verb":"PUT" }, { "description":"Soft delete a user with a given username.", "link":"/v2/users/soft/<username>", "verb":"DELETE" }, ... ], "self":"/v2/users/links" }

I shortened the API response for brevity. After viewing this list of API endpoints, a user can determine which ones to invoke to match their needs. Let's say you decide to invoke the GET endpoint /v2/users/<username> with the username andy. Doing so results in the following response:

{ "api_index":"/versions", "contact":"", "error_description":"Unauthorized", "exception":"401 Unauthorized: The server could not verify that you are authorized to access the URL requested. You either supplied the wrong credentials (e.g. a bad password), or your browser doesn't understand how to supply the credentials required." }

Invoking the endpoint resulted in an HTTP 401 error. Many of the endpoints in my API are protected, requiring a temporary token in the Authorization header in order to access the API. If you don't supply a token or provide an invalid token, the API returns 401 and 403 errors, respectively. Providing a valid token in the HTTP request header results in the response below.

curl -H "Authorization: Bearer xxx"
{ "self":"/v2/users/andy", "user":{ "activation_code":"BbXuat", "class_year":2017, "deleted":false, "description":"I sometimes like to run...", "email":"", "favorite_event":"Shakeout", "first":"Andy", "last":"Jarombek", "last_signin":"2021-05-30 18:42:42", "location":"New York, NY", "member_since":"2016-12-23", "password":"$2b$12$KDaX8hy3P1fZnG9nUVf1TeXw/rJJ4YaEXYdBi.Bx9k8v3DRFeHQ8a", "profilepic_name":"1629931871738.jpg", "salt":"RjJH6PIndLmr8S5sjgGUj8", "subscribed":null, "username":"andy", "week_start":"monday" } }

I omitted the actual API token, a JWT, from the curl command above. This time, the API responded successfully with a JSON object of my user in SaintsXCTF.

The SaintsXCTF API is a CRUD (Create Read Update Delete) REST API. Therefore, API endpoints aren't limited to GET requests. Users can also perform POST, PUT, and DELETE requests. The example below is a POST request that creates a new exercise log for my user.

RequestBody='{"username":"andy","first":"Andy","last":"Jarombek","name":"NYRR Night at the Races","location":"New York, NY","date":"2021-12-16","type":"run","distance":10.5,"metric":"miles","time":"00:00:00","feel":6,"description":""}' curl -X POST \ -d "${RequestBody}" \ -H "Content-Type: application/json" \ -H "Authorization: Bearer xxx" \
{ "added":true, "log":{ "date":"2021-12-16", "deleted":false, "description":"", "distance":10.5, "feel":6, "first":"Andy", "last":"Jarombek", "location":"New York, NY", "log_id":47462, "metric":"miles", "miles":10.5, "name":"NYRR Night at the Races", "pace":"00:00:00", "time":"00:00:00", "time_created":"Fri, 17 Dec 2021 22:56:40 GMT", "type":"run", "username":"andy" }, "self":"/v2/logs" }

One of the main objectives of the API is to make it as easy to use and navigate as possible. This not only improves the experience for other users, but also for myself as I revisit and refactor the API codebase. While adding links to the JSON responses makes the API easier to explore, there is still some information that it lacks. This includes authentication mechanisms and documentation of JSON request body structures (such as the request body JSON for the exercise log shown above). This information requires additional documentation. I am currently working on Swagger API documentation for this purpose, and will likely write about Swagger and the OpenAPI specification in a future article.

Flask is a lightweight web application framework which is commonly used to build REST APIs. Flask applications are written in Python, with the Flask library at their core. Flask doesn't do much beyond handling routing, so a lot of the API functionality comes from other libraries (such as flask-sqlalchemy for a Database ORM and flask-bcrypt for bcrypt password hashing).

In my API, SQLAlchemy is used for accessing a MySQL database. I use the flask-sqlalchemy library, which is a wrapper around SQLAlchemy, making it easier to use in a Flask application.

SQLAlchemy is an ORM and SQL database library for Python. SQLAlchemy works with many different database engines; in my case, SQLAlchemy is used to query data from MySQL.

In my saints-xctf-api repository, the Flask application exists in an api/src directory.

The top level directory of the Flask application contains configuration files and infrastructure setup. The infrastructure setup consists of an Nginx reverse-proxy server (nginx.conf), a uWSGI application server (uwsgi.ini), and Dockerfiles for each. The configuration files for Flask are discussed in the next section.

There are also several subdirectories in the Flask application. dao contains files that follow the Data Access Object (DAO) pattern1. In essence, each DAO file consists of a class with methods which interact with the MySQL database using SQLAlchemy. model contains models for use in the SQLAlchemy ORM. Each model is a Python class that corresponds to a table in MySQL. route defines all the routes (endpoints) in the API. Each route is bound to a Python function which performs the logic needed to return a proper HTTP response. tests contains unit and integration tests for the API. utils contains reusable utility functions used throughout the API.

The configuration of my Flask application begins with a small file named

# from app import app is the entrypoint for the Flask application in the production environment. A uWSGI application server uses this file to run the API. simply imports a variable named app from app is an instance of Flask, an object representing the Flask application. I initialize app in a create_app() function, found within

# def create_app(config_name) -> Flask: """ Application factory function for the Flask app. Source: """ application = Flask(__name__) application.config.from_object(config[config_name]) application.register_blueprint(activation_code_route) application.register_blueprint(api_route) application.register_blueprint(user_route) application.register_blueprint(forgot_password_route) application.register_blueprint(flair_route) application.register_blueprint(log_route) application.register_blueprint(log_feed_route) application.register_blueprint(group_route) application.register_blueprint(comment_route) application.register_blueprint(range_view_route) application.register_blueprint(notification_route) application.register_blueprint(team_route) application.config['SQLALCHEMY_DATABASE_URI'] = get_connection_url() application.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False application.config['SQLALCHEMY_RECORD_QUERIES'] = True application.config['SLOW_DB_QUERY_TIME'] = 0.5 db.init_app(application) flask_bcrypt.init_app(application) application.cli.add_command(test) # Custom Error Handling @application.errorhandler(401) def error_403(ex): """ Custom error handler for when 401 HTTP codes occur. :param ex: String representing the error that occurred. :return: JSON describing the error. """ return jsonify({ 'error_description': "Unauthorized", 'exception': str(ex), 'contact': '', 'api_index': '/versions' }), 401 ... return application flask_env = os.getenv('FLASK_ENV') or 'local' app = create_app(flask_env)

I simplified the create_app() code snippet a bit, making it easier to discuss. create_app() takes a single config_name argument, which is the environment that the Flask application is run within. For example, when run in production, the value of config_name is production. The first line, application = Flask(__name__), creates an instance of Flask. application is eventually the return value of create_app(), representing the Flask API.

The next line, application.config.from_object(config[config_name]), sets configuration key-value pairs for the Flask application. The values of the Flask configuration are environment specific. config[config_name] is a Python object with properties. config is defined in a file, as shown below.

# class LocalConfig: ENV = 'local' AUTH_URL = 'http://saints-xctf-auth:5000' FUNCTION_URL = '' class DevelopmentConfig: ENV = 'dev' AUTH_URL = '' FUNCTION_URL = '' class ProductionConfig: ENV = 'prod' AUTH_URL = '' FUNCTION_URL = '' config = { 'local': LocalConfig, 'development': DevelopmentConfig, 'production': ProductionConfig }

config is a dictionary where the keys are environment names and values are classes with Flask configuration properties. In a production environment, the ProductionConfig class is the configuration for my Flask application.

Back to the create_app() function, all the application.register_blueprint() invocations are used to configure routes in the API. register_blueprint() takes a Blueprint object as an argument. In Flask, a blueprint is an application component which is registered with the main Flask application2. In my application, each blueprint is bound to a specific URL and contains all the endpoints under that URL.

For example, the call to application.register_blueprint(user_route) registers a user_route blueprint, which is defined in The creation of the blueprint, which is assigned to the /v2/users route, is demonstrated below.

# from flask import Blueprint user_route = Blueprint('user_route', __name__, url_prefix='/v2/users')

After registering blueprints for the Flask application in create_app(), I set four additional configuration variables by adding key-value pairs to the application.config dictionary. These configuration variables are separate from the ones discussed previously, such as those found in, because they have the same values across all environments.

Next in create_app(), I call db.init_app(application). This line of code initializes SQLAlchemy in the Flask application. db is an instance of SQLAlchemy from the flask-sqlalchemy library, which is defined in

# from flask_sqlalchemy import SQLAlchemy db = SQLAlchemy()

The next line, flask_bcrypt.init_app(application), initializes the Bcrypt password hashing algorithm with Flask. application.cli.add_command(test) adds a new command to the Flask CLI for testing the application. The CLI command is defined in a separate file and is invoked using a python -m flask test shell command.

The remainder of the create_app() function sets custom error messages for different HTTPS error codes. The custom error messages are JSON strings, all of which can be viewed in the file.

The Flask API accesses data from a MySQL database using the SQLAlchemy library. SQLAlchemy contains an ORM (Object Relational Mapping), allowing applications to build Python classes representing SQL tables. Instantiated objects of these Python classes (referred to as "model classes") contain a single row of data from the database table.

Model classes in my application exist in a model directory. One example model class is User, which represents a user table in the MySQL database. This model exists in a file.

# from database import db from sqlalchemy import Column from sqlalchemy.orm import deferred from sqlalchemy.dialects.mysql import LONGBLOB class User(db.Model): __tablename__ = 'users' # Data Columns username = Column(db.VARCHAR(20), primary_key=True) first = Column(db.VARCHAR(30), nullable=False, index=True) last = Column(db.VARCHAR(30), nullable=False, index=True) salt = Column(db.VARCHAR(255)) password = Column(db.VARCHAR(255), nullable=False) profilepic = deferred(Column(LONGBLOB), group='pictures') profilepic_name = deferred(Column(db.VARCHAR(50)), group='pictures') description = Column(db.VARCHAR(255)) member_since = Column(db.DATE, nullable=False) class_year = Column(db.INTEGER, index=True) location = Column(db.VARCHAR(50)) favorite_event = Column(db.VARCHAR(20)) activation_code = Column(db.VARCHAR(8), nullable=False) email = Column(db.VARCHAR(50), index=True) subscribed = Column(db.CHAR(1)) last_signin = Column(db.DATETIME, nullable=False) week_start = Column(db.VARCHAR(15)) deleted = Column(db.BOOLEAN) # Audit Columns created_date = Column(db.DATETIME) created_user = Column(db.VARCHAR(31)) created_app = Column(db.VARCHAR(31)) modified_date = Column(db.DATETIME) modified_user = Column(db.VARCHAR(31)) modified_app = Column(db.VARCHAR(31)) deleted_date = Column(db.DATETIME) deleted_user = Column(db.VARCHAR(31)) deleted_app = Column(db.VARCHAR(31))

Model classes extend the db.Model base class provided by SQLAlchemy. The __tablename__ field holds the name of the table in MySQL that the model is associated with. The remaining fields are the columns in the table and their associated data types. For example, one column in the user table is username, which holds a VARCHAR of length 20. username is the primary key of the user table, since every user has a unique username. This column is defined as a field in the User model object with the line username = Column(db.VARCHAR(20), primary_key=True).

I omitted a few details of the User model from the code snippet above, which I will discuss now. First, User has an __init__ constructor for converting a dictionary to an instance of the model. Second, User implements __str__ and __repr__ methods for printing out model objects cleanly in application logs. Third and finally, User implements __eq__ to compare two instances of User for equality.

For each table in the MySQL database, my API has two corresponding model classes. The naming convention for these classes is TableName and TableNameData. For example, the user table has User and UserData model classes. The reason for this structure is that the main model class has auditing columns that shouldn't be included in API response bodies. Therefore, the User model class is used to query the database, and UserData is used in API responses. Note that many APIs won't require a structure like this, and a single model class per table will suffice. The UserData class exists in a UserData file.

Data Access Objects (DAOs) are classes that interface with a data source; in the case of my API, a MySQL database. Separate DAOs are used for individual tables or multiple tables with similar business logic. All the DAOs in my application exist in a dao directory.

For business logic related to users, I have a DAO named defines a single DAO class named UserDao. Most of the methods in UserDao are related to a user table, however there is also some logic that alters data in different tables. A shortened version of UserDao is shown below.

# from typing import List from database import db from dao.basicDao import BasicDao from model.User import User class UserDao: @staticmethod def get_users() -> List[User]: """ Get a list of all the users in the database. :return: A list containing User model objects. """ return User.query.filter(User.deleted.is_(False)).all() @staticmethod def get_user_by_username(username: str) -> User: """ Get a single user from the database based on their username. :param username: Username which uniquely identifies the user. :return: The result of the database query. """ return User.query\ .filter_by(username=username)\ .filter(User.deleted.is_(False))\ .first() @staticmethod def add_user(user: User) -> bool: """ Add a user if it has a valid activation code. :param user: Object representing a user for the application. :return: True if the user is inserted into the database, False otherwise. """ db.session.add(user) return BasicDao.safe_commit() @staticmethod def update_user(username: str, user: User) -> bool: """ Update a user in the database. This function does NOT update passwords. :param username: Username which uniquely identifies the user. :param user: Object representing an updated user for the application. :return: True if the user is updated in the database, False otherwise. """ db.session.execute( ''' UPDATE users SET first=:first, last=:last, email=:email, profilepic_name=:profilepic_name, description=:description, class_year=:class_year, location=:location, favorite_event=:favorite_event, week_start=:week_start WHERE username=:username AND deleted IS FALSE ''', { 'first': user.first, 'last': user.last, 'email':, 'profilepic_name': user.profilepic_name, 'description': user.description, 'class_year': user.class_year, 'location': user.location, 'favorite_event': user.favorite_event, 'week_start': user.week_start, 'username': username } ) return BasicDao.safe_commit()

UserDao contains static methods which interact with MySQL using SQLAlchemy (the db variable is an instance of SQLAlchemy). Two different approaches are used to interact with the database. The first is to use the ORM and the second is to write SQL queries and execute them. For example, get_user_by_username() uses the ORM and the User model class to find a user in the database with a specific username. On the other hand, update_user() uses a SQL query to update a user in the database. Which approach is better is often a matter of personal preference. SQL is more expressive and easier to read for complex queries. The ORM avoids mixing SQL and Python code, and is easy to use for simple queries. The great thing about SQLAlchemy is that it offers both approaches for engineers to use.

One final note about the DAOs in my application. For inserts, updates, and deletes, my DAO methods use a function called BasicDao.safe_commit(). This is a custom function that attempts to commit one or many changes to the database as a transaction, and performs a rollback if the transaction fails. The code for this function is shown below and exists in

# from database import db from flask import current_app from sqlalchemy.exc import SQLAlchemyError class BasicDao: @staticmethod def safe_commit() -> bool: """ Safely attempt to commit changes to MySQL. Rollback in case of a failure. :return: True if the commit was successful, False if a rollback occurred. """ try: db.session.commit()'SQL Safely Committed') return True except SQLAlchemyError as error: db.session.rollback() current_app.logger.error('SQL Commit Failed! Rolling back...') current_app.logger.error(error.args) return False

So far, we've seen how DAOs interact with the database and use SQLAlchemy model classes to retrieve, update, create, or delete data. Of course, user's of the API don't interface with the DAO directly. Instead, users call endpoints (routes) in the REST API and request operations to be performed. These endpoints are defined and configured in the route directory of my application.

Route files contain functions, each of which represents an endpoint in the API. Each file also defines a Flask blueprint - a component representing a route in the API with sub-routes attached to it. Flask blueprints are instantiated as Python objects. Blueprint objects are used to annotate functions in the route file. These annotations specify which route the function is associated with and what HTTP methods it responds to.

For example, I have a route file for all the endpoints that provide metadata about the API. These endpoints exist in a file. A shortened version of the route file is shown below.

# from flask import Blueprint, jsonify, Response api_route = Blueprint('api_route', __name__, url_prefix='/') @api_route.route('/', methods=['GET']) def api() -> Response: return jsonify({ 'self_link': '/', 'api_name': 'saints-xctf-api', 'versions_link': '/versions' }) @api_route.route('/versions', methods=['GET']) def versions() -> Response: return jsonify({ 'self': '/versions', 'version_latest': '/v2', 'version_1': None, 'version_2': '/v2' }) @api_route.route('/v2', methods=['GET']) def version2() -> Response: return jsonify({ 'self': '/v2', 'version': 2, 'latest': True, 'links': '/v2/links' }) @api_route.route('/v2/links', methods=['GET']) def links() -> Response: return jsonify({ 'self': '/v2/links', 'activation_code': '/v2/activation_code/links', 'comment': '/v2/comments/links', 'flair': '/v2/flair/links', 'forgot_password': '/v2/forgot_password/links', 'group': '/v2/groups/links', 'log_feed': '/v2/log_feed/links', 'log': '/v2/logs/links', 'message_feed': '/v2/message_feed/links', 'message': '/v2/messages/links', 'notification': '/v2/notifications/links', 'range_view': '/v2/range_view/links', 'team': '/v2/teams/links', 'user': '/v2/users/links' })

The Blueprint for the route file is assigned to the variable api_route and instantiated with a call to its constructor: Blueprint('api_route', __name__, url_prefix='/'). This blueprint is assigned to the root route in the API ('/') via the url_prefix keyword argument.

The first function, api(), configures an endpoint for the route /. The endpoint route is a combination of the url_prefix from the blueprint and the first argument to the @api_route.route() annotation. The URL prefix is / and the route annotation is /, which can be simplified to /. The methods keyword argument to @api_route.route() specifies all the HTTP methods that the endpoint responds to; in the case of the / route, only HTTP GET requests are handled.

The api() method returns a Flask Response object, representing an HTTP response. api() uses a jsonify() function from the Flask library to build this response. jsonify() takes a dictionary as an argument, converts it to JSON, and wraps that JSON in a Flask Response object. From the API user perspective, navigating to the route / returns this JSON object and a HTTP 200 success code.

{ "api_name":"saints-xctf-api", "self_link":"/", "versions_link":"/versions" }

The other routes in the file follow a similar pattern of returning static JSON objects. is a more complex route file, including dynamic JSON responses dependent on user inputs and database query results. It not only contains HTTP GET requests, but also POST, PUT, and DELETE requests. A generalized outline of is shown below.

# user_route = Blueprint('user_route', __name__, url_prefix='/v2/users') @user_route.route('', methods=['GET', 'POST']) @auth_required(enabled_methods=[GET]) def users_redirect() -> Response: """ Redirect endpoints looking for a resource named 'users' to the user routes. :return: Response object letting the caller know where to redirect the request to. """ if request.method == 'GET': ''' [GET] /v2/users ''' return redirect(url_for('user_route.users'), code=302) elif request.method == 'POST': ''' [POST] /v2/users ''' return redirect(url_for('user_route.users'), code=307) @user_route.route('/', methods=['GET', 'POST']) @auth_required(enabled_methods=[GET]) def users() -> Response: """ Endpoints for searching all the users or creating a user :return: JSON representation of a list of users and relevant metadata """ if request.method == 'GET': ''' [GET] /v2/users/ ''' return users_get() elif request.method == 'POST': ''' [POST] /v2/users/ ''' return user_post() @user_route.route('/<username>', methods=['GET', 'PUT', 'DELETE']) @auth_required() @disabled(disabled_methods=[DELETE]) def user(username) -> Response: """ Endpoints for specific users (searching, updating, or deleting) :param username: Username (or email) of a User :return: JSON representation of a user and relevant metadata """ if request.method == 'GET': ''' [GET] /v2/users/<username> ''' return user_by_username_get(username) elif request.method == 'PUT': ''' [PUT] /v2/users/<username> ''' return user_by_username_put(username) elif request.method == 'DELETE': ''' [DELETE] /v2/users/<username> ''' return user_by_username_delete(username) @user_route.route('/soft/<username>', methods=['DELETE']) @auth_required() def user_soft_by_username(username) -> Response: """ Endpoints for soft deleting a user. :param username: Username of a User. :return: JSON representation of users and relevant metadata. """ if request.method == 'DELETE': ''' [DELETE] /v2/users/soft/<username> ''' return user_by_username_soft_delete(username)

At its core, is the same as its simpler counterpart. It contains a blueprint with a URL prefix and functions associated with HTTP requests to sub-URLs. This time, functions handle multiple HTTP verbs and often require authentication.

For example, users() handles both HTTP GET and POST verbs, as specified by its decorator @user_route.route('/', methods=['GET', 'POST']). It also requires authentication, which is specified by a custom decorator @auth_required(enabled_methods=[GET]). The keyword argument enabled_methods=[GET] specifies that only GET requests require authentication; POST methods do not.

How requests are handled in users() depends on the HTTP verb, which is accessed through the request.method property. If a HTTP GET request is made, a users_get() function is invoked; if a HTTP POST request is made, a user_post() function is invoked. users_get() is shown below.

def users_get() -> Response: """ Retrieve all the users in the database. :return: A response object for the GET API request. """ all_users: list = UserDao.get_users() if all_users is None: response = jsonify({ 'self': '/v2/users', 'users': None, 'error': 'an unexpected error occurred retrieving users' }) response.status_code = 500 return response else: user_dicts = [] for user in all_users: user_dict = UserData(user).__dict__ user_dict['this_user'] = f'/v2/users/{user_dict["username"]}' if user_dict.get('member_since') is not None: user_dict['member_since'] = str(user_dict['member_since']) if user_dict.get('last_signin') is not None: user_dict['last_signin'] = str(user_dict['last_signin']) user_dicts.append(user_dict) response = jsonify({ 'self': '/v2/users', 'users': user_dicts }) response.status_code = 200 return response

Similar to the routes in, users_get() returns JSON responses generated by jsonify(). However, the logic is a bit more complex since a DAO is leveraged to get data from MySQL, specifically UserDao.get_users(). The format of the response JSON and the response status code depends on the result of UserDao.get_users(). If UserDao.get_users() unsuccessfully retrieves users from the database, it returns None. In this scenario, a status code of 500 is returned with a JSON response object describing the error. If UserDao.get_users() successfully retrieves users from the database, it returns a list of User model objects. These objects are converted into Python dictionaries, the data is cleaned, and they are returned in a JSON response body with a status code of 200.

My API has many endpoints of varying degrees of complexity, so feel free to explore them all on GitHub.

All the functions which are Flask endpoints have decorators attached to them. The decorator @<name>.route(), where <name> is replaced with the variable name of a Blueprint or Flask object, registers a function to handle requests at a specific URL. The @<name>.route() decorator is part of the Flask library.

You may have noticed that some of my routes have additional decorators. For example, the user() function, defined in file and displayed again below, has two additional annotations: @auth_required() and @disabled().

@user_route.route('/<username>', methods=['GET', 'PUT', 'DELETE']) @auth_required() @disabled(disabled_methods=[DELETE]) def user(username) -> Response: """ Endpoints for specific users (searching, updating, or deleting) :param username: Username (or email) of a User :return: JSON representation of a user and relevant metadata """ if request.method == 'GET': ''' [GET] /v2/users/<username> ''' return user_by_username_get(username) elif request.method == 'PUT': ''' [PUT] /v2/users/<username> ''' return user_by_username_put(username) elif request.method == 'DELETE': ''' [DELETE] /v2/users/<username> ''' return user_by_username_delete(username)

@auth_required() is a custom decorator that checks for a valid JWT token in the Authorization header of HTTP requests. Since most of my endpoints require authentication, placing JWT validation logic in a reusable decorator makes the codebase a lot cleaner. The code for @auth_required() is shown below and exists in a file.

def auth_required(enabled_methods: Optional[List[HTTPMethod]] = None): """ Make a custom decorator for endpoints, indicating that authentication is required. :param enabled_methods: HTTP methods (verbs) that require authentication for an endpoint. """ def decorator(f): @functools.wraps(f) def decorated_function(*args, **kwargs): if enabled_methods and request.method not in enabled_methods:'Authentication is skipped for {request.method} requests to {request.url}') else: if 'Authorization' not in request.headers: abort(401) authorization_header: str = request.headers['Authorization'] token = authorization_header.replace('Bearer ', '') async def authenticate(): async with aiohttp.ClientSession() as session: async with url=f"{current_app.config['AUTH_URL']}/authenticate", json={'token': token} ) as response: response_body = await response.json() if not response_body.get('result'):'User Unauthorized') abort(403) else:'User Authorized') return f(*args, **kwargs) return decorated_function return decorator

@auth_required() leverages the @functools.wraps() factory method to create a decorator. This approach is a Python convention. @auth_required() takes a single keyword argument enabled_methods, which is an optional list of HTTP methods. If enabled_methods is passed to the decorator, authentication requirements are only enforced for HTTP methods defined in the enabled_methods list. If enabled_methods is not passed to the decorator, all HTTP methods that an endpoint responds to will require authentication.

auth_required() first checks if an Authorization header exists on the incoming HTTP request. If the header does not exist, a 401 HTTP error code is returned to the user.

Next, auth_required() extracts a JWT from the Authorization header, placing it in a token variable. The value of the Authorization header follows the pattern Bearer j.w.t, where j.w.t is replaced by the JWT. This token is used in another asynchronous function authenticate(), which makes an HTTP request to an authentication route in my application. I host this route using AWS Lambda and AWS API Gateway, which I'll discuss in an upcoming article. The authentication route is defined in code as "{current_app.config['AUTH_URL']}/authenticate", with the domain name coming from the Flask configuration object, specifically the 'AUTH_URL' property. This is due to the authentication URL being unique in different environments.

authenticate() uses the response from the authentication route, stored in response, to determine if the user is permitted to use the API. The authentication API returns a JSON object with a single boolean result field. If result is true, the user is successfully authenticated. If result is false, the user is not allowed to use the API. The code is to configured to return a HTTP 403 error code to the user if result is false. Otherwise, authenticate() performs no further operations and control is passed through to the route's function handler.

@disabled() is also a custom decorator that disables a route in the API. If a user tries to call a disabled route, they receive a HTTP 403 error. The implementation of @disabled() is shown below, and exists in a file.

def disabled(disabled_methods: Optional[List[HTTPMethod]] = None): """ Make a custom decorator for endpoints that are currently disabled and should not be invoked. :param disabled_methods: HTTP methods (verbs) that are disabled and follow the rules of this annotation. """ def decorator(f): @functools.wraps(f) def decorated_function(*args, **kwargs): if disabled_methods and request.method not in disabled_methods:'{request.method} requests to {request.url} are not disabled.') else:'This endpoint is disabled.') abort(403) return f(*args, **kwargs) return decorated_function return decorator

Similar to @auth_required(), @disabled() is written with the @functools.wraps() factory method and accepts an optional disabled_methods keyword argument, with the same effect as the enabled_methods keyword argument in @auth_required(). If a route has the @disabled() decorator and disabled_methods isn't specified, all HTTP methods to that route are disabled and return HTTP 403 errors. If a route has the @disabled() decorator and disabled_methods is specified, only the HTTP methods in the disabled_methods list return 403 errors. All other methods continue with normal execution. In code, HTTP 403 errors are returned from the API with the abort(403) command.

Writing APIs with Flask is easy and flexible. Flask is not very opinionated, leaving engineers in control of design choices. For beginners or junior level programmers, Python is very easy to work with as well, making Flask a good option for large teams with engineers at different skill levels. All the code for my Flask API is available on GitHub.