Build a Flask REST API with JWT Auth

Build a production-structured Flask API with JWT authentication and task CRUD.

Introduction

30 Second Summary

Every app you log into needs a secure backend that knows who you are and what you're allowed to do. The system behind that experience is a REST API with authentication.

In this project, you will build a Task Manager REST API with Flask that handles user registration, login, and per-user task management. You'll structure it the way professional Flask applications are built in production.

What You'll Build

A modular Flask REST API with JWT-based authentication, database models managed by an ORM, and full CRUD endpoints for task management.

By the end of this project, you'll have:

  • A Flask application structured with the Application Factory pattern and Blueprints for modular organization.
  • A JWT authentication system that handles user registration, login, and secure token-based access.
  • A complete CRUD API for tasks where each user can only access their own data.
  • Secret Mission: Add token refresh and role-based access control so an admin can view all tasks across users.

Are there any prerequisites?

This project assumes you have basic Flask experience (routes, templates, simple apps) and solid Python fundamentals. If you've built a simple Flask app before, you're ready to level up to production patterns.

Set Up the Project Structure

Every production Flask app needs a solid foundation. Without proper structure, your code tangles into a single file that becomes impossible to test, extend, or hand off to another developer.

The Application Factory pattern solves this by separating how your app is configured from how it's created. In this step, you'll build the project skeleton, install dependencies, and wire up the factory so Flask can assemble your API on demand.

In this step, get ready to:
  • Create the project directory with a virtual environment and install Flask plus extensions.
  • Build the Application Factory with configuration, extensions, and a modular folder structure.
  • Verify the dev server runs and returns a health check response.
Create the project directory and install dependencies

A virtual environment isolates your project's packages from your system Python. This prevents version conflicts between different projects on your machine.

  • In VS Code, press Ctrl+` to open the integrated terminal.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir task-manager-api
cd task-manager-api
  • Create and activate a virtual environment by running the commands for your operating system:

macOS

python3 -m venv venv
source venv/bin/activate

Windows

python -m venv venv
venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt. This confirms the virtual environment is active.

  • Install Flask and all required extensions by running this command:
pip install Flask==3.1.3 Flask-SQLAlchemy==3.1.1 Flask-Migrate==4.1.0 Flask-JWT-Extended==4.7.4

What are these packages?

  • Flask is the web framework that handles HTTP requests and routing.
  • Flask-SQLAlchemy connects Flask to a database using Python objects instead of raw SQL.
  • Flask-Migrate tracks and applies database schema changes over time.
  • Flask-JWT-Extended adds JSON Web Token authentication so users can log in without server-side sessions.
  • Verify the installation succeeded by running:
flask --version

You should see Flask 3.1.3 listed in the output along with Python and Werkzeug version numbers.

Seeing 'command not found'?

  • Make sure your virtual environment is active. You should see (venv) in your terminal prompt.
  • If it's not active, re-run the activation command for your operating system from the platform tabs above.
  • On macOS, if python3 works but flask does not, your PATH may not include the venv's bin directory. Deactivate and reactivate with deactivate then source venv/bin/activate.

Help me fix my Flask installation issue.

Create the Application Factory and folder structure

The Application Factory pattern uses a function to create your Flask app. Instead of a global app object floating in a module, a create_app() function builds and configures the app each time it's called.

This makes testing easy (you can create fresh app instances with different configs) and prevents circular imports as your project grows.

  • Create the folder structure for your project by running these commands in the terminal:
mkdir app
mkdir app/models
mkdir app/routes
  • Confirm the folders exist by running:
ls app

You should see models and routes listed in the output.

  • Open the task-manager-api folder in VS Code. Click File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop, select the task-manager-api folder, and click Open.

Why open the folder?

Opening the project folder gives VS Code's Explorer sidebar a view of your entire project. You'll use this sidebar to create and navigate files for the rest of the project.

  • If your terminal closed when the folder opened, press Ctrl+` to reopen it. Make sure (venv) still appears in the prompt.

Now you'll create each file. The first file defines your extension instances so they can be shared across the app without circular imports.

  • In the Explorer sidebar, right-click the app folder and select New File. Name it extensions.py.
  • Add the following code to app/extensions.py:
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_jwt_extended import JWTManager

# Create extension instances without binding them to an app yet
db = SQLAlchemy()
migrate = Migrate()
jwt = JWTManager()

What does this code do?

Each extension is created without an app instance. The factory function will bind them later by calling db.init_app(app) and similar methods. This separation is what prevents circular imports in larger Flask projects.

  • Save app/extensions.py.
  • Right-click the app folder again and create a new file called config.py. Add the following code:
import os
from datetime import timedelta

basedir = os.path.abspath(os.path.dirname(__file__))


class Config:
    # Flask's built-in secret key for session signing
    SECRET_KEY = os.environ.get("SECRET_KEY", "dev-secret-key-change-in-production")
    # SQLite database stored in the project root
    SQLALCHEMY_DATABASE_URI = os.environ.get(
        "DATABASE_URL", "sqlite:///" + os.path.join(basedir, "..", "app.db")
    )
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    # Separate secret key for signing JWT tokens
    JWT_SECRET_KEY = os.environ.get("JWT_SECRET_KEY", "jwt-secret-key-change-in-production")
    # Access tokens expire after 1 hour
    JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)

