Create a Flask API with Python

Greetings, fellow developers! I’ve been wanting to share this post for a while, explaining how to build a simple API with Flask and Python. With this API, you’ll be able to quickly and easily create the backend for your web or mobile applications.

This post is quite long and comprehensive, but I assure you it’s worth reading all the way through. We’ll start with something easy for those new to Flask, and then gradually increase the difficulty until we build a complete RESTful API with user and task management. Let’s get started!

What is Flask?

Flask is a micro web framework for Python that allows you to create web applications quickly and easily. It is lightweight, flexible, and easy to learn, making it an excellent choice for developing RESTful APIs.

The fun thing about Flask is that, despite being a microframework, it is very powerful and allows you to create complete web applications with just a few lines of code. Additionally, it has a large number of libraries and extensions that facilitate integration with databases, authentication, form handling, among others. If you’re a ‘Punk’ who likes to have full control over your application, Flask is the perfect choice for you.

Prerequisites

• Python 3.12.x for this post I will use python 3.12.3 exactly, but any recent version should work.

• pip (the Python package manager) usually comes pre-installed with Python or you can install it by following the instructions at pip.pypa.io.

• Postman or cURL to test the API routes.

• An SMTP server to send emails, we will use it later to explain sending emails from the API.

• Virtual environment (optional but recommended) to work cleanly and isolated, we will use a virtual environment or ‘venv’, which will allow us to manage the dependencies of our project without affecting the rest of the system.

1. Set up the development environment

First, create a directory for your project and navigate to it:

mkdir flask-api-demo-blog
cd flask-api-demo-blog

Next, create a virtual environment for your project:

python -m venv .venv

You can use ‘venv’, ‘.venv’, or any other name; the most common are ‘.venv’ or ‘env’ as is usually done on GitHub.

Then, activate the virtual environment:

• On Windows:

.venv\Scripts\activate

• On macOS/Linux:

source .venv/bin/activate

Well, now we have a virtual environment ready to work in. We can verify that we are inside the virtual environment by looking at the terminal prompt, which should show the name of the virtual environment.

(.venv) marcosreyes@astronautmarkus:

You should see that parenthesis with the name of the virtual environment.

2. Install Flask

Now, install Flask using pip. Run the following command in the terminal:

pip install Flask

This will install Flask and its dependencies in our virtual environment.

Important: Constantly saving the requirements list is important to ensure that none escape us. We can do it with the following command:

pip freeze > requirements.txt

This will create a requirements.txt file that contains all the dependencies installed in our virtual environment.

cat requirements.txt

It should look something like this:

blinker==1.9.0
click==8.3.1
flask==3.1.2
itsdangerous==2.2.0
jinja2==3.1.6
markupsafe==3.0.3
werkzeug==3.1.4

A single pip install can bring many dependencies, which is why it is important to constantly save the requirements.txt file.

3. Create the project structure

To start, we will create a file called app.py in the root directory of our project. This file will contain the main code of our Flask API.

touch app.py

3.1 Write the Flask API code

Open the app.py file in your favorite text editor and add the following code:

from flask import Flask, jsonify, request
app = Flask(__name__)
# Test route
@app.route('/')
def home():
    return "Hello, World! This is a Flask API."
# Route to get data
@app.route('/api/data', methods=['GET'])
def get_data():
    data = {
        'message': 'Data retrieved successfully!',
        'data': [1, 2, 3, 4, 5]
    }
    return jsonify(data)
# Route to post data
@app.route('/api/data', methods=['POST'])
def post_data():
    new_data = request.json
    response = {
        'message': 'Data received successfully!',
        'received_data': new_data
    }
    return jsonify(response), 201
if __name__ == '__main__':
    app.run(debug=True)

4. Run the Flask API

To run the Flask API, making sure we are in the virtual environment with the dependencies installed, run the following command in the terminal:

python app.py

If there are no errors, you should see something like this in the terminal:

 * Serving Flask app 'app'
 * Debug mode: on
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on http://127.0.0.1:5000
Press CTRL+C to quit
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 837-295-431

By default, Flask runs on port 5000, so we can access our API at http://127.0.0.1:5000.

5. Test the created routes

Now that our Flask API is up and running, we can test the routes we have created.

5.1 Home route

Open your web browser and navigate to http://127.0.0.1:5000. You should see the message “Hello, World! This is a Flask API.”

Hello, World! This is a Flask API.

5.2 GET route to retrieve data

To test the GET route, you can use a tool like Postman or simply your web browser. Navigate to http://127.0.0.1:5000/api/data. You should see a JSON response similar to this:

{
  "data": [
    1,
    2,
    3,
    4,
    5
  ],
  "message": "Data retrieved successfully!"
}

This was the static information we wrote in the get_data function.

5.3 POST route to send data

To test the POST route, you can use Postman or cURL. Here is how to do it with cURL from the terminal:

curl -X POST http://127.0.0.1:5000/api/data -H "Content-Type: application/json" -d '{"new_data": "This is a new data"}'

You should see a JSON response similar to this:

{
  "message": "Data received successfully!",
  "received_data": {
    "new_data": "This is a new data"
  }
}

And that’s it! You have successfully created a simple API using Flask and Python. From here, you can expand your API by adding more routes, integrating databases, handling authentication, and much more.

Just kidding, this is only the foundation to start the project.

6. Creating the real project

Now that we have the base ready with Flask, let’s start building a real API. We are going to create an API to manage a to-do list with the following features:

• RESTful CRUD routes to manage tasks.
• Storage in a MySQL/MariaDB database.
• Error handling and validations.
• Blueprints to organize the code by modules.
• System configuration in a config.py.
• Environment variables to manage secrets and sensitive configurations in a .env file.
• A few other things.

6.1 List of dependencies

For this project, we are going to use the following dependencies:

• flask: of course.
• flask-cors: to handle CORS (Cross-Origin Resource Sharing).
• flask-mail: to send emails.
• flask-sqlalchemy: to handle the database.
• flask-migrate: to handle database migrations.
• pymysql: to connect with MySQL/MariaDB.
• python-dotenv: to manage environment variables.
• gunicorn: to deploy the application in production.
• Flask-JWT-Extended: to handle authentication with JWT.

6.2 Install dependencies

Using the same virtual environment we created before, install all the necessary dependencies with the following command:

pip install flask flask-cors flask-mail flask-sqlalchemy flask-migrate pymysql python-dotenv gunicorn Flask-JWT-Extended

Don’t forget to update the requirements.txt file after installing the dependencies:

pip freeze > requirements.txt

6.3 New project structure

Let’s organize the code into a cleaner and more modular project structure. The project structure will look like this:

flask-api-demo-blog/
├── .venv/
├── app/
│   ├── __init__.py
│   ├── scripts/
│   │   └── create_db.py
│   ├── models.py
│   ├── config.py
│   ├── middlewares/
│   │   └── auth_middleware.py
│   ├── routes/
│   │   ├── tasks/
│   │   │   ├── __init__.py
│   │   │   ├── get_tasks.py
│   │   │   ├── post_task.py
│   │   │   ├── put_task.py
│   │   │   └── delete_task.py
│   │   ├── users/
│   │   │   ├── __init__.py
│   │   │   ├── get_users.py
│   │   │   ├── post_user.py
│   │   │   ├── put_user.py
│   │   │   └── delete_user.py
│   │   └── auth/
│   │       ├── __init__.py
│   │       ├── register.py
│   │       ├── login.py
│   │       ├── logout.py
│   │       └── refresh_token.py
│   └── templates/
│       └─ welcome_email.html
├── .env
├── .env.example
├── app.py
├── wsgi.py
└── requirements.txt

You can create the files as blank for now, and then we will fill them with code little by little.

6.4 Project configuration

Let’s create a config.py file inside the app/ directory to handle the application’s configuration. This file will contain the necessary configurations to connect to the database and other important settings.

