Lab 11 — Task Management REST API

Hands-on Lab 90 min Advanced

Objectives

1. Create a Flask application factory with Blueprints
2. Define User and Task SQLAlchemy models
3. Set up JWT authentication (register/login/protected routes)
4. Implement full CRUD for tasks with ownership
5. Test all endpoints with HTTPie or Postman

---

Project Structure

task-api/
├── app.py
├── config.py
├── models.py
├── routes/
│   ├── __init__.py
│   ├── auth.py
│   └── tasks.py
├── requirements.txt
└── .env

---

Step 1 — Setup

mkdir task-api && cd task-api
python3 -m venv venv && source venv/bin/activate
pip install flask flask-sqlalchemy flask-migrate flask-jwt-extended python-dotenv
pip freeze > requirements.txt

.env

SECRET_KEY=change-me-in-production
JWT_SECRET_KEY=jwt-secret-key
DATABASE_URL=sqlite:///tasks.db

---

Step 2 — Configuration

config.py

import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    SECRET_KEY                     = os.getenv("SECRET_KEY", "dev")
    JWT_SECRET_KEY                 = os.getenv("JWT_SECRET_KEY", "jwt-dev")
    SQLALCHEMY_DATABASE_URI        = os.getenv("DATABASE_URL", "sqlite:///tasks.db")
    SQLALCHEMY_TRACK_MODIFICATIONS = False

---

Step 3 — Models

models.py

from datetime import datetime, timezone
from app import db

class User(db.Model):
    __tablename__ = "users"
    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", back_populates="owner", cascade="all, delete-orphan")

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


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, default="")
    completed   = db.Column(db.Boolean, default=False, nullable=False)
    priority    = db.Column(db.Integer, default=2)    # 1=high, 2=medium, 3=low
    due_date    = db.Column(db.Date, nullable=True)
    created_at  = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
    owner_id    = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)
    owner       = db.relationship("User", back_populates="tasks")

    def to_dict(self):
        return {
            "id":          self.id,
            "title":       self.title,
            "description": self.description,
            "completed":   self.completed,
            "priority":    self.priority,
            "due_date":    self.due_date.isoformat() if self.due_date else None,
            "created_at":  self.created_at.isoformat(),
            "owner":       self.owner.username,
        }

---

Step 4 — Authentication Routes

routes/auth.py

import hashlib
from flask import Blueprint, jsonify, request
from flask_jwt_extended import create_access_token
from app import db
from models import User

bp = Blueprint("auth", __name__, url_prefix="/api/auth")

def hash_password(pw: str) -> str:
    return hashlib.sha256(pw.encode()).hexdigest()   # Use bcrypt in production!

@bp.route("/register", methods=["POST"])
def register():
    data = request.get_json(silent=True) or {}

    if not data.get("username") or not data.get("email") or not data.get("password"):
        return jsonify({"error": "username, email, password required"}), 422

    if User.query.filter_by(username=data["username"]).first():
        return jsonify({"error": "Username taken"}), 409

    user = User(
        username      = data["username"],
        email         = data["email"],
        password_hash = hash_password(data["password"]),
    )
    db.session.add(user)
    db.session.commit()

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


@bp.route("/login", methods=["POST"])
def login():
    data = request.get_json(silent=True) or {}
    user = User.query.filter_by(username=data.get("username", "")).first()

    if not user or user.password_hash != hash_password(data.get("password", "")):
        return jsonify({"error": "Invalid credentials"}), 401

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

---

Step 5 — Task Routes

routes/tasks.py

from flask import Blueprint, jsonify, request
from flask_jwt_extended import jwt_required, get_jwt_identity
from app import db
from models import Task, User
from datetime import date

bp = Blueprint("tasks", __name__, url_prefix="/api/tasks")

def current_user() -> User:
    uid = int(get_jwt_identity())
    return db.session.get(User, uid)