What does this code do?

  • SECRET_KEY is used by Flask for session signing. The fallback value is fine for development but must be changed in production.
  • SQLALCHEMY_DATABASE_URI points to a SQLite file called app.db in the project root. No external database server needed.
  • JWT_ACCESS_TOKEN_EXPIRES sets how long a login token stays valid before the user must re-authenticate.
  • Save app/config.py.
  • Right-click the app folder and create a new file called __init__.py. Add the following code:
from flask import Flask
from app.config import Config
from app.extensions import db, migrate, jwt


def create_app(config_class=Config):
    # Create the Flask application instance
    app = Flask(__name__)
    app.config.from_object(config_class)

    # Bind extensions to this app instance
    db.init_app(app)
    migrate.init_app(app, db)
    jwt.init_app(app)

    # Health check endpoint to verify the app is running
    @app.route("/api/health")
    def health_check():
        return {"status": "healthy"}

    return app

What does this code do?

The create_app() function is the heart of the Application Factory pattern. It creates a fresh Flask instance, loads the configuration, binds all extensions, and registers a simple health check route.

Returning the app object at the end lets other parts of your project (like run.py) call this function to get a fully configured app.

  • Save app/__init__.py.
  • In the Explorer sidebar, right-click the project root (task-manager-api) and create a new file called run.py. Add the following code:
from app import create_app

# Call the factory to build the app
app = create_app()

if __name__ == "__main__":
    app.run(debug=True)

What does this code do?

This is the entry point for your application. It calls create_app() to build the configured Flask app. The flask run command will find this file and use the app variable as the application instance.

  • Save run.py.
  • Create two empty __init__.py files to mark the models and routes folders as Python packages. Right-click the models folder inside app and select New File. Name it __init__.py and leave it empty.
  • Do the same for the routes folder. Right-click routes, select New File, name it __init__.py, and leave it empty.
  • Save both files.

Why empty __init__.py files?

Python requires an __init__.py file in each folder that should be treated as a package. Without them, Python can't import modules from these directories. You'll add code to these files in later steps.

✔️ Awesome, I've got everything!

Great. Double check you've saved all files before moving on.

ⓧ I'd like to double check the full code

Here is the full content of each file you should have at this point:

from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_jwt_extended import JWTManager

db = SQLAlchemy()
migrate = Migrate()
jwt = JWTManager()
import os
from datetime import timedelta

basedir = os.path.abspath(os.path.dirname(__file__))


class Config:
    SECRET_KEY = os.environ.get("SECRET_KEY", "dev-secret-key-change-in-production")
    SQLALCHEMY_DATABASE_URI = os.environ.get(
        "DATABASE_URL", "sqlite:///" + os.path.join(basedir, "..", "app.db")
    )
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    JWT_SECRET_KEY = os.environ.get("JWT_SECRET_KEY", "jwt-secret-key-change-in-production")
    JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
from flask import Flask
from app.config import Config
from app.extensions import db, migrate, jwt


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

    db.init_app(app)
    migrate.init_app(app, db)
    jwt.init_app(app)

    @app.route("/api/health")
    def health_check():
        return {"status": "healthy"}

    return app
from app import create_app

app = create_app()

if __name__ == "__main__":
    app.run(debug=True)

The app/models/__init__.py and app/routes/__init__.py files should both be empty.

Verify the app starts with a health check

The flask run command needs to know which file contains your app. You'll point it at run.py using an environment variable.

  • Set the FLASK_APP environment variable and start the development server by running these commands:

macOS

export FLASK_APP=run.py
flask run

Windows

set FLASK_APP=run.py
flask run

You should see output indicating the server is running on http://127.0.0.1:5000.

  • Open a second terminal in VS Code by clicking the + icon in the terminal panel. Test the health check endpoint by running:
curl http://127.0.0.1:5000/api/health

You should see this response:

{"status":"healthy"}

Seeing an import error?

  • Make sure your (venv) is active in the terminal running flask run. If not, re-activate it.
  • Check that app/__init__.py exists and contains the create_app function. A missing or misnamed file causes Flask to fail on startup.
  • Verify that run.py is in the project root (task-manager-api/run.py), not inside the app folder.

Help me debug my Flask app startup error.

Why use a health check endpoint?

A /api/health endpoint gives you an instant way to confirm your app is alive. In production, load balancers and monitoring tools ping this endpoint to verify the service is responding.

Your project skeleton is solid. Next up, you'll build the User model and create authentication endpoints so people can register and log in to your API.

Build the User Model and Auth Endpoints

Your Flask app is running and returning a healthy response. Now it needs to know about users.

In this step, you will create a User model that stores credentials securely, then build two API endpoints that let someone register and log in. Both endpoints return a JWT access token that proves the caller's identity for future requests.

In this step, get ready to:
  • Create a User model with secure password hashing.
  • Build registration and login endpoints that return access tokens.
  • Initialize and run database migrations to create the users table.