import os
from dotenv import load_dotenv
load_dotenv()
class Config:
    SECRET_KEY = os.getenv('SECRET_KEY', 'mysecretkey')
    DB_HOST = os.getenv('DB_HOST', 'localhost')
    DB_PORT = os.getenv('DB_PORT', '3306')
    DB_USER = os.getenv('DB_USER', 'user')
    DB_PASSWORD = os.getenv('DB_PASSWORD', 'password')
    DB_NAME = os.getenv('DB_NAME', 'flask_api_db')
    SQLALCHEMY_DATABASE_URI = (
        f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
    )
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    MAIL_SERVER = os.getenv('MAIL_SERVER', 'smtp.gmail.com')
    MAIL_PORT = int(os.getenv('MAIL_PORT', 587))
    MAIL_USE_TLS = True
    MAIL_USERNAME = os.getenv('MAIL_USERNAME')
    MAIL_PASSWORD = os.getenv('MAIL_PASSWORD')
    MAIL_USE_TLS =  os.getenv('MAIL_USE_TLS', 'True')
    MAIL_USE_SSL =  os.getenv('MAIL_USE_SSL', 'False')
    JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY', 'myjwtsecretkey')

6.5 Environment variables

Let’s create a .env file in the root of the project to store sensitive environment variables. Make sure not to upload this file to your public repository.

SECRET_KEY=mysecretkey

DB_HOST=localhost
DB_PORT=3306
DB_USER=user
DB_PASSWORD=password
DB_NAME=flask_api_db

JWT_SECRET_KEY=myjwtsecretkey

MAIL_SERVER=smtp.gmail.com
MAIL_PORT=587
MAIL_USERNAME=your_email@gmail.com
MAIL_PASSWORD=your_email_password
MAIL_USE_TLS=True
MAIL_USE_SSL=False

We can also create the .env.example file so that other developers know which environment variables they need to configure or when deploying to production.

SECRET_KEY=mysecretkey

DB_HOST=localhost
DB_PORT=3306
DB_USER=user
DB_PASSWORD=password
DB_NAME=flask_api_db

JWT_SECRET_KEY=myjwtsecretkey

MAIL_SERVER=smtp.gmail.com
MAIL_PORT=587
MAIL_USERNAME=your_email@gmail.com
MAIL_PASSWORD=your_email_password
MAIL_USE_TLS=True
MAIL_USE_SSL=False

6.6 Modify app.py

Now with the new structure we need to modify our old and simple app.py to load the application from the app package.

from app import create_app
app = create_app()
if __name__ == '__main__':
    app.run(debug=True)

Replace all the previous code, that is, the endpoints we created, the entire file with these lines.

6.7 Create __init__.py file in app/

This file is important, it is the entry point to create the Flask application and register routes and extensions.

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_cors import CORS
from flask_mail import Mail
from .config import Config

db = SQLAlchemy()
migrate = Migrate()
mail = Mail()
cors = CORS()

def create_app():
    app = Flask(__name__)
    app.config.from_object(Config)

    db.init_app(app)
    migrate.init_app(app, db)
    mail.init_app(app)
    cors.init_app(app)

   # from app.routes.tasks import tasks_bp
   # app.register_blueprint(tasks_bp)

    return app

Let’s review the code to explain it a bit:

• Imports: The base Flask extensions that will be used are imported, such as SQLAlchemy for the database, Migrate for migrations, CORS to handle CORS, and Mail to send emails. The configuration is also imported from the config.py file to load the necessary environment variables and configurations.

• Extension initialization: Instances of the extensions that will be used in the Flask application are created, such as db, migrate, mail, and cors. This allows Flask to load the necessary modules to work with the database, migrations, email, and CORS.

• Function create_app(): This function is the entry point to create the Flask application. Inside this function:

  • An instance of the Flask application is created.
  • The configuration is loaded from the Config object.
  • The extensions are initialized with the Flask application.

• Blueprints (commented out for now): These two comments indicate that in the future we will import those files, which do not exist yet, but we will create them shortly. If we run the code uncommenting those lines and those files do not exist, it will give us an error.

6.8 Create the data model

First, we will create the User model and then the Task model, to associate each task with a user through the foreign key user_id. Create the models.py file inside the app/ directory and add the following code:

from . import db
from datetime import datetime

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(50), nullable=False)
    email = db.Column(db.String(100), unique=True, nullable=False)
    password = db.Column(db.String(200), nullable=False)

    tasks = db.relationship('Task', backref='user', lazy=True)

    def to_dict(self):
        return {
            'id': self.id,
            'username': self.username,
            'email': self.email
        }

class Task(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    description = db.Column(db.String(200), nullable=True)
    completed = db.Column(db.Boolean, default=False)
    date_created = db.Column(db.DateTime, server_default=db.func.now())
    date_completed = db.Column(db.DateTime, nullable=True)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)

    def to_dict(self):
        return {
            'id': self.id,
            'title': self.title,
            'description': self.description,
            'completed': self.completed,
            'date_created': self.date_created,
            'date_completed': self.date_completed,
            'user_id': self.user_id
        }

class RefreshToken(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False)
    token = db.Column(db.String(500), nullable=False, unique=True)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    expires_at = db.Column(db.DateTime, nullable=False)
    revoked = db.Column(db.Boolean, default=False)

    def to_dict(self):
        return {
            'id': self.id,
            'user_id': self.user_id,
            'token': self.token,
            'created_at': self.created_at,
            'expires_at': self.expires_at,
            'revoked': self.revoked
        }

Explanation of the code:

• First, the User model is defined, representing a user in the system. In addition to the basic fields, the tasks relationship is added to easily access the tasks associated with each user.
• Then, the Task model is defined, which now includes the user_id field as a foreign key, associating each task with a specific user.
• Both models include the to_dict() method to facilitate JSON serialization.
• Finally, the RefreshToken model is defined to handle refresh tokens in JWT authentication for later use.

6.9 Create script to create / delete the database

To avoid having to do everything manually or complicate things, I have created a simple script with which we can create the database and the tables defined in the models, or delete everything using a flag. Create the scripts/ directory inside app/ and then create the create_db.py file with the following code:

import os
import sys

project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, project_root)

from app import create_app
from app.models import db

def init_database():
    """Initialize the database by creating tables."""
    app = create_app()
    with app.app_context():
        print("Creating tables...")
        db.create_all()
        print("Tables created successfully")

def reset_database():
    """Delete all tables and recreate them."""
    app = create_app()
    with app.app_context():
        print("Deleting tables...")
        db.drop_all()
        print("Tables deleted")
        init_database()

if __name__ == "__main__":
    if len(sys.argv) > 1:
        flag = sys.argv[1]
        if flag == "--reset":
            reset_database()
        else:
            print(f"Invalid flag: {flag}")
            print("Valid flag: --reset")
            print("No action was performed.")
    else:
        init_database()

Explanation of the code:

• First, the sys.path is adjusted to ensure the script can correctly import the project’s modules.
• The init_database() function is defined to create the tables in the database using SQLAlchemy.
• The reset_database() function is defined to delete all existing tables and then call init_database() to recreate them.
• In the if __name__ == "__main__": block, it checks if a flag has been passed to the script. If the flag is --reset, it calls reset_database(). If no flag is passed, it calls init_database() to create the tables.

6.10 Run the script to create the database

To run the script and create the database along with the tables defined in the models, use the following command in the terminal:

python app/scripts/create_db.py

This will create the tables in the database according to the models defined in models.py.

6.11 Delete and recreate the database

If at any point you need to delete all tables and recreate them from scratch, you can run the script with the --reset flag as follows:

python app/scripts/create_db.py --reset

This will delete all existing tables and then recreate them according to the defined models. For example, if in the future you modify the models and need to update everything, simply run this command to reset the database.

6.12 Create routes (blueprints) for tasks

Now we will create the routes to handle CRUD operations for tasks. We will organize the routes in a blueprint, which will allow us to keep the code modular and organized.

Create the tasks/ directory inside app/routes/ and then create the following files:

• __init__.py

from flask import Blueprint
tasks_bp = Blueprint('tasks', __name__)