@bp.route("", methods=["GET"])
@jwt_required()
def list_tasks():
    user       = current_user()
    show_done  = request.args.get("completed", "").lower() == "true"
    query      = Task.query.filter_by(owner_id=user.id)
    if request.args.get("completed"):
        query = query.filter_by(completed=show_done)
    tasks = query.order_by(Task.priority.asc(), Task.created_at.desc()).all()
    return jsonify([t.to_dict() for t in tasks])

@bp.route("", methods=["POST"])
@jwt_required()
def create_task():
    user = current_user()
    data = request.get_json(silent=True) or {}

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

    task = Task(
        title       = data["title"],
        description = data.get("description", ""),
        priority    = data.get("priority", 2),
        owner_id    = user.id,
    )
    if data.get("due_date"):
        task.due_date = date.fromisoformat(data["due_date"])

    db.session.add(task)
    db.session.commit()
    return jsonify(task.to_dict()), 201

@bp.route("/<int:task_id>", methods=["PATCH"])
@jwt_required()
def update_task(task_id):
    user = current_user()
    task = Task.query.filter_by(id=task_id, owner_id=user.id).first_or_404()
    data = request.get_json(silent=True) or {}

    for field in ("title", "description", "priority"):
        if field in data:
            setattr(task, field, data[field])
    if "completed" in data:
        task.completed = bool(data["completed"])
    if "due_date" in data:
        task.due_date = date.fromisoformat(data["due_date"]) if data["due_date"] else None

    db.session.commit()
    return jsonify(task.to_dict())

@bp.route("/<int:task_id>", methods=["DELETE"])
@jwt_required()
def delete_task(task_id):
    user = current_user()
    task = Task.query.filter_by(id=task_id, owner_id=user.id).first_or_404()
    db.session.delete(task)
    db.session.commit()
    return "", 204

---

Step 6 — App Factory

app.py

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_jwt_extended import JWTManager
from config import Config

db      = SQLAlchemy()
migrate = Migrate()
jwt     = JWTManager()

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

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

    from routes.auth  import bp as auth_bp
    from routes.tasks import bp as tasks_bp
    app.register_blueprint(auth_bp)
    app.register_blueprint(tasks_bp)

    return app

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

---

Step 7 — Run and Test

# Initialize and apply migrations
flask --app app:create_app db init
flask --app app:create_app db migrate -m "initial schema"
flask --app app:create_app db upgrade

# Start the server
python3 app.py

# Register
http POST localhost:5000/api/auth/register username=alice email=alice@example.com password=secret123

# Login
http POST localhost:5000/api/auth/login username=alice password=secret123
# → copy the access_token

# Create tasks
TOKEN="eyJ..."
http POST localhost:5000/api/tasks title="Write unit tests" priority:=1 Authorization:"Bearer $TOKEN"
http POST localhost:5000/api/tasks title="Review PRs" priority:=2 due_date="2025-04-01" Authorization:"Bearer $TOKEN"

# List tasks
http GET localhost:5000/api/tasks Authorization:"Bearer $TOKEN"

# Mark complete
http PATCH localhost:5000/api/tasks/1 completed:=true Authorization:"Bearer $TOKEN"

# Filter completed
http GET "localhost:5000/api/tasks?completed=true" Authorization:"Bearer $TOKEN"

# Delete
http DELETE localhost:5000/api/tasks/2 Authorization:"Bearer $TOKEN"

---

Expected Output

[
  {
    "id": 1,
    "title": "Write unit tests",
    "description": "",
    "completed": true,
    "priority": 1,
    "due_date": null,
    "created_at": "2025-03-29T10:00:00",
    "owner": "alice"
  }
]

---

Summary

• Application factory + Blueprints organize auth and task routes cleanly
• flask db migrate && upgrade safely manages schema changes
• JWT tokens protect task routes — @jwt_required() + get_jwt_identity()
• Users can only read/modify their own tasks (ownership filter on every query)
• first_or_404() automatically returns 404 when a record is not found