Create the User model with password hashing

The User model defines how user data is stored in the database. Passwords must never be saved as plain text, so you will use hashing to protect them.

  • Create a new file called user.py inside the app/models/ folder and add the following code:
from datetime import datetime, timezone
from werkzeug.security import generate_password_hash, check_password_hash
from app.extensions import db


class User(db.Model):
    __tablename__ = "users"

    # Define the columns for the users table
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    password_hash = db.Column(db.String(256), nullable=False)
    created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
    tasks = db.relationship("Task", backref="owner", lazy=True)

    def set_password(self, password):
        # Hash the password so plain text is never stored
        self.password_hash = generate_password_hash(password)

    def check_password(self, password):
        # Compare a plain text password against the stored hash
        return check_password_hash(self.password_hash, password)

    def to_dict(self):
        # Return a safe dictionary representation (no password hash)
        return {
            "id": self.id,
            "username": self.username,
            "email": self.email,
            "created_at": self.created_at.isoformat(),
        }

What does this code do?

The User class maps to a users table in the database. Each column becomes a field in that table.

The set_password method uses Werkzeug's generate_password_hash to create a one-way hash. check_password compares a plain text attempt against that hash without ever reversing it.

The to_dict method serializes user data for JSON responses, deliberately excluding the password hash.

  • Now open app/models/__init__.py and add this import so that Flask-Migrate can discover the model:
from app.models.user import User
  • Save both app/models/user.py and app/models/__init__.py.

Why does the import matter?

Flask-Migrate only detects models that are imported during app creation. By importing User in the models package, any code that later imports from app.models will register the model with SQLAlchemy's metadata.

Create the auth blueprint with registration and login

A Blueprint groups related routes into a reusable module. The auth blueprint handles user registration and login, issuing a JWT access token on success.

  • Create a new file called auth.py inside the app/routes/ folder. Start with the imports, blueprint declaration, and the registration endpoint:
from flask import Blueprint, request, jsonify
from flask_jwt_extended import create_access_token
from app.extensions import db
from app.models.user import User

auth_bp = Blueprint("auth", __name__)


@auth_bp.route("/register", methods=["POST"])
def register():
    data = request.get_json()

    # Reject requests that aren't JSON
    if not data:
        return jsonify({"error": "Request body must be JSON"}), 400

    username = data.get("username")
    email = data.get("email")
    password = data.get("password")

    # Validate required fields
    if not username or not email or not password:
        return jsonify({"error": "username, email, and password are required"}), 400

    # Check for duplicate username
    if User.query.filter_by(username=username).first():
        return jsonify({"error": "Username already exists"}), 409

    # Check for duplicate email
    if User.query.filter_by(email=email).first():
        return jsonify({"error": "Email already exists"}), 409

    # Create the user and hash their password
    user = User(username=username, email=email)
    user.set_password(password)
    db.session.add(user)
    db.session.commit()

    # Issue a JWT access token with the user's ID as identity
    access_token = create_access_token(identity=str(user.id))
    return jsonify({"user": user.to_dict(), "access_token": access_token}), 201

What does the register endpoint do?

  • Parses the JSON body and checks that username, email, and password are present.
  • Returns a 409 Conflict if the username or email already exists in the database.
  • Hashes the password, saves the user, and returns a JWT access token so the user is immediately authenticated.
  • The identity passed to create_access_token must be a string in Flask-JWT-Extended 4.x.
  • Below the register function in the same file, add the login endpoint:
@auth_bp.route("/login", methods=["POST"])
def login():
    data = request.get_json()

    if not data:
        return jsonify({"error": "Request body must be JSON"}), 400

    username = data.get("username")
    password = data.get("password")

    if not username or not password:
        return jsonify({"error": "username and password are required"}), 400

    # Look up the user by username
    user = User.query.filter_by(username=username).first()

    # Verify the user exists and the password matches
    if not user or not user.check_password(password):
        return jsonify({"error": "Invalid username or password"}), 401

    access_token = create_access_token(identity=str(user.id))
    return jsonify({"user": user.to_dict(), "access_token": access_token}), 200

What does the login endpoint do?

It finds the user by username, verifies the password against the stored hash, and returns a fresh access token. A generic 401 error is returned for both "user not found" and "wrong password" so attackers cannot tell which one failed.

  • Save app/routes/auth.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved app/routes/auth.py with both the register and login functions.

ⓧ I'd like to double check the full code

from flask import Blueprint, request, jsonify
from flask_jwt_extended import create_access_token
from app.extensions import db
from app.models.user import User

auth_bp = Blueprint("auth", __name__)


@auth_bp.route("/register", methods=["POST"])
def register():
    data = request.get_json()

    if not data:
        return jsonify({"error": "Request body must be JSON"}), 400

    username = data.get("username")
    email = data.get("email")
    password = data.get("password")

    if not username or not email or not password:
        return jsonify({"error": "username, email, and password are required"}), 400

    if User.query.filter_by(username=username).first():
        return jsonify({"error": "Username already exists"}), 409

    if User.query.filter_by(email=email).first():
        return jsonify({"error": "Email already exists"}), 409

    user = User(username=username, email=email)
    user.set_password(password)
    db.session.add(user)
    db.session.commit()

    access_token = create_access_token(identity=str(user.id))
    return jsonify({"user": user.to_dict(), "access_token": access_token}), 201