from . import get_tasks
from . import post_task
from . import put_task
from . import delete_task

• get_tasks.py

from flask import jsonify
from app import db
from app.models import Task
from . import tasks_bp

@tasks_bp.route('/', methods=['GET'])
def get_tasks():
    tasks = Task.query.all()
    return jsonify([task.to_dict() for task in tasks])

@tasks_bp.route('/<int:task_id>', methods=['GET'])
def get_task(task_id):
    task = Task.query.get_or_404(task_id)
    return jsonify(task.to_dict())

• post_task.py

from flask import request, jsonify
from app import db
from app.models import Task
from . import tasks_bp

@tasks_bp.route('/', methods=['POST'])
def post_task():
    data = request.json
    new_task = Task(
        title=data['title'],
        description=data.get('description', ''),
        user_id=data['user_id']
    )
    db.session.add(new_task)
    db.session.commit()
    return jsonify(new_task.to_dict()), 201

• put_task.py

from flask import request, jsonify
from app import db
from app.models import Task
from datetime import datetime
from . import tasks_bp

@tasks_bp.route('/<int:task_id>', methods=['PUT'])
def put_task(task_id):
    task = Task.query.get_or_404(task_id)
    data = request.json
    task.title = data.get('title', task.title)
    task.description = data.get('description', task.description)
    task.completed = data.get('completed', task.completed)
    task.date_completed = datetime.utcnow() if task.completed else None 
    db.session.commit()
    return jsonify(task.to_dict())

• delete_task.py

from flask import jsonify
from app import db
from app.models import Task
from . import tasks_bp

@tasks_bp.route('/<int:task_id>', methods=['DELETE'])
def delete_task(task_id):
    task = Task.query.get_or_404(task_id)
    db.session.delete(task)
    db.session.commit()
    return jsonify({'message': 'Tarea eliminada con éxito.'})

Explanation of the code:

• __init__.py: Here the tasks_bp blueprint is created and the routes defined in the other files are imported.
• get_tasks.py: Defines the routes to get all tasks and a specific task by its ID.
• post_task.py: Defines the route to create a new task.
• put_task.py: Defines the route to update an existing task.
• delete_task.py: Defines the route to delete a task.

These routes have access to the database through SQLAlchemy and use the models defined earlier to interact with the data.

6.13 Update __init__.py to register the blueprint

Uncomment the lines in the app/__init__.py file to register the tasks blueprint:

# from app.routes.tasks import tasks_bp
# app.register_blueprint(tasks_bp)

Simply remove the comments:

from app.routes.tasks import tasks_bp
app.register_blueprint(tasks_bp)

Optional: Create a route prefix for the blueprint

If you want all task routes to have a common prefix, such as /tasks, you can modify the blueprint registration line in app/__init__.py as follows:

app.register_blueprint(tasks_bp, url_prefix='/tasks')

Now all routes defined in the tasks blueprint will be accessible under the /tasks prefix. For example, if your route was /my-route, it will now be /tasks/my-route.

We have the tasks part created, but we still won’t be able to create tasks because we need to create users. In the next section, we will create the routes to manage users.

6.14 Create the routes (blueprints) for users

Now we will create the routes to handle CRUD operations for users. Just like with tasks, we will organize the routes in a blueprint.

Create the users/ directory inside app/routes/ and then create the following files:

• __init__.py

from flask import Blueprint
users_bp = Blueprint('users', __name__)
from . import get_users
from . import post_user
from . import put_user
from . import delete_user

• get_users.py

from flask import jsonify
from app import db
from app.models import User
from . import users_bp
@users_bp.route('/', methods=['GET'])
def get_users():
    users = User.query.all()
    return jsonify([user.to_dict() for user in users])
@users_bp.route('/<int:user_id>', methods=['GET'])
def get_user(user_id):
    user = User.query.get_or_404(user_id)
    return jsonify(user.to_dict())

• post_user.py

from flask import request, jsonify
from app import db
from app.models import User
from . import users_bp
@users_bp.route('/', methods=['POST'])
def post_user():
    data = request.json
    new_user = User(
        username=data['username'],
        email=data['email'],
        password=data['password']
    )
    db.session.add(new_user)
    db.session.commit()
    return jsonify(new_user.to_dict()), 201

• put_user.py

from flask import request, jsonify
from app import db
from app.models import User
from . import users_bp
@users_bp.route('/<int:user_id>', methods=['PUT'])
def put_user(user_id):
    user = User.query.get_or_404(user_id)
    data = request.json
    user.username = data.get('username', user.username)
    user.email = data.get('email', user.email)
    user.password = data.get('password', user.password)
    db.session.commit()
    return jsonify(user.to_dict())

• delete_user.py