@auth_bp.route("/login", methods=["POST"])
def login():
    data = request.get_json()

    if not data:
        return jsonify({"error": "Request body must be JSON"}), 400

    username = data.get("username")
    password = data.get("password")

    if not username or not password:
        return jsonify({"error": "username and password are required"}), 400

    user = User.query.filter_by(username=username).first()

    if not user or not user.check_password(password):
        return jsonify({"error": "Invalid username or password"}), 401

    access_token = create_access_token(identity=str(user.id))
    return jsonify({"user": user.to_dict(), "access_token": access_token}), 200

Now you need to register this blueprint in the application factory so Flask knows about the auth routes.

  • Open app/__init__.py and find this line:
    jwt.init_app(app)
  • Directly below it, add the following lines to import and register the auth blueprint:
    from app.routes.auth import auth_bp

    app.register_blueprint(auth_bp, url_prefix="/api/auth")
  • Save app/__init__.py.

Why import inside the function?

Importing auth_bp inside create_app (rather than at the top of the file) avoids circular imports. The blueprint module imports from app.extensions and app.models, which need the app package to exist first.

✔️ Awesome, I've got everything!

Great. Make sure you have saved app/__init__.py with the blueprint registration.

ⓧ I'd like to double check the full code

from flask import Flask
from app.config import Config
from app.extensions import db, migrate, jwt


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

    db.init_app(app)
    migrate.init_app(app, db)
    jwt.init_app(app)

    from app.routes.auth import auth_bp

    app.register_blueprint(auth_bp, url_prefix="/api/auth")

    @app.route("/api/health")
    def health_check():
        return {"status": "healthy"}

    return app

With the blueprint registered, the import chain now reaches the User model. Flask-Migrate will detect it when you run migrations.

  • Initialize the migrations folder by running this command in your terminal:
flask db init
  • You should see output confirming a migrations/ directory was created inside your project folder.
  • Generate the first migration script by running:
flask db migrate
  • You should see a line that says Detected added table 'users'. This confirms Flask-Migrate found your User model.
  • Apply the migration to create the actual database table by running:
flask db upgrade
  • After this command, an app.db file appears in your project root. This is your SQLite database with a users table ready to store data.

Seeing 'No changes in schema detected'?

This means Flask-Migrate could not find any models. Make sure app/models/__init__.py contains from app.models.user import User and that the blueprint is registered in app/__init__.py. The import chain must reach the User model during app creation.

If you already ran flask db migrate before registering the blueprint, delete the migrations/ folder and run flask db init again.

Help me fix this

Test the endpoints with curl

The best way to verify an API works is to call it. You will use curl to send HTTP requests from a second terminal window.

  • Make sure your Flask dev server is running. If it stopped, start it again by running:
flask run
  • Open a second terminal window (keep the server running in the first one).
  • Register a new user by running this command:
curl -X POST http://127.0.0.1:5000/api/auth/register -H "Content-Type: application/json" -d '{"username": "testuser", "email": "test@example.com", "password": "secret123"}'
  • You should see a JSON response with a 201 status containing the user object and an access_token string.

What should I see?

The response looks something like this (your token will be different):

{
  "access_token": "eyJhbGciOiJIUzI1NiIs...",
  "user": {
    "id": 1,
    "username": "testuser",
    "email": "test@example.com",
    "created_at": "2026-07-03T..."
  }
}
  • Now test the login endpoint by running:
curl -X POST http://127.0.0.1:5000/api/auth/login -H "Content-Type: application/json" -d '{"username": "testuser", "password": "secret123"}'
  • You should see a 200 response with the same user object and a fresh access_token.

Getting a 404 or connection refused?

Make sure your Flask dev server is still running in the first terminal window. Check that the URL uses port 5000 and the path starts with /api/auth/.

If you see an import error in the server terminal, double-check that app/routes/auth.py exists and that the import statement in app/__init__.py matches exactly: from app.routes.auth import auth_bp.

Help me troubleshoot my Flask auth endpoints

Your API now has a working authentication system. Next up, you will build the task endpoints that use these JWT tokens to protect user data.

Build the Task CRUD Endpoints

Your authentication system is in place. Users can register and log in, and you get a JWT token back on success.

Now it is time to build the actual feature your API exists to serve: task management. Each authenticated user should be able to create, read, update, and delete their own tasks. Nobody else should be able to touch them.

In this step, get ready to:
  • Create a Task model linked to each user via a foreign key.
  • Build four protected CRUD endpoints in a tasks Blueprint.
  • Test the full flow with curl: register, login, create, list, update, and delete tasks.
Create the Task model with user ownership

Every task needs to belong to a specific user. A foreign key column on the Task table points back to the User who created it. This is how the API enforces per-user ownership.

  • Create a new file called task.py inside the app/models/ directory.
  • Add the following code to define the Task model:
from datetime import datetime, timezone
from app.extensions import db


class Task(db.Model):
    __tablename__ = "tasks"

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    description = db.Column(db.Text, nullable=True)
    completed = db.Column(db.Boolean, default=False)
    created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
    user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)

    def to_dict(self):
        return {
            "id": self.id,
            "title": self.title,
            "description": self.description,
            "completed": self.completed,
            "created_at": self.created_at.isoformat(),
            "user_id": self.user_id,
        }

What does this model do?

The Task model mirrors the structure of the User model you built earlier, with a few key differences.

  • user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False) creates the foreign key relationship. Every task must belong to a user.
  • completed defaults to False so new tasks start as incomplete.
  • to_dict() converts the model to JSON-friendly output, just like the User model.
  • Save app/models/task.py.
  • Now open app/models/__init__.py and update it to import the new Task model:
from app.models.user import User
from app.models.task import Task

Why import Task here?

Flask-Migrate scans imported models to detect database changes. If Task is not imported in this __init__.py, the migration tool will not see the new table and will generate an empty migration.

  • Save app/models/__init__.py.
  • Switch back to your terminal and stop the running Flask server by pressing Ctrl+C.
  • Generate the migration for the new tasks table by running:
flask db migrate -m "add tasks table"
  • Apply the migration to your database by running:
flask db upgrade
  • You should see output confirming the migration ran, ending with a line like Running upgrade ... -> ....

Migration not detecting the new table?

  • Make sure app/models/__init__.py imports Task. Without that import, Flask-Migrate cannot see the model.
  • Check that app/models/task.py is saved with no syntax errors. A missing colon or incorrect indentation will prevent the import.
  • help me debug my Flask migration
Create the tasks blueprint with protected CRUD endpoints

With the Task table in your database, you can now build the API endpoints that let users interact with their tasks. Every endpoint in this blueprint requires a valid JWT token and scopes queries to the authenticated user.

  • Create a new file called tasks.py inside the app/routes/ directory.
  • Add the following code for the GET and POST endpoints:
from flask import Blueprint, request, jsonify
from flask_jwt_extended import jwt_required, get_jwt_identity
from app.extensions import db
from app.models.task import Task

tasks_bp = Blueprint("tasks", __name__)


@tasks_bp.route("", methods=["GET"])
@jwt_required()
def get_tasks():
    user_id = int(get_jwt_identity())
    tasks = Task.query.filter_by(user_id=user_id).all()
    return jsonify({"tasks": [task.to_dict() for task in tasks]}), 200


@tasks_bp.route("", methods=["POST"])
@jwt_required()
def create_task():
    user_id = int(get_jwt_identity())
    data = request.get_json()

    if not data:
        return jsonify({"error": "Request body must be JSON"}), 400

    title = data.get("title")
    if not title:
        return jsonify({"error": "title is required"}), 400

    task = Task(
        title=title,
        description=data.get("description", ""),
        user_id=user_id,
    )
    db.session.add(task)
    db.session.commit()

    return jsonify({"task": task.to_dict()}), 201

What do these endpoints do?

  • @jwt_required() blocks any request without a valid token. The user must be logged in.
  • get_jwt_identity() extracts the user ID from the token so each endpoint knows who is making the request.
  • GET filters tasks by user_id so users only see their own data.
  • POST validates that a title exists, then creates the task with the authenticated user's ID attached.
  • Now add the PUT and DELETE endpoints below the existing code in the same file:

@tasks_bp.route("/<int:task_id>", methods=["PUT"])
@jwt_required()
def update_task(task_id):
    user_id = int(get_jwt_identity())
    task = Task.query.get(task_id)

    if not task:
        return jsonify({"error": "Task not found"}), 404

    if task.user_id != user_id:
        return jsonify({"error": "Forbidden"}), 403

    data = request.get_json()
    if not data:
        return jsonify({"error": "Request body must be JSON"}), 400

    if "title" in data:
        task.title = data["title"]
    if "description" in data:
        task.description = data["description"]
    if "completed" in data:
        task.completed = data["completed"]

    db.session.commit()
    return jsonify({"task": task.to_dict()}), 200


@tasks_bp.route("/<int:task_id>", methods=["DELETE"])
@jwt_required()
def delete_task(task_id):
    user_id = int(get_jwt_identity())
    task = Task.query.get(task_id)

    if not task:
        return jsonify({"error": "Task not found"}), 404

    if task.user_id != user_id:
        return jsonify({"error": "Forbidden"}), 403

    db.session.delete(task)
    db.session.commit()
    return jsonify({"message": "Task deleted"}), 200

How does ownership protection work?

Both PUT and DELETE first look up the task, then compare task.user_id against the authenticated user's ID. If they do not match, the endpoint returns a 403 Forbidden response. This prevents users from modifying or deleting tasks that belong to someone else.

  • Save app/routes/tasks.py.
  • Now open app/__init__.py to register the new blueprint.
  • Find this line:
    from app.routes.auth import auth_bp

    app.register_blueprint(auth_bp, url_prefix="/api/auth")
  • Replace that section with the following, which adds the tasks blueprint import and registration:
    from app.routes.auth import auth_bp
    from app.routes.tasks import tasks_bp

    app.register_blueprint(auth_bp, url_prefix="/api/auth")
    app.register_blueprint(tasks_bp, url_prefix="/api/tasks")
  • Save app/__init__.py.
  • Restart the Flask development server by running:
flask run
  • You should see the server start on http://127.0.0.1:5000/ with no errors.

Seeing an ImportError on startup?

  • Check that app/routes/tasks.py defines tasks_bp exactly as tasks_bp = Blueprint("tasks", __name__). A typo in the variable name causes the import to fail.
  • Verify that both from app.routes.tasks import tasks_bp and app.register_blueprint(tasks_bp, url_prefix="/api/tasks") are inside the create_app function, not at the top of the file.
  • help me fix my Flask import error
Test the full CRUD flow with curl

With the server running, you can now test the entire flow: register a user, log in, and perform all four task operations. You will need a second terminal window for these commands.

  • Open a new terminal window or tab.
  • Register a new user and save the access token by running:
curl -X POST http://127.0.0.1:5000/api/auth/register \
  -H "Content-Type: application/json" \
  -d '{"username": "taskuser", "email": "task@example.com", "password": "secret123"}'
  • You should see a JSON response with the user object and an access_token field. Copy the token value (the long string after "access_token":).
  • Create a task using your token by running this command (replace YOUR_TOKEN with the token you copied):
curl -X POST http://127.0.0.1:5000/api/tasks \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -d '{"title": "Buy groceries", "description": "Milk, eggs, bread"}'
  • You should see a response with the new task object including an id of 1.
  • List your tasks by running:
curl -X GET http://127.0.0.1:5000/api/tasks \
  -H "Authorization: Bearer YOUR_TOKEN"
  • You should see a tasks array containing the task you just created.
  • Update the task to mark it as completed by running:
curl -X PUT http://127.0.0.1:5000/api/tasks/1 \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -d '{"completed": true}'
  • The response should show the task with "completed": true.
  • Delete the task by running:
curl -X DELETE http://127.0.0.1:5000/api/tasks/1 \
  -H "Authorization: Bearer YOUR_TOKEN"
  • You should see {"message": "Task deleted"} confirming the task is gone.

What is the Authorization header doing?

The Authorization: Bearer YOUR_TOKEN header sends your JWT with each request. Flask-JWT-Extended reads this header, verifies the token signature, and extracts your user ID. Without it, the endpoint returns a 401 Unauthorized error.

Getting a 401 Unauthorized response?

  • Make sure you replaced YOUR_TOKEN with the actual token string from the register or login response. Do not include the surrounding quotes.
  • Tokens expire after 1 hour. If you waited too long, log in again to get a fresh token.
  • Check that you typed Bearer with a capital B and a space before the token.
  • help me fix my JWT authentication error

✔️ Awesome, I've got everything!

Great. Double check you have saved all your files before moving on.

ⓧ I'd like to double check the full code

Here is the complete code for each file at this point in the project.

from datetime import datetime, timezone
from app.extensions import db


class Task(db.Model):
    __tablename__ = "tasks"

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    description = db.Column(db.Text, nullable=True)
    completed = db.Column(db.Boolean, default=False)
    created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
    user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)

    def to_dict(self):
        return {
            "id": self.id,
            "title": self.title,
            "description": self.description,
            "completed": self.completed,
            "created_at": self.created_at.isoformat(),
            "user_id": self.user_id,
        }
from app.models.user import User
from app.models.task import Task
from flask import Blueprint, request, jsonify
from flask_jwt_extended import jwt_required, get_jwt_identity
from app.extensions import db
from app.models.task import Task

tasks_bp = Blueprint("tasks", __name__)


@tasks_bp.route("", methods=["GET"])
@jwt_required()
def get_tasks():
    user_id = int(get_jwt_identity())
    tasks = Task.query.filter_by(user_id=user_id).all()
    return jsonify({"tasks": [task.to_dict() for task in tasks]}), 200


@tasks_bp.route("", methods=["POST"])
@jwt_required()
def create_task():
    user_id = int(get_jwt_identity())
    data = request.get_json()

    if not data:
        return jsonify({"error": "Request body must be JSON"}), 400

    title = data.get("title")
    if not title:
        return jsonify({"error": "title is required"}), 400

    task = Task(
        title=title,
        description=data.get("description", ""),
        user_id=user_id,
    )
    db.session.add(task)
    db.session.commit()

    return jsonify({"task": task.to_dict()}), 201


@tasks_bp.route("/<int:task_id>", methods=["PUT"])
@jwt_required()
def update_task(task_id):
    user_id = int(get_jwt_identity())
    task = Task.query.get(task_id)

    if not task:
        return jsonify({"error": "Task not found"}), 404

    if task.user_id != user_id:
        return jsonify({"error": "Forbidden"}), 403

    data = request.get_json()
    if not data:
        return jsonify({"error": "Request body must be JSON"}), 400

    if "title" in data:
        task.title = data["title"]
    if "description" in data:
        task.description = data["description"]
    if "completed" in data:
        task.completed = data["completed"]

    db.session.commit()
    return jsonify({"task": task.to_dict()}), 200