from flask import jsonify
from app import db
from app.models import User
from . import users_bp
@users_bp.route('/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    user = User.query.get_or_404(user_id)
    db.session.delete(user)
    db.session.commit()
    return jsonify({'message': 'Usuario eliminado con éxito.'})

Explanation of the code:

• __init__.py: Here the users_bp blueprint is created and the routes defined in the other files are imported.
• get_users.py: Defines the routes to get all users and a specific user by their ID.
• post_user.py: Defines the route to create a new user.
• put_user.py: Defines the route to update an existing user.
• delete_user.py: Defines the route to delete a user.

Basically, we repeat the same pattern as with tasks, but now for users, with their own rules and logic. This time we use the /users prefix for user routes. That is, the route to get all users will be /users/, and so on.

6.15 Update __init__.py to register the users blueprint

We need to do the same as we did with tasks, but now for users. Uncomment the lines in the app/__init__.py file to register the users blueprint:

from app.routes.users import users_bp
app.register_blueprint(users_bp, url_prefix='/users')

Add these lines just below the tasks blueprint registration.

And with this, we already have a semi-functional app to manage users and tasks, first creating users and then creating tasks associated with those users. Now let’s test that this works to continue.

7. Test the created routes

So far, if everything has gone well, we have the following routes:

• /users/ [GET, POST]
• /users/<int:user_id> [GET, PUT, DELETE]
• /tasks/ [GET, POST]
• /tasks/<int:task_id> [GET, PUT, DELETE]

We can test these routes using Postman, cURL, or any other tool for making HTTP requests. Make sure the Flask application is running before testing the routes.

For this simple example, we will use cURL from the terminal to test the routes.

7.1 Test user routes

• Create a new user:

curl -X POST http://localhost:5000/users/ -H "Content-Type: application/json" -d '{"username": "newuser", "email": "newuser@example.com", "password": "password123"}'

If everything goes well, the system will respond with the data of the newly created user:

{
  "email": "newuser@example.com",
  "id": 1,
  "username": "newuser"
}

• Get all users:

curl http://localhost:5000/users/

This will return a list of all users in JSON format:

[
  {
    "email": "newuser@example.com",
    "id": 1,
    "username": "newuser"
  }
]

• Get a specific user by ID:

curl http://localhost:5000/users/1

This will return the data of the user with ID 1:

{
  "email": "newuser@example.com",
  "id": 1,
  "username": "newuser"
}

• Update a user:

curl -X PUT http://localhost:5000/users/1 -H "Content-Type: application/json" -d '{"username": "updateduser", "email": "updateduser@example.com", "password": "newpassword123"}'

This will return the updated user data:

{
  "email": "updateduser@example.com",
  "id": 1,
  "username": "updateduser"
}

• Delete a user:

curl -X DELETE http://localhost:5000/users/1

This will return a confirmation message:

{
  "message": "User successfully deleted."
}

7.2 Test task routes

• Create a new task (make sure to use a valid user_id):

curl -X POST http://localhost:5000/tasks/ -H "Content-Type: application/json" -d '{"title": "New Task", "description": "This is a new task", "user_id": 1}'

Note: Make sure the user_id you use exists in the database. If you deleted the user created in the previous step, create a new one before creating the task and check the ID it has.

This will return the data of the newly created task:

{
  "completed": false,
  "date_completed": null,
  "date_created": "Sat, 29 Nov 2025 03:19:52 GMT",
  "description": "This is a new task",
  "id": 2,
  "title": "New Task",
  "user_id": 2
}

The ID here is 2 because ID 1 was deleted, but if you haven’t deleted anything, it will be 1 hehe.

• Get all tasks:

curl http://localhost:5000/tasks/

This will return a list of all tasks in JSON format:

[
  {
    "completed": false,
    "date_completed": null,
    "date_created": "Sat, 29 Nov 2025 03:19:52 GMT",
    "description": "This is a new task",
    "id": 2,
    "title": "New Task",
    "user_id": 2
  }
]

• Get a specific task by ID:

curl http://localhost:5000/tasks/2

This will return the data of the task with ID 2:

{
  "completed": false,
  "date_completed": null,
  "date_created": "Sat, 29 Nov 2025 03:19:52 GMT",
  "description": "This is a new task",
  "id": 2,
  "title": "New Task",
  "user_id": 2
}

• Update a task:

curl -X PUT http://localhost:5000/tasks/2 -H "Content-Type: application/json" -d '{"title": "Updated Task", "description": "This task has been updated", "completed": true}'

This will return the updated task data:

{
  "completed": true,
  "date_completed": "Sat, 29 Nov 2025 06:24:47 GMT",
  "date_created": "Sat, 29 Nov 2025 03:19:52 GMT",
  "description": "This task has been updated",
  "id": 2,
  "title": "Updated Task",
  "user_id": 2
}

Note: The value of date_completed will be automatically updated when the task is marked as completed, i.e., if completed is true. If it is sent as false, it will be set to null.

• Delete a task:

curl -X DELETE http://localhost:5000/tasks/2

This will return a confirmation message:

{
  "message": "Task successfully deleted."
}

With this, we have confirmed that the API is working correctly, but don’t rest yet—there’s still more to do! Next, we’ll implement error handling, validations, and a few other important details in the following section.

8. Implementation of error handling and validations

So far we have created a basic API with Flask and Python, which by copying and pasting the shared code will work without problems. However, there is no data validation or error handling, which can lead to problems and critical errors in a production environment.

8.1 User error handling

Let’s start with error handling in the user routes. We will add validations to the routes and ensure that the received data is correct.

8.1.1 Update post_user.py for validations

First, let’s update the post_user.py file to add validations when creating a new user:

from flask import request, jsonify
from app import db
from app.models import User
from . import users_bp
import re
from werkzeug.security import generate_password_hash

def is_password_secure(password, min_length=8, max_length=32, allowed_chars=r'^[A-Za-z0-9@#$%^&+=]*$'):
    """
    Check if the password meets security requirements.
    - min_length: minimum length
    - max_length: maximum length
    - allowed_chars: regex of allowed characters
    """
    if not isinstance(password, str):
        return False, 'Password must be a string.'
    if len(password) < min_length:
        return False, f'Password must be at least {min_length} characters.'
    if len(password) > max_length:
        return False, f'Password must be at most {max_length} characters.'
    if not re.match(allowed_chars, password):
        return False, 'Password contains invalid characters.'
    return True, None

@users_bp.route('/', methods=['POST'])
def post_user():

    if not request.is_json:
        return jsonify({'error': 'Request body must be JSON.'}), 400

    data = request.json

    # Validate required fields and collect errors
    required_fields = ['username', 'email', 'password', 'confirm_password']
    errors = {}
    for field in required_fields:
        if field not in data or not data[field]:
            errors[field] = f'The field {field} is required.'

    # Validate password confirmation
    if 'password' in data and 'confirm_password' in data:
        if data['password'] != data['confirm_password']:
            errors['confirm_password'] = 'Passwords do not match.'

    # Validate password security
    if 'password' in data and data.get('password'):
        secure, msg = is_password_secure(
            data['password'],
            min_length=8,      # Modify this as needed
            max_length=32,
            allowed_chars=r'^[A-Za-z0-9@#$%^&+=]*$'
        )
        if not secure:
            errors['password'] = msg

    # Validate email uniqueness
    if 'email' in data and data.get('email'):
        if User.query.filter_by(email=data['email']).first():
            errors['email'] = 'The email is already registered.'

    if errors:
        return jsonify({'errors': errors}), 422

    # Encrypt password and create user
    hashed_password = generate_password_hash(data['password'])

    new_user = User(
        username=data['username'],
        email=data['email'],
        password=hashed_password
    )
    db.session.add(new_user)
    db.session.commit()
    return jsonify({'message': 'User created successfully.', 'user': new_user.to_dict()}), 201

Explanation of the code:

• A function is_password_secure is added to validate the security of the password according to the specified criteria. It is modifiable as needed.
• It checks that the request body is JSON.
• Required fields are validated and errors are collected in an errors dictionary, returning a 422 status code if there are errors. Yes, it is also usually done with 400, but 422 is more specific for failed validations and you can send a dictionary with all the errors found.
• It validates that the passwords match and that the email is not already registered.
• If there are no errors, the new user is created and saved in the database.
• The password is encrypted before storing it using werkzeug.security.generate_password_hash for added security. MANDATORY.
• When creating the user, a success message is returned along with the created user’s data. With the status code 201 (Created).

We go from creating a simple form that received data without validation to having a robust system that validates input data and handles errors properly. At least you are starting to get an idea of how to do it.

8.1.2 Update put_user.py for validations

Now let’s update the put_user.py file to add validations when updating an existing user:

from flask import request, jsonify
from app import db
from app.models import User
from . import users_bp
import re
from werkzeug.security import generate_password_hash

def is_password_secure(password, min_length=8, max_length=32, allowed_chars=r'^[A-Za-z0-9@#$%^&+=]*$'):
    """
    Check if the password meets security requirements.
    - min_length: minimum length
    - max_length: maximum length
    - allowed_chars: regex of allowed characters
    """
    if not isinstance(password, str):
        return False, 'Password must be a string.'
    if len(password) < min_length:
        return False, f'Password must be at least {min_length} characters.'
    if len(password) > max_length:
        return False, f'Password must be at most {max_length} characters.'
    if not re.match(allowed_chars, password):
        return False, 'Password contains invalid characters.'
    return True, None

@users_bp.route('/<int:user_id>', methods=['PUT'])
def put_user(user_id):
    if not request.is_json:
        return jsonify({'error': 'Request body must be JSON.'}), 400

    data = request.json
    user = User.query.get(user_id)
    if not user:
        return jsonify({'error': 'User not found.'}), 404

    errors = {}

    # Validate email uniqueness if email is being updated
    if 'email' in data and data.get('email') and data['email'] != user.email:
        if User.query.filter_by(email=data['email']).first():
            errors['email'] = 'The email is already registered.'

    # Validate password security if password is being updated
    if 'password' in data and data.get('password'):
        secure, msg = is_password_secure(
            data['password'],
            min_length=8,
            max_length=32,
            allowed_chars=r'^[A-Za-z0-9@#$%^&+=]*$'
        )
        if not secure:
            errors['password'] = msg

    if errors:
        return jsonify({'errors': errors}), 422

    # Update user fields
    user.username = data.get('username', user.username)
    user.email = data.get('email', user.email)

    # Encrypt password if it's being updated
    if 'password' in data and data.get('password'):
        user.password = generate_password_hash(data['password'])

    db.session.commit()
    return jsonify({'message': 'User updated successfully.', 'user': user.to_dict()}), 200

Explanation of the code:

• The is_password_secure function is reused to validate password security, just like before.
• It checks that the request body is JSON.
• It validates the fields being updated and collects errors in an errors dictionary, returning a 422 status code if there are errors.
• It validates that the email is not already registered if it is being updated.
• If there are no errors, the user fields are updated and saved in the database.
• It encrypts the password if it is being updated.
• When updating the user, it returns a success message along with the updated user data. With status code 200 (OK).
• If the user does not exist, it returns a JSON with a 404 error.

I know that changing the password is usually done in a separate route, with email validation and so on, but to simplify the example I left it like this.

8.1.3 Update get_users.py for error handling and pagination

This is a simple example, but in a real system you can easily have hundreds or thousands of users, so it is important to handle errors properly. Let’s update the get_users.py file to handle errors and have functional pagination:

from flask import jsonify, request
from app import db
from app.models import User
from . import users_bp

@users_bp.route('/', methods=['GET'])
def get_users():
    # Pagination parameters
    try:
        page = int(request.args.get('page', 1))
        per_page = int(request.args.get('per_page', 10))
        if page < 1 or per_page < 1 or per_page > 100:
            raise ValueError
    except (ValueError, TypeError):
        return jsonify({'error': 'Invalid pagination parameters'}), 400

    pagination = User.query.paginate(page=page, per_page=per_page, error_out=False)
    users = [user.to_dict() for user in pagination.items]
    response = {
        'users': users,
        'pagination': {
            'total': pagination.total,
            'pages': pagination.pages,
            'page': pagination.page,
            'per_page': pagination.per_page,
            'has_next': pagination.has_next,
            'has_prev': pagination.has_prev
        }
    }
    return jsonify(response)

@users_bp.route('/<int:user_id>', methods=['GET'])
def get_user(user_id):
    user = User.query.get(user_id)
    if not user:
        return jsonify({'message': 'User not found'}), 404
    return jsonify(user.to_dict())

Explanation of the code:

• In the get_users route, pagination parameters (page and per_page) are added to limit the number of users returned in a single request. Errors are handled if the parameters are invalid.
• In the get_user route, the case where the user does not exist is handled, returning an error message with a 404 status code. The get_or_404 method is no longer used to have more explicit error control. Now we return a JSONified message, so the client side can handle it visually if desired.

8.1.4 Update delete_user.py for error handling

Let’s update the delete_user.py file to handle errors when deleting a user:

from flask import jsonify
from app import db
from app.models import User
from . import users_bp

@users_bp.route('/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    user = User.query.get(user_id)
    if not user:
        return jsonify({'message': 'User not found'}), 404
    db.session.delete(user)
    db.session.commit()
    return jsonify({'message': 'User deleted successfully.'})

Explanation of the code:

• The case where the user does not exist is handled, returning an error message with a 404 status code. As in the previous endpoint, the get_or_404 method is no longer used to have more explicit error control.

8.2 Error handling for tasks

Now let’s implement similar error handling for the task routes. We will update each corresponding file to add validations and error handling.

8.2.1 Update post_task.py for validations

First, let’s update the post_task.py file to add validations when creating a new task:

from flask import request, jsonify
from app import db
from app.models import Task, User
from . import tasks_bp

@tasks_bp.route('/', methods=['POST'])
def post_task():
    if not request.is_json:
        return jsonify({'error': 'Request body must be JSON.'}), 400
    data = request.json
    # Validate required fields and collect errors
    required_fields = ['title', 'user_id']
    errors = {}
    for field in required_fields:
        if field not in data or not data[field]:
            errors[field] = f'The field {field} is required.'
    # Validate user_id exists
    if 'user_id' in data and data.get('user_id'):
        if not User.query.get(data['user_id']):
            errors['user_id'] = 'The specified user_id does not exist.'
    if errors:
        return jsonify({'errors': errors}), 422
    new_task = Task(
        title=data['title'],
        description=data.get('description', ''),
        user_id=data['user_id']
    )
    db.session.add(new_task)
    db.session.commit()
    return jsonify({'message': 'Task created successfully.', 'task': new_task.to_dict()}), 201

Explanation of the code:

• It checks that the request body is JSON, as we did with Users.
• It validates the required fields and collects errors in an errors dictionary, returning a 422 status code if there are errors.
• It validates that the provided user_id exists in the database. This way, the system won’t have issues with foreign key violations.
• If there are no errors, it creates the new task and saves it to the database.
• When creating the task, it returns a success message along with the created task data. With the status code 201.

8.2.2 Update put_task.py for validations

from flask import request, jsonify
from app import db
from app.models import Task, User
from datetime import datetime
from . import tasks_bp

@tasks_bp.route('/<int:task_id>', methods=['PUT'])
def put_task(task_id):
    if not request.is_json:
        return jsonify({'error': 'Request body must be JSON.'}), 400
    data = request.json
    task = Task.query.get(task_id)
    if not task:
        return jsonify({'error': 'Task not found.'}), 404
    errors = {}
    # Validate user_id exists if being updated
    if 'user_id' in data and data.get('user_id'):
        if not User.query.get(data['user_id']):
            errors['user_id'] = 'The specified user_id does not exist.'
    if errors:
        return jsonify({'errors': errors}), 422
    # Update task fields
    task.title = data.get('title', task.title)
    task.description = data.get('description', task.description)
    task.completed = data.get('completed', task.completed)
    task.date_completed = datetime.utcnow() if task.completed else None
    if 'user_id' in data and data.get('user_id'):
        task.user_id = data['user_id']
    db.session.commit()
    return jsonify({'message': 'Task updated successfully.', 'task': task.to_dict()}), 200

Explanation of the code:

• It checks that the request body is JSON.
• It validates the fields being updated and collects errors in an errors dictionary, returning a 422 status code if there are errors.
• It validates that the provided user_id exists in the database if it is being updated.
• If there are no errors, it updates the task fields and saves it to the database.
• When updating the task, it returns a success message along with the updated task data. With the status code 200.
• If the task does not exist, it returns a json with a 404 error.

8.2.3 Update get_tasks.py for error handling and pagination

from flask import jsonify, request
from app import db
from app.models import Task
from . import tasks_bp

@tasks_bp.route('/', methods=['GET'])
def get_tasks():
    # Pagination parameters
    try:
        page = int(request.args.get('page', 1))
        per_page = int(request.args.get('per_page', 10))
        if page < 1 or per_page < 1 or per_page > 100:
            raise ValueError
    except (ValueError, TypeError):
        return jsonify({'error': 'Invalid pagination parameters'}), 400
    pagination = Task.query.paginate(page=page, per_page=per_page, error_out=False)
    tasks = [task.to_dict() for task in pagination.items]
    response = {
        'tasks': tasks,
        'pagination': {
            'total': pagination.total,
            'pages': pagination.pages,
            'page': pagination.page,
            'per_page': pagination.per_page,
            'has_next': pagination.has_next,
            'has_prev': pagination.has_prev
        }
    }
    return jsonify(response)
@tasks_bp.route('/<int:task_id>', methods=['GET'])
def get_task(task_id):
    task = Task.query.get(task_id)
    if not task:
        return jsonify({'message': 'Task not found'}), 404
    return jsonify(task.to_dict())

Explanation of the code:

• In the get_tasks route, pagination parameters (page and per_page) are added to limit the number of tasks returned in a single request. Errors are handled if the parameters are not valid.
• In the get_task route, the case where the task does not exist is handled, returning an error message with a 404 status code.

8.2.4 Update delete_task.py for error handling

from flask import jsonify
from app import db
from app.models import Task
from . import tasks_bp

@tasks_bp.route('/<int:task_id>', methods=['DELETE'])
def delete_task(task_id):
    task = Task.query.get(task_id)
    if not task:
        return jsonify({'message': 'Task not found'}), 404
    db.session.delete(task)
    db.session.commit()
    return jsonify({'message': 'Task deleted successfully.'}), 200

Explanation of the code:

• The case where the task does not exist is handled, returning an error message with a 404 status code. As in the previous endpoints, we no longer use the get_or_404 method to have more explicit error control.

Perfect! With this, we have significantly improved the robustness of our API by adding validations and error handling to the user and task routes. Now the system should not easily break due to non-existent data such as users or tasks, and input data is validated to ensure it meets the expected requirements, even protecting user integrity by encrypting passwords.

Next, we will focus on creating a new authentication and authorization module to protect certain routes and ensure that only authenticated users can access them. We will use JWT (JSON Web Tokens) for this purpose.

9. Implement Authentication and Authorization with JWT Tokens

To implement authentication and authorization in our Flask API, we will use JSON Web Tokens (JWT). JWT is an open standard that defines a compact and self-contained way to securely transmit information between parties as a JSON object. It is widely used for authentication in web applications.

9.1 Create a new auth/ module to handle authentication

For this example, we will do everything with blueprints, just like we did with users and tasks. We will create a new auth/ directory inside app/routes/ and then create the following files:

• __init__.py

from flask import Blueprint
auth_bp = Blueprint('auth', __name__)
from . import register
from . import login
from . import logout
from . import refresh_token

• register.py

from flask import request, jsonify
from app import db
from app.models import User
from . import auth_bp
import re
from werkzeug.security import generate_password_hash

def is_password_secure(password, min_length=8, max_length=32, allowed_chars=r'^[A-Za-z0-9@#$%^&+=]*$'):
    """
    Check if the password meets security requirements.
    - min_length: minimum length
    - max_length: maximum length
    - allowed_chars: regex of allowed characters
    """
    if not isinstance(password, str):
        return False, 'Password must be a string.'
    if len(password) < min_length:
        return False, f'Password must be at least {min_length} characters.'
    if len(password) > max_length:
        return False, f'Password must be at most {max_length} characters.'
    if not re.match(allowed_chars, password):
        return False, 'Password contains invalid characters.'
    return True, None

@auth_bp.route('/register', methods=['POST'])
def register():
    if not request.is_json:
        return jsonify({'error': 'Request body must be JSON.'}), 400
    data = request.json
    # Validate required fields and collect errors
    required_fields = ['username', 'email', 'password', 'confirm_password']
    errors = {}
    for field in required_fields:
        if field not in data or not data[field]:
            errors[field] = f'The field {field} is required.'
    # Validate password confirmation
    if 'password' in data and 'confirm_password' in data:
        if data['password'] != data['confirm_password']:
            errors['confirm_password'] = 'Passwords do not match.'
    # Validate password security
    if 'password' in data and data.get('password'):
        secure, msg = is_password_secure(
            data['password'],
            min_length=8,
            max_length=32,
            allowed_chars=r'^[A-Za-z0-9@#$%^&+=]*$'
        )
        if not secure:
            errors['password'] = msg
    # Validate email uniqueness
    if 'email' in data and data.get('email'):
        if User.query.filter_by(email=data['email']).first():
            errors['email'] = 'The email is already registered.'
    if errors:
        return jsonify({'errors': errors}), 422
    # Encrypt password and create user
    hashed_password = generate_password_hash(data['password'])
    new_user = User(
        username=data['username'],
        email=data['email'],
        password=hashed_password
    )
    db.session.add(new_user)
    db.session.commit()
    return jsonify({'message': 'User registered successfully.', 'user': new_user.to_dict()}), 201

Explanation of the code:

• An auth_bp blueprint is created to handle authentication routes.
• In register.py, the /register route is defined to register new users.
• The is_password_secure function is reused to validate password security.
• Similar validations to those in post_user.py are performed to ensure input data is correct.
• The password is encrypted before storing it using werkzeug.security.generate_password_hash.
• Upon successful registration, a success message is returned along with the created user’s data, using status code 201 (Created).

Wait… isn’t this the same as post_user.py? Yes, but in this case, it’s done in a separate module to separate responsibilities. In a real app, you might have more specific logic for user registration, such as sending verification emails, etc. Ideally, you can disable post_user once the auth module is working.

With this understanding, let’s continue with the next file.

• login.py

from flask import request, jsonify, current_app
from app.models import User, RefreshToken
from . import auth_bp
from werkzeug.security import check_password_hash
import jwt
import datetime
from app import db

def generate_tokens(user_id):
    access_token = jwt.encode({
        'user_id': user_id,
        'exp': datetime.datetime.utcnow() + datetime.timedelta(minutes=15)
    }, current_app.config['JWT_SECRET_KEY'], algorithm='HS256')

    refresh_exp = datetime.datetime.utcnow() + datetime.timedelta(days=7)
    refresh_token = jwt.encode({
        'user_id': user_id,
        'exp': refresh_exp
    }, current_app.config['JWT_SECRET_KEY'], algorithm='HS256')

    # Store refresh token in DB
    new_refresh = RefreshToken(
        user_id=user_id,
        token=refresh_token,
        created_at=datetime.datetime.utcnow(),
        expires_at=refresh_exp,
        revoked=False
    )
    db.session.add(new_refresh)
    db.session.commit()

    return access_token, refresh_token

@auth_bp.route('/login', methods=['POST'])
def login():
    if not request.is_json:
        return jsonify({'error': 'Request body must be JSON.'}), 400
    data = request.json

    # Validate required fields and collect errors
    required_fields = ['email', 'password']
    errors = {}
    for field in required_fields:
        if field not in data or not data[field]:
            errors[field] = f'The field {field} is required.'

    if errors:
        return jsonify({'errors': errors}), 422

    user = User.query.filter_by(email=data['email']).first()
    if not user or not check_password_hash(user.password, data['password']):
        return jsonify({'error': 'Invalid email or password.'}), 401

    access_token, refresh_token = generate_tokens(user.id)
    return jsonify({
        'message': 'Login successful.',
        'access_token': access_token,
        'refresh_token': refresh_token
    }), 200

Explanation of the code:

• In login.py, the /login route is defined to authenticate users.

• It checks that the request body is JSON and contains the email and password fields.

• It looks up the user in the database and verifies the password using werkzeug.security.check_password_hash.

• If authentication is successful, an access token and a refresh token are generated using JWT.

• The refresh token is stored in the database for later validation.

• A success message is returned along with the generated tokens.

• If there are authentication errors or missing data, an error message is returned with the list of corresponding errors and status code 422.

• logout.py

from flask import request, jsonify
from app.models import RefreshToken
from . import auth_bp
from app import db

@auth_bp.route('/logout', methods=['POST'])
def logout():
    data = request.json
    refresh_token = data.get('refresh_token')
    if not refresh_token:
        return jsonify({'error': 'Refresh token required.'}), 400

    token_obj = RefreshToken.query.filter_by(token=refresh_token, revoked=False).first()
    if not token_obj:
        return jsonify({'error': 'Invalid or already revoked refresh token.'}), 400

    token_obj.revoked = True
    db.session.commit()
    return jsonify({'message': 'Logout successful. Token revoked.'}), 200

Explanation of the code:

• In logout.py, the /logout route is defined to log out users.

• The refresh token is received in the request and validated to ensure it exists and is not revoked.

• If the token is valid, it is marked as revoked in the database.

• A success message is returned indicating that the token has been revoked.

• refresh_token.py

from flask import request, jsonify, current_app
from app.models import RefreshToken
from . import auth_bp
import jwt
import datetime
from app import db

@auth_bp.route('/refresh', methods=['POST'])
def refresh():
    data = request.json
    refresh_token = data.get('refresh_token')
    if not refresh_token:
        return jsonify({'error': 'Refresh token required.'}), 400

    token_obj = RefreshToken.query.filter_by(token=refresh_token, revoked=False).first()
    if not token_obj or token_obj.expires_at < datetime.datetime.utcnow():
        return jsonify({'error': 'Invalid or expired refresh token.'}), 401

    try:
        payload = jwt.decode(refresh_token, current_app.config['JWT_SECRET_KEY'], algorithms=['HS256'])
        user_id = payload['user_id']
    except jwt.ExpiredSignatureError:
        return jsonify({'error': 'Refresh token expired.'}), 401
    except jwt.InvalidTokenError:
        return jsonify({'error': 'Invalid refresh token.'}), 401

    # Generate new access token
    access_token = jwt.encode({
        'user_id': user_id,
        'exp': datetime.datetime.utcnow() + datetime.timedelta(minutes=15)
    }, current_app.config['JWT_SECRET_KEY'], algorithm='HS256')

    return jsonify({'access_token': access_token}), 200

Explanation of the code:

• In refresh_token.py, the /refresh route is defined to obtain a new access token using a valid refresh token.
• The refresh token is received in the request and validated to ensure it exists, is not revoked, and has not expired.
• The refresh token is decoded to obtain the user_id.
• If the token is valid, a new access token is generated and returned in the response.

9.2 Modify app/__init__.py to register the Auth blueprint

Just like we did before with Users and Tasks, now we register the Auth blueprint in app/__init__.py:

from app.routes.auth import auth_bp
app.register_blueprint(auth_bp, url_prefix='/auth')

Simply add these two lines below the previous register blueprint and that’s it.

10. Test authentication routes

Now that we have implemented the authentication module, it’s time to test the routes we have created. Make sure your Flask server is running before performing the tests.

10.1 Test registration route

• Register a new user:

curl -X POST http://localhost:5000/auth/register -H "Content-Type: application/json" -d '{"username": "newuser", "email": "newuser@example.com", "password": "securepassword", "confirm_password": "securepassword"}'

If everything goes well, the system will return a success message along with the created user’s data:

{
  "message": "User registered successfully.",
  "user": {
    "email": "newuser@example.com",
    "id": 1,
    "username": "newuser"
  }
}

10.2 Test login route

• Log in with the registered user:

curl -X POST http://localhost:5000/auth/login -H "Content-Type: application/json" -d '{"email": "newuser@example.com", "password": "securepassword"}'

If the credentials are correct, the system will return a success message along with the generated tokens:

{
  "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoxLCJleHAiOjE3NjQ0NDQzMzV9.GZbvYvJ9BsaLfp3CyXwY6dh6CO-tfLlSfZJSeqP7J-s",
  "message": "Login successful.",
  "refresh_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoxLCJleHAiOjE3NjUwNDgyMzV9.jzGeEhcjGcbl_zWHpS0sxMLx0nEGH4fw729et5HDnyY"
}

10.3 Test refresh token route

• Obtain a new access token using the refresh token:

curl -X POST http://localhost:5000/auth/refresh -H "Content-Type: application/json" -d '{"refresh_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoxLCJleHAiOjE3NjUwNDgyMzV9.jzGeEhcjGcbl_zWHpS0sxMLx0nEGH4fw729et5HDnyY"}'

If the refresh token is valid, the system will return a new access token:

{
  "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoxLCJleHAiOjE3NjQ0NDQzODd9.iVG5KwpUNze5lrJurlznYyk02jcnCc8ZbP2F4obys6Q"
}

10.4 Test logout route

• Log out and revoke the refresh token:

curl -X POST http://localhost:5000/auth/logout -H "Content-Type: application/json" -d '{"refresh_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoxLCJleHAiOjE3NjUwNDgyMzV9.jzGeEhcjGcbl_zWHpS0sxMLx0nEGH4fw729et5HDnyY"}'

If the refresh token is valid, the system will return a success message indicating that the token has been revoked:

{
  "message": "Logout successful. Token revoked."
}

Ya con esto tenemos un functional authentication system, using JWT for login and logout, we could now implement it in a front-end, such as a web page or a mobile app.

Now, let’s move on to protecting certain routes of our API so that only authenticated users can access them.

11. Protect routes with JWT

To protect the routes of our API and ensure that only authenticated users can access them, we will use a decorator that will verify the validity of the JWT access token in incoming requests.

For this, we will create a middleware, which will be a file called auth_middleware.py inside the app/middlewares/ directory:

from flask import request, jsonify, current_app
import jwt
from functools import wraps
def token_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = None
        # Check if the token is provided in the Authorization header
        if 'Authorization' in request.headers:
            auth_header = request.headers['Authorization']
            if auth_header.startswith('Bearer '):
                token = auth_header.split(' ')[1]
        if not token:
            return jsonify({'error': 'Token is missing!'}), 401
        try:
            payload = jwt.decode(token, current_app.config['JWT_SECRET_KEY'], algorithms=['HS256'])
            current_user_id = payload['user_id']
        except jwt.ExpiredSignatureError:
            return jsonify({'error': 'Token has expired!'}), 401
        except jwt.InvalidTokenError:
            return jsonify({'error': 'Invalid token!'}), 401
        return f(current_user_id, *args, **kwargs)
    return decorated

Explanation of the code:

• A decorator token_required is defined that checks for the presence and validity of the JWT access token in the authorization header of the request.
• If the token is valid, the user_id is extracted from the payload and passed as an argument to the protected function.
• If the token is missing, expired, or invalid, an error message with status code 401 (Unauthorized) is returned.

11.1 Apply the middleware to protected routes

Now that we have our token_required middleware, we can apply it to the routes we want to protect. For example, we can protect the create, update, and delete task routes so that only authenticated users can access them. For this example, we will modify the post_task.py file to apply the middleware:

from flask import request, jsonify
from app import db
from app.models import Task, User
from . import tasks_bp
from app.middlewares.auth_middleware import token_required

@tasks_bp.route('/', methods=['POST'])
@token_required # Add middleware here
def post_task(current_user_id):
    if not request.is_json:
        return jsonify({'error': 'Request body must be JSON.'}), 400
    data = request.json
    # Validate required fields and collect errors
    required_fields = ['title', 'user_id']
    errors = {}
    for field in required_fields:
        if field not in data or not data[field]:
            errors[field] = f'The field {field} is required.'
    # Validate user_id exists
    if 'user_id' in data and data.get('user_id'):
        if not User.query.get(data['user_id']):
            errors['user_id'] = 'The specified user_id does not exist.'
    if errors:
        return jsonify({'errors': errors}), 422
    new_task = Task(
        title=data['title'],
        description=data.get('description', ''),
        user_id=data['user_id']
    )
    db.session.add(new_task)
    db.session.commit()
    return jsonify({'message': 'Task created successfully.', 'task': new_task.to_dict()}), 201

Explanation of the code:

• The token_required decorator is imported from the middleware.
• The @token_required decorator is applied to the post_task route, which requires the user to be authenticated to access this route. If not authenticated, they will receive a 401 error as programmed in the middleware.
• The post_task function now receives an additional argument current_user_id, which is the ID of the authenticated user extracted from the JWT token. Similarly, you can apply the @token_required decorator to other routes you want to protect, such as update and delete task routes, or even user routes if necessary. Now, when a client tries to access a protected route, they must include the JWT access token in the authorization header of the request. If the token is valid, they will be able to access the route; otherwise, they will receive an authorization error.

11.1.1 Sending the token in requests

To send the token in requests to protected routes, you must include it in the Authorization header as follows:

Authorization: Bearer <access_token>

Replacing <access_token> with the JWT token you received when logging in.

For the post_task example, the curl request would look like this:

curl -X POST http://localhost:5000/tasks/ -H "Content-Type: application/json" -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoxLCJleHAiOjE3NjQ0NDQzMzV9.GZbvYvJ9BsaLfp3CyXwY6dh6CO-tfLlSfZJSeqP7J-s" -d '{"title": "New Task", "user_id": 1}'

You must send the valid token you obtained during login, along with the necessary data to create the task.

If everything goes well, the system will return a success message along with the data of the created task. If the token is invalid or has expired, you will receive a 401 error.

{
  "message": "Task created successfully.",
  "task": {
    "completed": false,
    "date_completed": null,
    "date_created": "Sat, 29 Nov 2025 16:19:08 GMT",
    "description": "",
    "id": 1,
    "title": "New Task",
    "user_id": 1
  }
}

Now this route is secured so that if someone wants to create a task, they must be authenticated beforehand and can only create tasks for themselves.

11.2 Protect sensitive information routes

Besides protecting the routes for creating, updating, and deleting tasks, it is also important to protect routes that return sensitive information, such as user data. For example, we can protect the get_user route so that only the authenticated user can access their own information.

from flask import jsonify
from app import db
from app.models import User
from . import users_bp
from app.middlewares.auth_middleware import token_required

# Only copy this function don't modify get users route
@users_bp.route('/<int:user_id>', methods=['GET'])
@token_required
def get_user(current_user_id, user_id):
    if current_user_id != user_id:
        return jsonify({'error': 'Unauthorized access.'}), 403
    user = User.query.get(user_id)
    if not user:
        return jsonify({'message': 'User not found'}), 404
    return jsonify(user.to_dict())

Explanation of the code:

• The @token_required decorator is applied to the get_user route, which requires the user to be authenticated to access this route.
• Inside the get_user function, it checks that the current_user_id (ID of the authenticated user) matches the requested user_id. If they do not match, a 403 (Forbidden) error is returned indicating that access is not authorized.

If I were the user with ID 1, for example, I could only access my own information by making a request like this:

curl -X GET http://localhost:5000/users/1 -H "Authorization: Bearer <access_token>"

{
    "email": "newuser@example.com",
    "id": 1,
    "username": "newuser"
}

If I try to access another user’s information, I will receive a 403 error:

curl -X GET http://localhost:5000/users/2 -H "Authorization: Bearer <access_token>"

{
    "error": "Unauthorized access."
}

Explanation of the code:

• It ensures that only the authenticated user can access their own information, thus protecting sensitive data of other users.

With this, we have implemented a robust authentication and authorization system using JWT in our Flask API. Now only authenticated users can access protected routes, which significantly improves the security of our application.

Always remember: Never send passwords in responses, it is a bad security practice even if you protect with JWT Tokens, etc.

We are almost done, to finish we will do two more things: first, explain how to send emails from our Flask API, and second, deploy our API using gunicorn, making it ready for production.

12. Sending emails from Flask

To send emails from our Flask API, we can use the Flask-Mail extension, which provides a simple interface for sending emails. We will set up Flask-Mail in our application and create a function to send a welcome email to users upon registration.

12.1 Create an HTML template for the email

To send styled emails, we can use HTML. We will create a simple HTML template for the email we will send. Create a directory called templates/ at the root of your project and inside it create a file called welcome_email.html with the following content:

<!DOCTYPE html>
<html>
<head>
    <title>Welcome to Our Service</title>
</head>
<body>
    <h1>Welcome, {{ username }}!</h1>
    <p>Thank you for registering with our service. We're excited to have you on board.</p>
    <p>Best regards,<br>The Team</p>
</body>
</html>

Explanation of the code:

• This is a simple HTML template that uses Jinja2 to insert the username into the welcome email.
• It does not have advanced CSS styles, but you can add more styles according to your needs. Be aware that this is not as simple as it seems, as many email clients do not support advanced CSS. Research a bit about “CSS in emails” or look for a pre-made template to have a better visual experience.

12.2 Send the email upon user registration

We will modify the register.py file in the authentication module to send a welcome email to the user after successful registration:

from flask import request, jsonify, render_template
from app import db, mail
from app.models import User
from app import config
from . import auth_bp
import re
from werkzeug.security import generate_password_hash
from flask_mail import Message

def is_password_secure(password, min_length=8, max_length=32, allowed_chars=r'^[A-Za-z0-9@#$%^&+=]*$'):
    """
    Check if the password meets security requirements.
    - min_length: minimum length
    - max_length: maximum length
    - allowed_chars: regex of allowed characters
    """
    if not isinstance(password, str):
        return False, 'Password must be a string.'
    if len(password) < min_length:
        return False, f'Password must be at least {min_length} characters.'
    if len(password) > max_length:
        return False, f'Password must be at most {max_length} characters.'
    if not re.match(allowed_chars, password):
        return False, 'Password contains invalid characters.'
    return True, None

@auth_bp.route('/register', methods=['POST'])
def register():
    if not request.is_json:
        return jsonify({'error': 'Request body must be JSON.'}), 400
    data = request.json
    # Validate required fields and collect errors
    required_fields = ['username', 'email', 'password', 'confirm_password']
    errors = {}
    for field in required_fields:
        if field not in data or not data[field]:
            errors[field] = f'The field {field} is required.'
    # Validate password confirmation
    if 'password' in data and 'confirm_password' in data:
        if data['password'] != data['confirm_password']:
            errors['confirm_password'] = 'Passwords do not match.'
    # Validate password security
    if 'password' in data and data.get('password'):
        secure, msg = is_password_secure(
            data['password'],
            min_length=8,
            max_length=32,
            allowed_chars=r'^[A-Za-z0-9@#$%^&+=]*$'
        )
        if not secure:
            errors['password'] = msg
    # Validate email uniqueness
    if 'email' in data and data.get('email'):
        if User.query.filter_by(email=data['email']).first():
            errors['email'] = 'The email is already registered.'
    if errors:
        return jsonify({'errors': errors}), 422
    # Encrypt password and create user
    hashed_password = generate_password_hash(data['password'])
    new_user = User(
        username=data['username'],
        email=data['email'],
        password=hashed_password
    )
    db.session.add(new_user)
    db.session.commit()

    # Send welcome email
    try:
        msg = Message('Welcome to Our Service',
                      sender=config.Config.MAIL_USERNAME,
                      recipients=[data['email']])
        msg.html = render_template('welcome_email.html', username=data['username'])
        mail.send(msg)
    except Exception as e:
        print(f'Error sending email: {e}')

    return jsonify({'message': 'User registered successfully.', 'user': new_user.to_dict()}), 201

Explanation of the code:

• render_template is imported to render the HTML template for the email and Message from flask_mail to create the email message.
• After creating and saving the new user in the database, an email message is created using the welcome_email.html template.
• The email is sent using mail.send(msg).
• Any exceptions that may occur during the sending of the email are handled by printing an error message to the console.

With this, every time a user successfully registers, they will receive a welcome email at the provided email address.

Now that we have covered how to send emails from our Flask API, in the next section we will see how to deploy our API using Gunicorn for production.

13. Deploying the API with Gunicorn

This step is relatively straightforward. Gunicorn is a WSGI server for Python applications that is lightweight and easy to use. It is ideal for deploying Flask applications in production.

13.1 Create wsgi.py

Let’s create a file named wsgi.py at the root of our project with the following content:

from app import create_app

app = create_app()

13.2 Deploying with Gunicorn

Now we should not run python app.py to start the Flask server. Instead, we will use Gunicorn from the command line.

gunicorn -w 4 -b 0.0.0.0:8000 wsgi:app

Explanation of the command:

• -w 4: Specifies the number of worker processes that Gunicorn will use. You can adjust this number based on your server’s capacity.
• -b 0.0.0.0:8000: Specifies the address and port on which Gunicorn will listen for requests. In this case, it listens on all network interfaces on port 8000.
• wsgi:app: Tells Gunicorn to load the Flask application from the wsgi module and use the variable app as the WSGI application.

With this, your Flask API will be deployed and ready to handle requests in a production environment. We have laid the foundation so you can deploy the API on an HTTP server like Nginx, simply by setting up a reverse proxy to the 8000 port where Gunicorn is listening.

I have written a detailed blog on how to deploy Flask applications with Gunicorn and Nginx, you can read it here: Deploy a ‘Reverse Proxy’ with Nginx on a Raspberry Pi - Blog AstronautMarkus (this is a fictitious link, replace it with the real link if you have one).

Conclusion

Congratulations, you have completed the creation of a full RESTful API with Flask, including authentication and authorization with JWT, error handling, email sending, and deployment with Gunicorn. You now have a solid foundation to build more complex web applications using Flask.

During this post we have created and modified the following things:

• Flask project structure using blueprints.
• Data models for users, tasks, and refresh tokens.
• Script to easily initialize and recreate the database.
• Routes for managing users and tasks with validation and error handling.
• Authentication and authorization module using JWT.
• Middleware to protect routes with JWT tokens.
• Sending emails using Flask-Mail.
• API deployment using Gunicorn.

Bonus: Complete source code

I have created a GitHub repository with the complete source code of this project so you can clone it and use it as a reference or starting point for your own projects. You can find the repository here: GitHub - Flask API Example

Feel free to fork or clone it for your own educational or development purposes!

Now you can start building on this foundation, adding more features according to the needs of your application. Good luck and happy coding with Flask!

There no knowledge that is not power.