@tasks_bp.route("/<int:task_id>", methods=["DELETE"])
@jwt_required()
def delete_task(task_id):
    user_id = int(get_jwt_identity())
    task = Task.query.get(task_id)

    if not task:
        return jsonify({"error": "Task not found"}), 404

    if task.user_id != user_id:
        return jsonify({"error": "Forbidden"}), 403

    db.session.delete(task)
    db.session.commit()
    return jsonify({"message": "Task deleted"}), 200
from flask import Flask
from app.config import Config
from app.extensions import db, migrate, jwt


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

    db.init_app(app)
    migrate.init_app(app, db)
    jwt.init_app(app)

    from app.routes.auth import auth_bp
    from app.routes.tasks import tasks_bp

    app.register_blueprint(auth_bp, url_prefix="/api/auth")
    app.register_blueprint(tasks_bp, url_prefix="/api/tasks")

    @app.route("/api/health")
    def health_check():
        return {"status": "healthy"}

    return app

Your API now supports full task management with per-user ownership. Next up, you will add proper error handling and input validation to make the API production-ready.

Add Error Handling and Input Validation

Your API endpoints already validate inputs and return helpful error messages. But what happens when Flask itself encounters an unhandled error, like a request to a URL that does not exist?

By default, Flask returns an HTML error page. That is useless for API clients expecting JSON. In this step, you will add global error handlers that guarantee every error returns a consistent JSON response, then test the full error-handling picture.

In this step, get ready to:
  • Create global error handlers that return JSON for any unhandled HTTP error.
  • Register the error handlers in the application factory.
  • Test error cases across registration, login, and task endpoints.
Add global error handlers

Without global handlers, Flask returns HTML for errors like 404 (page not found) or 500 (internal server error). API consumers need structured JSON they can parse programmatically.

  • Create a new file called app/errors.py inside the app folder by adding the following code:
from flask import jsonify


def register_error_handlers(app):
    # Handle 400 Bad Request
    @app.errorhandler(400)
    def bad_request(e):
        return jsonify({"error": "Bad request"}), 400

    # Handle 404 Not Found
    @app.errorhandler(404)
    def not_found(e):
        return jsonify({"error": "Resource not found"}), 404

    # Handle 422 Unprocessable Entity
    @app.errorhandler(422)
    def unprocessable(e):
        return jsonify({"error": "Unprocessable entity"}), 422

    # Handle 500 Internal Server Error
    @app.errorhandler(500)
    def internal_error(e):
        return jsonify({"error": "Internal server error"}), 500

What does this code do?

The register_error_handlers function accepts the Flask app instance and registers four error handlers using the @app.errorhandler decorator.

  • Each handler catches a specific HTTP status code (400, 404, 422, 500) and returns a JSON object with a consistent "error" key.
  • This replaces Flask's default HTML error pages, so any client consuming the API always gets machine-readable JSON.
  • Save app/errors.py.

Why define handlers in a separate module?

Keeping error handlers in their own file follows the same modular pattern as your routes and models. It prevents the factory function from growing too large and makes the handlers easy to locate and update.

Register error handlers in the application factory

The handlers exist but Flask does not know about them yet. You need to call register_error_handlers inside create_app so they activate when the app starts.

  • In app/__init__.py, find the two register_blueprint lines. Locate this section:
    app.register_blueprint(auth_bp, url_prefix="/api/auth")
    app.register_blueprint(tasks_bp, url_prefix="/api/tasks")

    @app.route("/api/health")
  • Add the error handler registration between the blueprint lines and the health check route, so it looks like this:
    app.register_blueprint(auth_bp, url_prefix="/api/auth")
    app.register_blueprint(tasks_bp, url_prefix="/api/tasks")

    from app.errors import register_error_handlers
    register_error_handlers(app)

    @app.route("/api/health")
  • Save app/__init__.py.

Seeing an ImportError?

Make sure app/errors.py is saved in the app folder (not in the project root). The import path from app.errors import register_error_handlers expects the file inside the app package.

Help me fix an ImportError in my Flask application factory.

✔️ Awesome, I've got everything!

Great. Double check you have saved both files before continuing.

ⓧ I'd like to double check the full code

from flask import Flask
from app.config import Config
from app.extensions import db, migrate, jwt


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

    db.init_app(app)
    migrate.init_app(app, db)
    jwt.init_app(app)

    from app.routes.auth import auth_bp
    from app.routes.tasks import tasks_bp

    app.register_blueprint(auth_bp, url_prefix="/api/auth")
    app.register_blueprint(tasks_bp, url_prefix="/api/tasks")

    from app.errors import register_error_handlers
    register_error_handlers(app)

    @app.route("/api/health")
    def health_check():
        return {"status": "healthy"}

    return app
Test error handling and input validation

Your endpoints already include inline validation (checking for missing fields, duplicate users, and task ownership). The global handlers catch anything that falls through. Time to verify the full picture.

  • Start the development server by running this command in your terminal:
flask run

First, test that missing required fields return a 400 error.

  • Open a second terminal window (keep the server running in the first) and test a registration request with missing fields by running:
curl -X POST http://127.0.0.1:5000/api/auth/register -H "Content-Type: application/json" -d '{"username": "incomplete"}'
  • You should see a JSON response with status 400:
{"error": "username, email, and password are required"}

Next, test that duplicate registration returns a 409 conflict.

  • Register a user first by running:
curl -X POST http://127.0.0.1:5000/api/auth/register -H "Content-Type: application/json" -d '{"username": "alice", "email": "alice@example.com", "password": "secret123"}'
  • Try registering the same username again by running:
curl -X POST http://127.0.0.1:5000/api/auth/register -H "Content-Type: application/json" -d '{"username": "alice", "email": "alice2@example.com", "password": "secret123"}'
  • You should see a 409 response:
{"error": "Username already exists"}

Now test that accessing a nonexistent task returns a 404 error.

  • Copy the access_token value from the successful registration response above. Then test deleting a task that does not exist by running:
curl -X DELETE http://127.0.0.1:5000/api/tasks/999 -H "Authorization: Bearer your-access-token-here"
  • You should see a 404 response:
{"error": "Task not found"}

Finally, test that one user cannot access another user's task (403 Forbidden).

  • Create a task as alice by running:
curl -X POST http://127.0.0.1:5000/api/tasks -H "Content-Type: application/json" -H "Authorization: Bearer your-access-token-here" -d '{"title": "Alice task"}'
  • Note the task id from the response (it will likely be 1).
  • Register a second user by running:
curl -X POST http://127.0.0.1:5000/api/auth/register -H "Content-Type: application/json" -d '{"username": "bob", "email": "bob@example.com", "password": "secret456"}'
  • Copy bob's access_token from the response. Then try to delete alice's task using bob's token:
curl -X DELETE http://127.0.0.1:5000/api/tasks/1 -H "Authorization: Bearer bobs-access-token-here"
  • You should see a 403 response:
{"error": "Forbidden"}

What did you just verify?

  • 400 errors are returned for missing required fields (inline validation in your route handlers).
  • 409 errors catch duplicate registration attempts.
  • 404 errors fire when a task does not exist.
  • 403 errors enforce per-user ownership so users cannot modify each other's data.
  • The global handlers in app/errors.py catch any unhandled errors (like hitting a URL that has no route) and return clean JSON instead of HTML.

Getting unexpected HTML instead of JSON?

Make sure you saved both app/errors.py and app/__init__.py. If the server was already running, it should auto-reload. If it did not, stop the server with Ctrl+C and restart it with flask run.

Help me debug why Flask is returning HTML instead of JSON for errors.

Secret mission

Add Token Refresh and Role-Based Access Control

Your API authenticates users, but what happens when their access token expires? And what if you need certain users to have elevated permissions? In this secret mission, you'll add a token refresh mechanism and role-based access control so admins can view all tasks across the entire system.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally with SQLite, so there are no ongoing costs.

Resources you used:

  • Project folder on your Desktop (all application code, migrations, and configuration).
  • Python virtual environment (venv/ folder inside your project).
  • SQLite database file (app.db at the project root).
  • Python packages installed inside the virtual environment (Flask, Flask-SQLAlchemy, Flask-Migrate, Flask-JWT-Extended).

Keep everything running

No action needed. Choose this if you're still actively building or plan to extend the API further.

  • Your project folder stays on your Desktop with all code, database, and migrations intact.
  • You can restart the Flask dev server at any time by activating your virtual environment and running flask run.
  • Your SQLite database retains all registered users and tasks.

Pause - I'll come back to this later

Shut down running processes to free up system resources. Your files stay on disk for when you return.

  • Stop the Flask development server by pressing Ctrl+C in the terminal where it is running.
  • Deactivate your virtual environment by running:
  • When you return, reactivate the virtual environment and run flask run to pick up where you left off.
deactivate

Everything else (code, database, migrations) stays in place.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes your code, database, and virtual environment permanently.

  • Stop the Flask development server by pressing Ctrl+C in the terminal if it is still running.
  • Deactivate your virtual environment by running:
deactivate
  • Close the project in VS Code by clicking File, then Close Folder.
  • Delete the entire project folder from your Desktop. On macOS/Linux, run:
cd ~/Desktop
rm -rf flask-task-api

On Windows, run:

cd ~\Desktop
rmdir /s /q flask-task-api

This removes all application code, the virtual environment, the SQLite database, and the migrations folder.

Nice Work!

Nice Work!

Nice work! You just built a production-structured REST API with Flask that handles user authentication, data ownership, and clean error responses.

You've learned how to:

  • Build a Flask REST API using the Application Factory pattern with Blueprints for modular organization.
  • Implement secure JWT authentication with user registration, login, and token-based access control.
  • Create a full CRUD API for tasks with per-user ownership so users can only access their own data.
  • Secret Mission: Added a token refresh flow and role-based access control for admin functionality.

Ready to quiz yourself?