Build a Secure Team Task API

Introduction

30 Second Summary

A shared task list becomes risky when anyone can change another person's work. It also becomes frustrating when every task vanishes after a restart.

In this project, you will build a secure team task API with FastAPI. Its layered files make each request easy to trace from sign-in to database storage.

What You'll Build

Your Swagger UI demo rejects Bob's attempt to change Alice's persistent task with a visible 403 Forbidden response.

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

• A working user flow where teammates register through Swagger UI. Each teammate receives a JWT bearer token after sign-in. The token unlocks an owner-scoped task list.
• A persistent task list backed by local SQLite at no cost. Your saved tasks remain visible after a server restart.
• A traceable request flow that returns 401 for missing identity. Forbidden cross-user edits return 403. Each result stays easy to locate because every module owns one responsibility.
• Secret Mission: Add an administrator policy that can inspect and manage tasks across users.

Are there any prerequisites?

You'll need Windows with Python 3.10 or newer. Basic file creation in Visual Studio Code is the only editor skill you need.

Before We Start

This is your moment to commit to the goal before any hands-on work begins. You are building a local Secure Team Task API where users can register, sign in, and manage persistent tasks through separated application layers.

Set Up a Clean Python Environment

Your backend depends on exact package versions. A shared Python installation could let another project change how this API runs. A virtual environment gives this project its own package boundary.

In this step, you'll open secure-task-api in Visual Studio Code. You'll install the pinned FastAPI stack inside tutorial-env.

In this step, get ready to:

• Prepare the project folder with its pinned requirements file.
• Isolate the project dependencies inside a virtual environment.
• Confirm the pinned backend packages are installed.

Create the project folder and requirements file

The requirements.txt file records every package version needed by the backend. This makes the environment repeatable whenever you reinstall it.

• Press the Windows key to open Windows search.
• Type Visual Studio Code into the search field.
• Press Enter to open Visual Studio Code.

You'll see Visual Studio Code open to its start screen.

• Click File in the top menu.
• Click Open Folder.
• Select Desktop in the folder picker.
• Click the new-folder control in the folder picker toolbar.
• Enter secure-task-api as the folder name.
• Press Enter to create the folder.

You should see the new secure-task-api folder selected on your Desktop.

• Select the secure-task-api folder.
• Click Select Folder to open it in Visual Studio Code.

You should see secure-task-api at the top of the Explorer sidebar.

• Click the New File control in the Explorer sidebar.
• Enter requirements.txt as the file name.
• Press Enter to create the file.

You should see requirements.txt open in the editor.

• Add the pinned backend packages by pasting this content into requirements.txt:

fastapi==0.142.2
uvicorn==0.54.0
SQLAlchemy==2.0.54
pydantic==2.13.5
PyJWT==2.15.1
pwdlib[argon2]==0.3.1
python-multipart==0.0.32

Why pin every package?

Each pin freezes one dependency at a known version. Reinstalling from this file gives you the same backend stack.

• FastAPI provides the web framework for the API.
• Uvicorn runs the application as a local web server.
• SQLAlchemy handles database persistence.
• Pydantic validates request data and shapes responses.
• PyJWT creates signed JWT access tokens.
• The pwdlib package hashes passwords.
• The python-multipart package parses the sign-in form data.

• Save requirements.txt.
• Confirm requirements.txt appears directly inside secure-task-api in the Explorer sidebar.

Can't see requirements.txt?

• Check that the Explorer sidebar shows secure-task-api as the open folder.
• Check that requirements.txt sits directly under that folder.
• Check the file name for an extra extension or spelling difference.

Help me check my project folder and requirements file.

✔️ Awesome, I've got everything!

Your pinned package list is ready. Double check that you saved requirements.txt.

ⓧ I'd like to double check the full code

fastapi==0.142.2
uvicorn==0.54.0
SQLAlchemy==2.0.54
pydantic==2.13.5
PyJWT==2.15.1
pwdlib[argon2]==0.3.1
python-multipart==0.0.32

Create and activate the virtual environment

A Windows Command Prompt terminal supports the activation command used in this project. The integrated terminal also starts inside the open secure-task-api folder.

The terminal profile control is tucked beside the plus icon in the terminal panel. This small menu lets you choose Command Prompt instead of another shell.

• Click Terminal in the Visual Studio Code top menu.
• Click New Terminal.
• Click the profile picker beside the plus icon in the terminal toolbar.
• Select Command Prompt.
• Confirm the terminal path ends with secure-task-api.

• Create the tutorial-env virtual environment by running this command:

python -m venv tutorial-env

What does this command do?

The command asks Python to create an isolated environment named tutorial-env. Its own package directory keeps this project's dependencies separate from other Python projects.

• Wait for the Command Prompt cursor to return.
• Confirm a tutorial-env folder appears in the Explorer sidebar.

Can't create tutorial-env?

• Confirm the terminal profile is Command Prompt.
• Confirm the terminal path ends with secure-task-api.
• Confirm the environment command uses tutorial-env with the same spelling.

Help me create my Windows Python virtual environment.

• Activate tutorial-env by running this command:

tutorial-env\Scripts\activate

What does activation change?

Activation points python and pip at the executables inside tutorial-env. Packages installed from this prompt stay inside the project environment.

You should see the environment name at the beginning of the Command Prompt line. This confirms that future package commands use tutorial-env.

Don't see the environment name?

• Confirm the tutorial-env folder finished creating before activation.
• Confirm you are using Command Prompt inside Visual Studio Code.
• Run the activation command again from the secure-task-api folder.

Help me activate tutorial-env in Command Prompt.

Install packages and verify FastAPI

The active environment is empty until pip reads the pinned package list. Installing from requirements.txt keeps every dependency aligned with the project.

• Install the pinned packages inside tutorial-env by running this command:

python -m pip install -r requirements.txt

What does this command do?

Python runs the pip installer from the active environment. The installer reads each exact package pin from requirements.txt.

You should see the terminal process the packages before returning to the prompt. The environment name remains visible at the beginning of the prompt.

Package installation stopped?

• Confirm the environment name remains visible at the beginning of the prompt.
• Confirm requirements.txt is saved directly inside secure-task-api.
• Check your internet connection before running the installation command again.

Help me install the pinned Python packages.

Before you run the check, which FastAPI version do you expect this environment to report?

• Verify the installed FastAPI package by running this command:

python -m pip show fastapi

What does this check prove?

The command reads the FastAPI package information from the active environment. Its version field confirms whether pip followed the pin in requirements.txt.

You should see package details that include Version: 0.142.2.

Seeing a different FastAPI version?

• Confirm the environment name appears at the beginning of the Command Prompt line.
• Compare every line in requirements.txt with the full-code tab above.
• Run the package installation command again after saving the file.

Help me fix the FastAPI version in tutorial-env.

That's your clean setup complete. FastAPI 0.142.2 now runs from an isolated environment with the rest of the pinned backend stack.

Your project environment is ready. Next up, you'll build the first working version of the task API in one file.

Build a One-File Task API

Your isolated Python environment is ready. You can now build a running FastAPI backend.

A one-file version gives you a useful baseline before the architecture grows. You can test its behavior before judging how safely it stores data or separates responsibilities.

In this step, get ready to:

• Define the application plus its task input model.
• Run the API locally with Uvicorn.
• Create a task through Swagger UI before testing what survives a restart.

Create the application and input schema

A FastAPI application receives requests. A Pydantic model defines the body that each task-creation request must contain.

• Select the secure-task-api folder in VS Code's Explorer sidebar.
• Select the new-file icon at the top of the Explorer sidebar.
• Type main.py into the filename box.
• Press Enter to create the file.
• Add the application foundation by pasting this code into main.py:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="Secure Team Task API")


class TaskCreate(BaseModel):
    title: str


tasks: list[dict] = []

What does this code do?

• The app object represents your FastAPI application.
• The TaskCreate schema requires every new task to have a text title.
• The tasks list holds task dictionaries inside the running Python process.

• Save main.py.
• Start the API from the activated Command Prompt terminal by running this command:

uvicorn main:app

What does this command do?

The uvicorn command imports the app object from main.py. The terminal stays occupied while the server handles requests.

• Return to your browser.
• Open http://127.0.0.1:8000/docs.

You'll see Swagger UI with Secure Team Task API at the top. Your first backend page is now live.

Swagger UI not loading?

Check that the Command Prompt terminal still shows the running server. Confirm that tutorial-env remains active.

Make sure main.py is inside the open secure-task-api folder.

Help me troubleshoot my local FastAPI server.

Add create and list routes

An HTTP POST route accepts new task data. The route turns that data into a dictionary stored in the shared list.

• Switch back to the Command Prompt terminal running Uvicorn.
• Press Ctrl+C to stop the server.
• Return to main.py in VS Code.
• Find the tasks: list[dict] = [] line.
• Add the create route below that line by pasting this code:

@app.post("/tasks")
def create_task(payload: TaskCreate):
    task = {"id": len(tasks) + 1, "title": payload.title, "completed": False}
    tasks.append(task)
    return task

How does task creation work?

• The @app.post("/tasks") decorator connects POST requests to create_task().
• The task ID comes from the current length of the tasks list.
• Every new task starts with completed set to False.
• The route appends the new dictionary before returning it as the response.

• Save main.py.
• Switch back to the Command Prompt terminal.
• Press the Up Arrow key to recall the previous Uvicorn command.
• Press Enter to restart the server.

Why restart Uvicorn?

The current command loads your code when Uvicorn starts. Restarting the process loads the route you just added.

• Return to Swagger UI in your browser.
• Refresh the page.
• Expand the POST /tasks route.
• Click Try it out.
• Replace the title value in the request body with Plan sprint.
• Click Execute.

You'll see id set to 1. The returned title is Plan sprint.

The completed field is false because new tasks start unfinished.

Task request not succeeding?

Confirm that the request body still contains a title field. Make sure its value remains inside double quotes.

Check the Command Prompt terminal for the file and line responsible for any Python error.

Help me fix my task creation route.

Creating one task proves that the write path works. A GET route lets you inspect every task held by the same process.

• Switch back to the Command Prompt terminal.
• Press Ctrl+C to stop Uvicorn.
• Return to main.py.
• Place your cursor below the create_task() function.
• Add the list route by pasting this code:

@app.get("/tasks")
def list_tasks():
    return tasks

How does task listing work?

• The @app.get("/tasks") decorator connects GET requests to list_tasks().
• The function returns the same tasks list used by the create route.

• Save main.py.
• Switch back to the Command Prompt terminal.
• Press the Up Arrow key to recall the Uvicorn command.
• Press Enter to restart the server.

Two Routes Share One List

Both routes now use the same tasks object. A task created by the POST route becomes available to the GET route during the same server process.

• Return to Swagger UI.
• Refresh the page.
• Repeat the POST /tasks request with Review architecture as the title.
• Expand the GET /tasks route.
• Click Try it out.
• Click Execute.

You'll see a task with the title Review architecture in the response body. Your API can now create tasks plus list them through two separate routes.

Task missing from the list?

Create the task after the most recent server restart. Confirm that both route functions reference the same tasks variable.

Refresh Swagger UI if the GET route does not appear after restarting Uvicorn.

Help me trace my in-memory task list.

✔️ Awesome, I've got everything!

Great. Double check that main.py is saved before testing the restart.

ⓧ I'd like to double check the full code

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="Secure Team Task API")


class TaskCreate(BaseModel):
    title: str


tasks: list[dict] = []


@app.post("/tasks")
def create_task(payload: TaskCreate):
    task = {"id": len(tasks) + 1, "title": payload.title, "completed": False}
    tasks.append(task)
    return task


@app.get("/tasks")
def list_tasks():
    return tasks

Test what survives a restart

The API works during a single server session. A restart tests whether the task belongs to durable storage or only to the current process.

Before you restart Uvicorn, do you think the task will still appear in the list?

• Switch back to the Command Prompt terminal.
• Press Ctrl+C to stop Uvicorn.
• Start a fresh Uvicorn process by running the same command:

uvicorn main:app

What does this restart do?

This launches a new Uvicorn process. It imports main.py again.

• Return to Swagger UI.
• Refresh the page.
• Expand the GET /tasks route.
• Click Try it out.
• Click Execute.

You'll see an empty task list. The task you created disappeared when the previous server process stopped.

The Empty List Is the Lesson

The tasks list exists only in process memory. Starting Uvicorn again creates a new empty list from the declaration in main.py.

This one-file API is useful as a baseline. Persistent storage becomes necessary once data must survive a restart.

You've built a working API and exposed its first architectural weakness. Next, you'll move task data into SQLite so it survives beyond one Uvicorn process.

Persist Tasks with SQLite

Your one-file FastAPI application can create tasks. However, every server restart erases them because the list only lives in process memory.

This step moves task storage into SQLite through SQLAlchemy. Separate Pydantic schemas keep database details out of the public API contract.

In this step, get ready to:

• Create the database connection, table model, and API schemas.
• Move database operations into a repository.
• Restart Uvicorn and prove that a task remains stored.

Define the database boundary

A database boundary separates stored records from the data your routes accept or return. The ORM model defines the table while the schemas define the API contract.

Why separate models and schemas?

A combined model library can reduce repetition. This project keeps SQLAlchemy tables separate from Pydantic schemas so you can see where persistence ends and HTTP validation begins.

That separation lets the database structure change without automatically exposing every column through the API.

• Stop the Uvicorn process in the terminal from the previous step.
• Create an app folder inside secure-task-api using the new-folder control in VS Code's Explorer sidebar.
• Create an empty __init__.py file inside app using the new-file control.

You should see app/__init__.py in the Explorer sidebar. This file marks app as the package that holds your backend layers.

• Create models.py inside the app folder using the new-file control.
• Paste the table model into models.py using the following code:

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    completed: Mapped[bool]

What Does This Model Do?

• The Base class gives SQLAlchemy the metadata it needs to create tables.
• The Task class maps Python attributes to the tasks table.
• The primary key generates a persistent identifier for every stored task.

• Save app/models.py.
• Confirm that models.py appears below __init__.py in the Explorer sidebar.

Seeing Problems in the Model?

Check that models.py sits inside the app folder. Confirm that each class line and indented field matches the code block.

Make sure the activated tutorial-env still contains the pinned SQLAlchemy package from Step 1.

Help me fix my SQLAlchemy task model.

• Create schemas.py inside the app folder using the new-file control.
• Paste the API schemas into schemas.py using the following code:

from pydantic import BaseModel, ConfigDict


class TaskCreate(BaseModel):
    title: str


class TaskPublic(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    title: str
    completed: bool

What Do These Schemas Do?

• The TaskCreate schema accepts the title supplied by the caller.
• The TaskPublic schema controls which stored fields appear in the response.
• The ConfigDict(from_attributes=True) setting lets Pydantic read values from a SQLAlchemy object.

• Save app/schemas.py.
• Confirm that the file contains separate input and public response classes.

Seeing Schema Warnings?

Check the spelling and capitalization of BaseModel and ConfigDict. Confirm that model_config is indented inside TaskPublic.

Help me fix my task schemas.

• Create database.py inside the app folder using the new-file control.
• Paste the engine and session lifecycle into database.py using the following code:

from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from app.models import Base

DATABASE_URL = "sqlite:///./task_api.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})


def create_db_and_tables():
    Base.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        yield session

How Does the Database Boundary Work?

• The DATABASE_URL points SQLAlchemy at the local task_api.db file.
• The engine manages connections to SQLite.
• The create_db_and_tables() function creates tables from the model metadata.
• The get_session() generator supplies one database session to each request.

• Save app/database.py.
• Confirm that the app folder now lists __init__.py, database.py, models.py, and schemas.py.

Database File Missing Later?

Check that DATABASE_URL contains sqlite:///./task_api.db. Confirm that create_db_and_tables() calls Base.metadata.create_all(engine).

Help me check my SQLite setup.

Isolate persistence in a repository

A repository gives database reads and writes one home. Route functions can request a task operation without managing SQLAlchemy queries or commits themselves.

• Create repositories.py inside the app folder using the new-file control.
• Paste the task persistence operations into repositories.py using the following code:

from sqlalchemy import select
from sqlalchemy.orm import Session

from app import models


def create_task(db: Session, title: str) -> models.Task:
    task = models.Task(title=title, completed=False)
    db.add(task)
    db.commit()
    db.refresh(task)
    return task


def list_tasks(db: Session) -> list[models.Task]:
    return list(db.scalars(select(models.Task).order_by(models.Task.id)))

What Does the Repository Do?

• The create_task() function creates a model object and commits it to SQLite.
• The refresh step loads the generated task identifier back onto the object.
• The list_tasks() function selects every task in identifier order.

• Save app/repositories.py.
• Confirm that repositories.py appears with the four existing files inside app.

Repository Imports Not Resolving?

Confirm that repositories.py and models.py are both inside app. Check that the package also contains the empty __init__.py file.

Help me fix my repository layer.

Connect FastAPI and prove persistence

The route layer now needs to request a session and pass it into the repository. FastAPI handles that handoff through dependency injection.

• Switch back to main.py from the previous step.
• Select all of the existing in-memory application code.
• Replace the selected code with the following database-backed application:

from contextlib import asynccontextmanager

from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session

from app.database import create_db_and_tables, get_session
from app import repositories, schemas


@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()
    yield


app = FastAPI(title="Secure Team Task API", lifespan=lifespan)


@app.post("/tasks", response_model=schemas.TaskPublic)
def create_task(payload: schemas.TaskCreate, db: Session = Depends(get_session)):
    return repositories.create_task(db, payload.title)


@app.get("/tasks", response_model=list[schemas.TaskPublic])
def list_tasks(db: Session = Depends(get_session)):
    return repositories.list_tasks(db)

How Does the Request Flow Work?

• The lifespan function creates the database table when the application starts.
• The Depends(get_session) parameter supplies a SQLAlchemy session to each route.
• The routes delegate storage work to repositories.create_task() and repositories.list_tasks().
• The response models convert stored task objects into the public schema.

• Save main.py.

Before you start the application, which new file do you expect SQLAlchemy to create in secure-task-api?

• Start the database-backed application by running this command in the activated terminal:

uvicorn main:app

What Does This Command Start?

Uvicorn imports the app object from main.py. The lifespan function runs during startup and creates the SQLite table.

You should see Uvicorn start successfully. The task_api.db file should now appear beside main.py in the Explorer sidebar.

Server Fails During Startup?

Confirm that the terminal is still inside secure-task-api with tutorial-env active. Check that every file shown in the earlier structure is saved.

If the problem mentions an import, compare the module names in main.py with the filenames inside app.

Help me start the database-backed API.

The refreshed Swagger UI now sends task operations through a database session and repository.

• Return to the Swagger UI tab at http://127.0.0.1:8000/docs.
• Refresh the page.
• Expand the POST method for /tasks.
• Select Try it out.
• Set the title value to Persistence survives restarts.
• Select Execute.

You should see a response containing an identifier, the title Persistence survives restarts, and a false completed value.

• Expand the GET method for /tasks.
• Select Try it out.
• Select Execute.

You should see the stored task inside the returned list. This proves that the route can read the row back from SQLite.

What Have You Proved So Far?

The task now exists in task_api.db after the repository commits it. The restart test checks whether that record survives beyond the current Uvicorn process.

Before you restart Uvicorn, do you expect the task list to keep the task or return to an empty list?

• Stop the running Uvicorn process in its terminal.
• Restart the application by running the same command:

uvicorn main:app

Why Restart the Server?

Restarting creates a new Python process with fresh memory. The new process connects to the existing task_api.db file, so committed rows remain available.

• Return to the Swagger UI tab from earlier.
• Refresh the page.
• Expand the GET method for /tasks.
• Select Try it out.
• Select Execute.

You should still see the task titled Persistence survives restarts in the returned list. That is the payoff: your task survived a full server restart because SQLite holds it in task_api.db.

Task List Empty After Restart?

Confirm that you restarted Uvicorn from the same secure-task-api folder. Starting it from another location can point the relative database URL at a different file.

Check that create_task() calls db.commit() before returning the task. Confirm that you did not delete task_api.db between runs.

Help me find why my task disappeared.

✔️ Awesome, I've got everything!

Great. Double check that every file is saved and keep Uvicorn running for the next step.

ⓧ I'd like to double check the full code

The app/__init__.py file should be empty. Its presence makes app a Python package.

app/database.py:

from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from app.models import Base

DATABASE_URL = "sqlite:///./task_api.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})


def create_db_and_tables():
    Base.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        yield session

Database Boundary

This file owns the SQLite engine, table creation, and request-scoped session lifecycle.

app/models.py:

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    completed: Mapped[bool]

Table Model

This file defines the metadata base and the persistent task table.

app/repositories.py:

from sqlalchemy import select
from sqlalchemy.orm import Session

from app import models


def create_task(db: Session, title: str) -> models.Task:
    task = models.Task(title=title, completed=False)
    db.add(task)
    db.commit()
    db.refresh(task)
    return task


def list_tasks(db: Session) -> list[models.Task]:
    return list(db.scalars(select(models.Task).order_by(models.Task.id)))

Repository Boundary

This file owns the SQLAlchemy operations that create and list stored tasks.

app/schemas.py:

from pydantic import BaseModel, ConfigDict


class TaskCreate(BaseModel):
    title: str


class TaskPublic(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    title: str
    completed: bool

API Contracts

This file defines the input and public response shapes used by the task routes.

main.py:

from contextlib import asynccontextmanager

from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session

from app.database import create_db_and_tables, get_session
from app import repositories, schemas


@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()
    yield


app = FastAPI(title="Secure Team Task API", lifespan=lifespan)


@app.post("/tasks", response_model=schemas.TaskPublic)
def create_task(payload: schemas.TaskCreate, db: Session = Depends(get_session)):
    return repositories.create_task(db, payload.title)


@app.get("/tasks", response_model=list[schemas.TaskPublic])
def list_tasks(db: Session = Depends(get_session)):
    return repositories.list_tasks(db)

Application Wiring

This file starts the table lifecycle and connects the HTTP routes to schemas, sessions, and repository operations.

requirements.txt:

fastapi==0.142.2
uvicorn==0.54.0
SQLAlchemy==2.0.54
pydantic==2.13.5
PyJWT==2.15.1
pwdlib[argon2]==0.3.1
python-multipart==0.0.32

Pinned Packages

This unchanged file keeps every backend dependency at the version installed in Step 1.

The generated task_api.db file should sit beside main.py. It contains the persistent tasks table and the record created through Swagger UI.

Your API now keeps tasks beyond the lifetime of one server process. Next, you'll separate the remaining layers and add user sign-in.

Separate Layers and Add Sign-In

Your SQLite database now keeps tasks after the server restarts. However, every request is still anonymous.

This step adds user registration, password hashing, and JWT authentication. You will also split HTTP handling, business rules, database access, and security into separate modules.

In this step, get ready to:

• Add users, authentication schemas, and security helpers.
• Move business rules into services and HTTP handling into routers.
• Register Alice and Bob before testing authenticated task access.

Define users and authentication

Authentication needs three distinct pieces. A user table stores identity, public schemas control exposed data, and security helpers protect passwords plus tokens.

Why separate SQLAlchemy and Pydantic models?

The SQLAlchemy models describe stored database rows. The Pydantic schemas describe accepted requests and returned responses.

This boundary prevents fields such as password hashes from becoming part of the public API contract.

• Stop the Uvicorn process in the terminal from earlier by pressing Ctrl+C.
• In the VS Code Explorer sidebar, replace the contents of app/models.py with this code:

from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    role: Mapped[str] = mapped_column(String(20))


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    completed: Mapped[bool]
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

What do these models store?

• The User model stores a unique username, a password hash, and a role.
• The Task model now links each task to a user through owner_id.
• The ForeignKey declaration makes each stored owner ID refer to the users table.

• Save app/models.py.
• Replace the contents of app/schemas.py with the user contracts below:

from pydantic import BaseModel, ConfigDict


class UserCreate(BaseModel):
    username: str
    password: str


class UserPublic(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    username: str
    role: str


class Token(BaseModel):
    access_token: str
    token_type: str

What do the user schemas protect?

The registration request accepts a username and password through UserCreate. The response uses UserPublic, which has no password or hash field.

The Token schema gives successful sign-ins a consistent response shape.

• Append the task contracts below Token in app/schemas.py using this code:

class TaskCreate(BaseModel):
    title: str


class TaskUpdate(BaseModel):
    title: str | None = None
    completed: bool | None = None


class TaskPublic(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    title: str
    completed: bool
    owner_id: int

How do task updates work?

The optional fields in TaskUpdate support partial updates. A request can change the title, the completion state, or both.

The public response includes owner_id so you can observe which user owns each task.

• Save app/schemas.py.
• Replace app/database.py with the session dependency below:

from typing import Annotated

from fastapi import Depends
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from app.models import Base

DATABASE_URL = "sqlite:///./task_api.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False},
)


def create_db_and_tables():
    Base.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        yield session


DBSession = Annotated[Session, Depends(get_session)]

What does DBSession provide?

The DBSession alias combines a SQLAlchemy session type with a FastAPI dependency. Routes can request a session without repeating the dependency declaration.

The dependency opens one session for a request and closes it after the request finishes.

• Save app/database.py.
• Replace app/repositories.py with these imports and user queries:

from sqlalchemy import select
from sqlalchemy.orm import Session

from app import models


def get_user_by_username(db: Session, username: str) -> models.User | None:
    statement = select(models.User).where(models.User.username == username)
    return db.scalars(statement).first()


def get_first_user(db: Session) -> models.User | None:
    statement = select(models.User).order_by(models.User.id)
    return db.scalars(statement).first()

What do these queries answer?

• The username query finds the account used during registration or sign-in.
• The first-user query tells the service whether it is creating the initial local account.
• Keeping these queries in the repository stops SQLAlchemy statements from spreading into route functions.

• Append the user creation function to app/repositories.py using this code:

def create_user(
    db: Session,
    username: str,
    hashed_password: str,
    role: str,
) -> models.User:
    user = models.User(
        username=username,
        hashed_password=hashed_password,
        role=role,
    )
    db.add(user)
    db.commit()
    db.refresh(user)
    return user

How is a user persisted?

The repository receives an existing password hash from the service. It adds the user row, commits the transaction, and refreshes the object with its generated ID.

• Append the task lookup and list functions to app/repositories.py using this code:

def get_task(db: Session, task_id: int) -> models.Task | None:
    return db.get(models.Task, task_id)


def list_tasks_for_owner(db: Session, owner_id: int) -> list[models.Task]:
    statement = (
        select(models.Task)
        .where(models.Task.owner_id == owner_id)
        .order_by(models.Task.id)
    )
    return list(db.scalars(statement))

How are tasks selected?

The repository can retrieve one task by ID. It can also list tasks whose owner_id matches the signed-in user.

• Append the task write functions to app/repositories.py using this code:

def create_task(db: Session, title: str, owner_id: int) -> models.Task:
    task = models.Task(title=title, completed=False, owner_id=owner_id)
    db.add(task)
    db.commit()
    db.refresh(task)
    return task


def save_task(db: Session, task: models.Task) -> models.Task:
    db.commit()
    db.refresh(task)
    return task


def delete_task(db: Session, task: models.Task) -> None:
    db.delete(task)
    db.commit()

Where do database writes happen?

The repository owns every SQLAlchemy write and commit. The service layer can apply rules without managing transaction details.

• Save app/repositories.py.
• Create app/security.py from the VS Code Explorer sidebar.
• Add the password and token configuration below to app/security.py using this code:

from datetime import datetime, timedelta, timezone

import jwt
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash

SECRET_KEY = "replace-this-development-key-before-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

password_hash = PasswordHash.recommended()
DUMMY_HASH = password_hash.hash("dummypassword")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")


def hash_password(password: str) -> str:
    return password_hash.hash(password)


def verify_password(plain_password: str, hashed_password: str) -> bool:
    return password_hash.verify(plain_password, hashed_password)

How are passwords protected?

Password hashing converts a password into a one-way stored value. Registration stores the hash while sign-in verifies the submitted password against it.

The dummy hash supports the same verification work when a username is missing. This reduces timing differences between unknown users and incorrect passwords.

• Append the access-token functions to app/security.py using this code:

def create_access_token(subject: str) -> str:
    expires_at = datetime.now(timezone.utc) + timedelta(
        minutes=ACCESS_TOKEN_EXPIRE_MINUTES
    )
    return jwt.encode(
        {"sub": subject, "exp": expires_at},
        SECRET_KEY,
        algorithm=ALGORITHM,
    )


def decode_access_token(token: str) -> str | None:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        subject = payload.get("sub")
        return subject if isinstance(subject, str) else None
    except InvalidTokenError:
        return None

What goes inside the JWT?

The token places the username in sub and an expiry timestamp in exp. The configured lifetime is 30 minutes.

Decoding verifies the HS256 signature before returning the username. Invalid tokens produce no identity.

Use a managed secret in production

The development signing key keeps this local project reproducible. A deployed service must load a strong secret from protected configuration instead of committing it to source control.

• Save app/security.py.
• Create app/dependencies.py from the VS Code Explorer sidebar.
• Add the current-user dependency below to app/dependencies.py using this code:

from typing import Annotated

from fastapi import Depends, HTTPException, status
from sqlalchemy.orm import Session

from app import models, repositories, security
from app.database import get_session

DBSession = Annotated[Session, Depends(get_session)]


def get_current_user(
    token: Annotated[str, Depends(security.oauth2_scheme)],
    db: DBSession,
) -> models.User:
    username = security.decode_access_token(token)
    user = repositories.get_user_by_username(db, username) if username else None
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user


CurrentUser = Annotated[models.User, Depends(get_current_user)]

How does dependency injection identify a user?

Dependency injection lets a protected route request CurrentUser. FastAPI extracts the bearer token, decodes its username, and loads the matching user.

Missing or invalid identity stops the request with 401 before task logic runs.

Move rules into services and routers

The service layer decides what the application allows. Routers stay focused on HTTP input, response models, and translating domain errors into status codes.

• Create app/exceptions.py from the VS Code Explorer sidebar.
• Add the domain exceptions below to app/exceptions.py using this code:

class UsernameTakenError(Exception):
    pass


class InvalidCredentialsError(Exception):
    pass


class TaskNotFoundError(Exception):
    pass


class ForbiddenError(Exception):
    pass

Why use domain exceptions?

These exceptions describe application failures without mentioning HTTP. Routers decide whether each failure becomes a 401, 403, 404, or 409 response.

• Create app/services.py from the VS Code Explorer sidebar.
• Add the registration and authentication rules below to app/services.py using this code:

from sqlalchemy.orm import Session

from app import exceptions, models, repositories, schemas, security


def register_user(db: Session, payload: schemas.UserCreate) -> models.User:
    if repositories.get_user_by_username(db, payload.username):
        raise exceptions.UsernameTakenError

    role = "admin" if repositories.get_first_user(db) is None else "member"
    return repositories.create_user(
        db,
        username=payload.username,
        hashed_password=security.hash_password(payload.password),
        role=role,
    )


def authenticate_user(db: Session, username: str, password: str) -> models.User:
    user = repositories.get_user_by_username(db, username)
    if user is None:
        security.verify_password(password, security.DUMMY_HASH)
        raise exceptions.InvalidCredentialsError
    if not security.verify_password(password, user.hashed_password):
        raise exceptions.InvalidCredentialsError
    return user

Which rules live here?

Registration rejects duplicate usernames and hashes new passwords. The first local account receives the admin role while later accounts receive member.

Authentication returns a user only when the submitted password matches the stored hash.

• Append the task creation and listing rules to app/services.py using this code:

def create_task(
    db: Session,
    payload: schemas.TaskCreate,
    current_user: models.User,
) -> models.Task:
    return repositories.create_task(db, payload.title, current_user.id)


def list_own_tasks(db: Session, current_user: models.User) -> list[models.Task]:
    return repositories.list_tasks_for_owner(db, current_user.id)

How is ownership assigned?

Task creation sends the current user's ID to the repository. Task listing filters records by that same ID.

• Append the task update rule to app/services.py using this code:

def update_task(
    db: Session,
    task_id: int,
    payload: schemas.TaskUpdate,
    current_user: models.User,
) -> models.Task:
    task = repositories.get_task(db, task_id)
    if task is None:
        raise exceptions.TaskNotFoundError

    for field, value in payload.model_dump(exclude_unset=True).items():
        if value is not None:
            setattr(task, field, value)
    return repositories.save_task(db, task)

How are partial updates applied?

The service loads the requested task before applying only the fields supplied by the caller. model_dump(exclude_unset=True) leaves omitted fields unchanged.

• Append the task deletion rule to app/services.py using this code:

def delete_task(
    db: Session,
    task_id: int,
    current_user: models.User,
) -> None:
    task = repositories.get_task(db, task_id)
    if task is None:
        raise exceptions.TaskNotFoundError
    repositories.delete_task(db, task)

How is deletion coordinated?

The service checks that the task exists before asking the repository to delete it. A missing task becomes a domain error for the router to translate.

• Save app/services.py.
• Create an app/routers folder from the VS Code Explorer sidebar.
• Create an empty app/routers/__init__.py file inside app/routers.
• Create app/routers/auth.py inside app/routers.
• Add the registration route below to app/routers/auth.py using this code:

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm

from app import exceptions, schemas, security, services
from app.database import DBSession
from app.dependencies import CurrentUser

router = APIRouter(prefix="/auth", tags=["auth"])


@router.post(
    "/register",
    response_model=schemas.UserPublic,
    status_code=status.HTTP_201_CREATED,
)
def register(payload: schemas.UserCreate, db: DBSession):
    try:
        return services.register_user(db, payload)
    except exceptions.UsernameTakenError:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Username already exists",
        )

What does the registration router handle?

The router sends valid registration data to the service. It translates a duplicate username into an HTTP 409 response.

The UserPublic response model prevents the stored password hash from appearing in Swagger UI.

• Append the sign-in and current-user routes to app/routers/auth.py using this code:

@router.post("/token", response_model=schemas.Token)
def login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: DBSession,
):
    try:
        user = services.authenticate_user(
            db,
            username=form_data.username,
            password=form_data.password,
        )
    except exceptions.InvalidCredentialsError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )

    return schemas.Token(
        access_token=security.create_access_token(user.username),
        token_type="bearer",
    )


@router.get("/me", response_model=schemas.UserPublic)
def read_me(current_user: CurrentUser):
    return current_user

How does sign-in issue a token?

The token route authenticates the submitted form credentials. Successful authentication creates a bearer token whose subject is the username.

The protected GET /auth/me route proves that CurrentUser can resolve that token back to a public user record.

• Save app/routers/auth.py.
• Create app/routers/tasks.py inside app/routers.
• Add the protected create and list routes below to app/routers/tasks.py using this code:

from fastapi import APIRouter, HTTPException, status

from app import exceptions, schemas, services
from app.database import DBSession
from app.dependencies import CurrentUser

router = APIRouter(prefix="/tasks", tags=["tasks"])


@router.post(
    "",
    response_model=schemas.TaskPublic,
    status_code=status.HTTP_201_CREATED,
)
def create_task(
    payload: schemas.TaskCreate,
    db: DBSession,
    current_user: CurrentUser,
):
    return services.create_task(db, payload, current_user)


@router.get("", response_model=list[schemas.TaskPublic])
def list_tasks(db: DBSession, current_user: CurrentUser):
    return services.list_own_tasks(db, current_user)

What protects these routes?

Both routes require CurrentUser. A valid bearer token must resolve to a stored user before the service receives the request.

• Append the update route to app/routers/tasks.py using this code:

@router.patch("/{task_id}", response_model=schemas.TaskPublic)
def update_task(
    task_id: int,
    payload: schemas.TaskUpdate,
    db: DBSession,
    current_user: CurrentUser,
):
    try:
        return services.update_task(db, task_id, payload, current_user)
    except exceptions.TaskNotFoundError:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found",
        )
    except exceptions.ForbiddenError:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="You do not own this task",
        )

How does the router translate failures?

A missing task becomes an HTTP 404 response. The router is also ready to translate ForbiddenError into an HTTP 403 response.

• Append the delete route to app/routers/tasks.py using this code:

@router.delete("/{task_id}")
def delete_task(
    task_id: int,
    db: DBSession,
    current_user: CurrentUser,
):
    try:
        services.delete_task(db, task_id, current_user)
        return {"ok": True}
    except exceptions.TaskNotFoundError:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found",
        )
    except exceptions.ForbiddenError:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="You do not own this task",
        )

What does deletion return?

A successful deletion returns {"ok": True}. Missing or forbidden tasks follow the same HTTP translation pattern as updates.

• Save app/routers/tasks.py.
• Replace the contents of main.py with this application composition code:

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.database import create_db_and_tables
from app.routers import auth, tasks


@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()
    yield


app = FastAPI(title="Secure Team Task API", lifespan=lifespan)
app.include_router(auth.router)
app.include_router(tasks.router)


@app.get("/health")
def health():
    return {"status": "ok"}

What remains in main.py?

The application entry point now creates database tables, includes both routers, and exposes a health route. Routing details and business rules live in their dedicated modules.

This leaves main.py responsible for startup and composition.

• Save main.py.

Seeing import warnings in VS Code?

Confirm that app/routers/__init__.py exists. Confirm that both auth.py and tasks.py are inside the app/routers folder.

Check that the active VS Code interpreter comes from tutorial-env.

Help me trace an import error across my FastAPI app package.

Rebuild the database and test sign-in

The existing database was created before users and task owners existed. Rebuilding it once gives SQLAlchemy a clean schema for this learning project.

Deleting the practice database removes the tasks you created earlier. That is expected here because those rows have no owner IDs.

• In the VS Code Explorer sidebar, select task_api.db.
• Press the Delete key to remove the practice database.
• Confirm the deletion if VS Code asks for confirmation.
• Return to the Command Prompt terminal with tutorial-env active.

Before you start the server, which route groups do you expect the rebuilt Swagger UI to show?

• Start the layered application by running this command:

uvicorn main:app

What does this command start?

Uvicorn imports the app object from main.py. The lifespan function rebuilds task_api.db with the new tables before requests begin.

• Visit http://127.0.0.1:8000/docs in your browser.

You should see separate auth and tasks groups. You should also see an authorization control for protected routes.

• Expand GET /tasks.
• Click Try it out.
• Click Execute without authorizing.

You should see a 401 response. This proves the route rejects requests that have no authenticated identity.

• Expand POST /auth/register.
• Click Try it out.
• Enter Alice as the username.
• Enter a local practice password that you can re-enter during this step.
• Click Execute.

The response should contain Alice's public user record with the role admin. No password or password hash should appear.

• Use POST /auth/register again.
• Enter Bob as the username.
• Enter a second local practice password that you can re-enter during this step.
• Click Execute.

The second public user record should show Bob with the role member.

Why is Alice the local administrator?

The registration service assigns the first local account the admin role. Every later account receives the member role.

A production system should provision administrators through controlled configuration or an audited process.

• Click Authorize in Swagger UI.
• Enter Alice's username in the username field.
• Enter Alice's local practice password in the password field.
• Submit the authorization form.
• Close the authorization dialog.

Swagger UI now sends Alice's bearer token with protected requests.

• Expand POST /tasks.
• Click Try it out.
• Enter Alice planning task as the task title.
• Click Execute.

You should receive a created task with Alice's integer owner_id.

• Record the new task ID here: Alice's task ID.

Registration or authorization failing?

A duplicate registration returns a conflict response. Delete task_api.db only if you want to restart this local two-user scenario from the beginning.

If authorization rejects the login, re-enter the same username and password used during registration. Usernames and passwords are case-sensitive.

Help me troubleshoot registration or OAuth2 password sign-in in my local FastAPI project.

Before Bob sends the update, do you think authentication alone prevents him from changing Alice's task?

• Return to the Swagger UI authorization dialog.
• Clear Alice's current authorization.
• Enter Bob's username in the username field.
• Enter Bob's local practice password in the password field.
• Submit the authorization form.
• Close the authorization dialog.
• Expand PATCH /tasks/{task_id}.
• Click Try it out.

• Enter Alice's task ID in the task ID field.
• Set the completed field to true.
• Click Execute.

You should receive a successful response showing Alice's task as completed. Bob has changed a task whose owner_id belongs to Alice.

This successful update is the intended problem

The API knows that Bob is signed in, but the update service never compares Bob's ID with the task's owner_id. Authentication has established identity without enforcing permission.

The next step closes this gap in the service layer and turns Bob's cross-user update into a 403 Forbidden response.

✔️ Awesome, I've got everything!

Great work. Your API now registers users, issues bearer tokens, protects task routes, and exposes the authorization gap that the next step fixes.

ⓧ I'd like to double check the full code

Compare each file below with your saved project. The service code intentionally reflects the current authentication-only state.

from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    role: Mapped[str] = mapped_column(String(20))


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    completed: Mapped[bool]
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

This file defines the stored users and owned tasks.

from pydantic import BaseModel, ConfigDict


class UserCreate(BaseModel):
    username: str
    password: str


class UserPublic(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    username: str
    role: str


class Token(BaseModel):
    access_token: str
    token_type: str


class TaskCreate(BaseModel):
    title: str


class TaskUpdate(BaseModel):
    title: str | None = None
    completed: bool | None = None


class TaskPublic(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    title: str
    completed: bool
    owner_id: int

This file defines the public user, token, and task contracts.

from typing import Annotated

from fastapi import Depends
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from app.models import Base

DATABASE_URL = "sqlite:///./task_api.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False},
)


def create_db_and_tables():
    Base.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        yield session


DBSession = Annotated[Session, Depends(get_session)]

This file creates the SQLite engine and request-scoped database sessions.

from sqlalchemy import select
from sqlalchemy.orm import Session

from app import models


def get_user_by_username(db: Session, username: str) -> models.User | None:
    statement = select(models.User).where(models.User.username == username)
    return db.scalars(statement).first()


def get_first_user(db: Session) -> models.User | None:
    statement = select(models.User).order_by(models.User.id)
    return db.scalars(statement).first()


def create_user(
    db: Session,
    username: str,
    hashed_password: str,
    role: str,
) -> models.User:
    user = models.User(
        username=username,
        hashed_password=hashed_password,
        role=role,
    )
    db.add(user)
    db.commit()
    db.refresh(user)
    return user


def get_task(db: Session, task_id: int) -> models.Task | None:
    return db.get(models.Task, task_id)


def list_tasks_for_owner(db: Session, owner_id: int) -> list[models.Task]:
    statement = (
        select(models.Task)
        .where(models.Task.owner_id == owner_id)
        .order_by(models.Task.id)
    )
    return list(db.scalars(statement))


def create_task(db: Session, title: str, owner_id: int) -> models.Task:
    task = models.Task(title=title, completed=False, owner_id=owner_id)
    db.add(task)
    db.commit()
    db.refresh(task)
    return task


def save_task(db: Session, task: models.Task) -> models.Task:
    db.commit()
    db.refresh(task)
    return task


def delete_task(db: Session, task: models.Task) -> None:
    db.delete(task)
    db.commit()

This file owns user and task queries plus database commits.

class UsernameTakenError(Exception):
    pass


class InvalidCredentialsError(Exception):
    pass


class TaskNotFoundError(Exception):
    pass


class ForbiddenError(Exception):
    pass

This file names application failures independently from HTTP.

from datetime import datetime, timedelta, timezone

import jwt
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash

SECRET_KEY = "replace-this-development-key-before-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

password_hash = PasswordHash.recommended()
DUMMY_HASH = password_hash.hash("dummypassword")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")


def hash_password(password: str) -> str:
    return password_hash.hash(password)


def verify_password(plain_password: str, hashed_password: str) -> bool:
    return password_hash.verify(plain_password, hashed_password)


def create_access_token(subject: str) -> str:
    expires_at = datetime.now(timezone.utc) + timedelta(
        minutes=ACCESS_TOKEN_EXPIRE_MINUTES
    )
    return jwt.encode(
        {"sub": subject, "exp": expires_at},
        SECRET_KEY,
        algorithm=ALGORITHM,
    )


def decode_access_token(token: str) -> str | None:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        subject = payload.get("sub")
        return subject if isinstance(subject, str) else None
    except InvalidTokenError:
        return None

This file hashes passwords and creates or validates access tokens.

from typing import Annotated

from fastapi import Depends, HTTPException, status
from sqlalchemy.orm import Session

from app import models, repositories, security
from app.database import get_session

DBSession = Annotated[Session, Depends(get_session)]


def get_current_user(
    token: Annotated[str, Depends(security.oauth2_scheme)],
    db: DBSession,
) -> models.User:
    username = security.decode_access_token(token)
    user = repositories.get_user_by_username(db, username) if username else None
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user


CurrentUser = Annotated[models.User, Depends(get_current_user)]

This file resolves bearer tokens into the current stored user.

from sqlalchemy.orm import Session

from app import exceptions, models, repositories, schemas, security


def register_user(db: Session, payload: schemas.UserCreate) -> models.User:
    if repositories.get_user_by_username(db, payload.username):
        raise exceptions.UsernameTakenError

    role = "admin" if repositories.get_first_user(db) is None else "member"
    return repositories.create_user(
        db,
        username=payload.username,
        hashed_password=security.hash_password(payload.password),
        role=role,
    )


def authenticate_user(db: Session, username: str, password: str) -> models.User:
    user = repositories.get_user_by_username(db, username)
    if user is None:
        security.verify_password(password, security.DUMMY_HASH)
        raise exceptions.InvalidCredentialsError
    if not security.verify_password(password, user.hashed_password):
        raise exceptions.InvalidCredentialsError
    return user


def create_task(
    db: Session,
    payload: schemas.TaskCreate,
    current_user: models.User,
) -> models.Task:
    return repositories.create_task(db, payload.title, current_user.id)


def list_own_tasks(db: Session, current_user: models.User) -> list[models.Task]:
    return repositories.list_tasks_for_owner(db, current_user.id)


def update_task(
    db: Session,
    task_id: int,
    payload: schemas.TaskUpdate,
    current_user: models.User,
) -> models.Task:
    task = repositories.get_task(db, task_id)
    if task is None:
        raise exceptions.TaskNotFoundError

    for field, value in payload.model_dump(exclude_unset=True).items():
        if value is not None:
            setattr(task, field, value)
    return repositories.save_task(db, task)


def delete_task(
    db: Session,
    task_id: int,
    current_user: models.User,
) -> None:
    task = repositories.get_task(db, task_id)
    if task is None:
        raise exceptions.TaskNotFoundError
    repositories.delete_task(db, task)

This file owns registration, authentication, and the current authentication-only task rules.

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm

from app import exceptions, schemas, security, services
from app.database import DBSession
from app.dependencies import CurrentUser

router = APIRouter(prefix="/auth", tags=["auth"])


@router.post(
    "/register",
    response_model=schemas.UserPublic,
    status_code=status.HTTP_201_CREATED,
)
def register(payload: schemas.UserCreate, db: DBSession):
    try:
        return services.register_user(db, payload)
    except exceptions.UsernameTakenError:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Username already exists",
        )


@router.post("/token", response_model=schemas.Token)
def login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: DBSession,
):
    try:
        user = services.authenticate_user(
            db,
            username=form_data.username,
            password=form_data.password,
        )
    except exceptions.InvalidCredentialsError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )

    return schemas.Token(
        access_token=security.create_access_token(user.username),
        token_type="bearer",
    )


@router.get("/me", response_model=schemas.UserPublic)
def read_me(current_user: CurrentUser):
    return current_user

This router translates authentication requests and errors into HTTP responses.

from fastapi import APIRouter, HTTPException, status

from app import exceptions, schemas, services
from app.database import DBSession
from app.dependencies import CurrentUser

router = APIRouter(prefix="/tasks", tags=["tasks"])


@router.post(
    "",
    response_model=schemas.TaskPublic,
    status_code=status.HTTP_201_CREATED,
)
def create_task(
    payload: schemas.TaskCreate,
    db: DBSession,
    current_user: CurrentUser,
):
    return services.create_task(db, payload, current_user)


@router.get("", response_model=list[schemas.TaskPublic])
def list_tasks(db: DBSession, current_user: CurrentUser):
    return services.list_own_tasks(db, current_user)


@router.patch("/{task_id}", response_model=schemas.TaskPublic)
def update_task(
    task_id: int,
    payload: schemas.TaskUpdate,
    db: DBSession,
    current_user: CurrentUser,
):
    try:
        return services.update_task(db, task_id, payload, current_user)
    except exceptions.TaskNotFoundError:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found",
        )
    except exceptions.ForbiddenError:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="You do not own this task",
        )


@router.delete("/{task_id}")
def delete_task(
    task_id: int,
    db: DBSession,
    current_user: CurrentUser,
):
    try:
        services.delete_task(db, task_id, current_user)
        return {"ok": True}
    except exceptions.TaskNotFoundError:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found",
        )
    except exceptions.ForbiddenError:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="You do not own this task",
        )

This router exposes protected task operations and translates domain errors.

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.database import create_db_and_tables
from app.routers import auth, tasks


@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()
    yield


app = FastAPI(title="Secure Team Task API", lifespan=lifespan)
app.include_router(auth.router)
app.include_router(tasks.router)


@app.get("/health")
def health():
    return {"status": "ok"}

This file composes the routers and handles application startup.

You have a layered API with persistent users, hashed passwords, token-protected routes, and a visible permission gap. Next, you will enforce task ownership where the business rules belong.

Enforce Task Ownership

Your API already recognizes Alice and Bob through JWT bearer tokens. However, Bob's successful edit of Alice's task proved that identity alone cannot protect someone else's data.

This step adds ownership authorization inside the service layer. The FastAPI routers will translate forbidden operations into visible responses.

In this step, get ready to:

• Add ownership checks to task update operations.
• Apply the same ownership rule to task deletion.
• Contrast identity failure with forbidden access.

Guard task updates in the service layer

Authentication establishes who sent a request. Ownership authorization decides whether that person can change the selected task.

Why does the service own this rule?

The service layer already coordinates task lookups with business decisions. Keeping the ownership comparison here makes the same rule available to every route that calls the service.

The router stays focused on translating the result into an HTTP response. The repository stays focused on database access.

• Record the id from Alice's task response here: alice's task ID.
• Switch back to Visual Studio Code.
• Select app/services.py in the Explorer sidebar.
• Replace the complete update_task() function with this version:

def update_task(
    db: Session,
    task_id: int,
    payload: schemas.TaskUpdate,
    current_user: models.User,
) -> models.Task:
    task = repositories.get_task(db, task_id)
    if task is None:
        raise exceptions.TaskNotFoundError
    if task.owner_id != current_user.id:
        raise exceptions.ForbiddenError

    for field, value in payload.model_dump(exclude_unset=True).items():
        if value is not None:
            setattr(task, field, value)
    return repositories.save_task(db, task)

What does this code do?

• The repository lookup retrieves the task before any change is applied.
• The TaskNotFoundError path handles an unknown task ID.
• The owner_id comparison stops a different user before the update loop runs.
• The existing partial-update loop saves permitted changes.

• Save app/services.py.
• Confirm the ownership comparison appears before the model_dump() loop.
• Select app/routers/tasks.py in the Explorer sidebar.
• Replace the complete router-level update_task() function with this version:

@router.patch("/{task_id}", response_model=schemas.TaskPublic)
def update_task(
    task_id: int,
    payload: schemas.TaskUpdate,
    db: DBSession,
    current_user: CurrentUser,
):
    try:
        return services.update_task(db, task_id, payload, current_user)
    except exceptions.TaskNotFoundError:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found",
        )
    except exceptions.ForbiddenError:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="You do not own this task",
        )

What does this route do?

The route passes CurrentUser into the service. It catches ForbiddenError after the service rejects a cross-user update.

The router converts that domain error into an HTTP response using status.HTTP_403_FORBIDDEN. This keeps transport details outside the business rule.

• Save app/routers/tasks.py.
• Switch back to the Windows Command Prompt terminal from earlier.
• Stop the running Uvicorn process with Ctrl+C.
• Restart the Secure Team Task API by running:

uvicorn main:app

Why restart Uvicorn?

The running process loaded the earlier function definitions. Restarting Uvicorn loads the new ownership check while preserving Alice's task in task_api.db.

Before you resend Bob's update, do you expect the authenticated request to reach the service or stop at the token check?

• Return to the Swagger UI browser tab from earlier.
• Keep Bob's saved authorization active.
• Expand PATCH /tasks/{task_id}.
• Enter alice's task ID as the path value.
• Replace the example title value with Bob blocked.
• Send the request from the expanded route.

You'll receive 403 Forbidden with the detail You do not own this task. Bob is authenticated, but the service now refuses his update.

Still seeing a successful update?

Confirm you restarted Uvicorn after saving both files. Check that the ownership comparison uses task.owner_id with current_user.id.

Make sure the router catches exceptions.ForbiddenError at the same indentation level as the missing-task handler.

Help me debug the ownership check.

Apply the same rule to task deletion

Deletion needs the same ownership boundary as updates. A shared rule prevents a signed-in user from removing another person's persistent task.

• Return to app/services.py in Visual Studio Code.
• Replace the complete delete_task() function with this version:

def delete_task(
    db: Session,
    task_id: int,
    current_user: models.User,
) -> None:
    task = repositories.get_task(db, task_id)
    if task is None:
        raise exceptions.TaskNotFoundError
    if task.owner_id != current_user.id:
        raise exceptions.ForbiddenError
    repositories.delete_task(db, task)

What does this code do?

• The service retrieves the task before attempting deletion.
• The ownership comparison runs before repositories.delete_task().
• Only the owner reaches the repository operation that commits the deletion.

• Save app/services.py.
• Confirm both task mutation functions now raise ForbiddenError for an ownership mismatch.

✔️ Awesome, I've got everything!

Great. Your service layer now protects both update and delete operations with the same owner comparison.

ⓧ I'd like to double check the full code

from sqlalchemy.orm import Session

from app import exceptions, models, repositories, schemas, security


def register_user(db: Session, payload: schemas.UserCreate) -> models.User:
    if repositories.get_user_by_username(db, payload.username):
        raise exceptions.UsernameTakenError

    role = "admin" if repositories.get_first_user(db) is None else "member"
    return repositories.create_user(
        db,
        username=payload.username,
        hashed_password=security.hash_password(payload.password),
        role=role,
    )


def authenticate_user(db: Session, username: str, password: str) -> models.User:
    user = repositories.get_user_by_username(db, username)
    if user is None:
        security.verify_password(password, security.DUMMY_HASH)
        raise exceptions.InvalidCredentialsError
    if not security.verify_password(password, user.hashed_password):
        raise exceptions.InvalidCredentialsError
    return user


def create_task(
    db: Session,
    payload: schemas.TaskCreate,
    current_user: models.User,
) -> models.Task:
    return repositories.create_task(db, payload.title, current_user.id)


def list_own_tasks(db: Session, current_user: models.User) -> list[models.Task]:
    return repositories.list_tasks_for_owner(db, current_user.id)


def update_task(
    db: Session,
    task_id: int,
    payload: schemas.TaskUpdate,
    current_user: models.User,
) -> models.Task:
    task = repositories.get_task(db, task_id)
    if task is None:
        raise exceptions.TaskNotFoundError
    if task.owner_id != current_user.id:
        raise exceptions.ForbiddenError

    for field, value in payload.model_dump(exclude_unset=True).items():
        if value is not None:
            setattr(task, field, value)
    return repositories.save_task(db, task)


def delete_task(
    db: Session,
    task_id: int,
    current_user: models.User,
) -> None:
    task = repositories.get_task(db, task_id)
    if task is None:
        raise exceptions.TaskNotFoundError
    if task.owner_id != current_user.id:
        raise exceptions.ForbiddenError
    repositories.delete_task(db, task)

What should I compare?

Check that update_task() applies its ownership guard before changing fields. Check that delete_task() applies the same guard before calling the repository.

• Return to app/routers/tasks.py.
• Replace the complete router-level delete_task() function with this version:

@router.delete("/{task_id}")
def delete_task(
    task_id: int,
    db: DBSession,
    current_user: CurrentUser,
):
    try:
        services.delete_task(db, task_id, current_user)
        return {"ok": True}
    except exceptions.TaskNotFoundError:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found",
        )
    except exceptions.ForbiddenError:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="You do not own this task",
        )

What does this route do?

A permitted deletion returns a small confirmation object. A missing task becomes a 404 response.

A task owned by someone else becomes a 403 response. Both mutation routes now translate the same domain error consistently.

• Save app/routers/tasks.py.
• Confirm the delete route catches ForbiddenError after the missing-task handler.

✔️ Awesome, I've got everything!

Your task router now translates forbidden updates and forbidden deletions into matching 403 responses.

ⓧ I'd like to double check the full code

from fastapi import APIRouter, HTTPException, status

from app import exceptions, schemas, services
from app.database import DBSession
from app.dependencies import CurrentUser

router = APIRouter(prefix="/tasks", tags=["tasks"])


@router.post(
    "",
    response_model=schemas.TaskPublic,
    status_code=status.HTTP_201_CREATED,
)
def create_task(
    payload: schemas.TaskCreate,
    db: DBSession,
    current_user: CurrentUser,
):
    return services.create_task(db, payload, current_user)


@router.get("", response_model=list[schemas.TaskPublic])
def list_tasks(db: DBSession, current_user: CurrentUser):
    return services.list_own_tasks(db, current_user)


@router.patch("/{task_id}", response_model=schemas.TaskPublic)
def update_task(
    task_id: int,
    payload: schemas.TaskUpdate,
    db: DBSession,
    current_user: CurrentUser,
):
    try:
        return services.update_task(db, task_id, payload, current_user)
    except exceptions.TaskNotFoundError:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found",
        )
    except exceptions.ForbiddenError:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="You do not own this task",
        )


@router.delete("/{task_id}")
def delete_task(
    task_id: int,
    db: DBSession,
    current_user: CurrentUser,
):
    try:
        services.delete_task(db, task_id, current_user)
        return {"ok": True}
    except exceptions.TaskNotFoundError:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found",
        )
    except exceptions.ForbiddenError:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="You do not own this task",
        )

What should I compare?

Both mutation routes pass CurrentUser into the service. Both routes translate ForbiddenError into status.HTTP_403_FORBIDDEN.

• Switch back to the Command Prompt terminal.
• Stop Uvicorn with Ctrl+C.
• Reload both ownership rules by running:

uvicorn main:app

What does this command confirm?

Uvicorn imports the updated service and router modules. A successful startup confirms the edited Python files have valid syntax.

Contrast authentication and authorization responses

The final check follows one task through three security states. You will remove identity, use the wrong owner, then use the correct owner.

Before you send the next requests, where do you expect the boundary between identity and permission to appear?

• Return to Swagger UI.
• Use the Authorize control to clear Bob's bearer authorization.
• Expand PATCH /tasks/{task_id}.
• Enter alice's task ID as the path value.
• Replace the example title value with Identity check.
• Send the unauthenticated request.

You'll receive a 401 response because CurrentUser cannot resolve an identity. The request stops before the ownership rule runs.

• Select the Authorize control.
• Enter bob as the username.
• Enter Bob's existing password.
• Submit the sign-in form.
• Send the same PATCH request again.

You'll receive 403 Forbidden because Bob has a valid identity without ownership permission.

• Expand DELETE /tasks/{task_id}.
• Enter alice's task ID as the path value.
• Send the delete request as Bob.

You'll receive another 403 Forbidden. The same service-layer policy now protects both ways a task can be changed.

• Use the Authorize control to clear Bob's authorization.
• Select the Authorize control again.
• Enter alice as the username.
• Enter Alice's existing password.
• Submit the sign-in form.

Swagger UI now sends Alice's token with protected requests. The API can compare Alice's user ID with the task's owner ID.

• Expand PATCH /tasks/{task_id}.
• Enter alice's task ID as the path value.
• Replace the example title value with Alice owns this task.
• Send the request as Alice.

You'll receive the updated task record with Alice owns this task as its title. The matching owner ID lets the service save the change.

• Expand DELETE /tasks/{task_id}.
• Enter alice's task ID as the path value.
• Send the delete request as Alice.

You'll receive a response confirming the deletion. Alice can manage her own persistent task from creation through removal.

What did the three responses prove?

• The 401 response shows that the authentication dependency rejected a request without valid identity.
• The 403 response shows that the service rejected a valid user without ownership permission.
• Alice's successful response shows that the same service allowed the matching owner.

Seeing the wrong response?

Confirm Swagger UI is authorized with the intended account before each request. Check that every path value uses Alice's original task ID.

If Alice also receives 403, compare the task's owner_id with the id returned for Alice during registration.

Help me trace the 401 and 403 responses.

You closed the security gap that Bob exposed. Your API now proves identity before enforcing owner-only task changes.

Secret mission

Add an Administrator Policy

Ownership keeps each member inside their own task list. Trusted administrators sometimes need broader access to support the whole team. In this mission, you will add an administrator policy that can inspect every task and manage tasks across users.

Clean Up Your Resources

Clean Up Your Resources

Your Secure Team Task API runs entirely on your Windows computer. Choose whether to keep the local resources, pause the server for later, or delete everything with no ongoing cost to consider.

Resources you used:

• Local files: The secure-task-api project folder containing your FastAPI source files plus the tutorial-env Python virtual environment.
• Local data: The task_api.db SQLite database containing your users plus tasks.
• Running process: The local Uvicorn server process.

Keep everything running

No action is needed. Choose this option if you are still testing authentication, ownership rules, or administrator access.

• Keep the Uvicorn server running to continue testing the API.
• Leave the secure-task-api folder on your computer.
• Leave task_api.db in place to retain every user plus task.
• Continue experimenting without any recurring charges.

Pause - I'll come back to this later

Stop the local server to release the terminal. Your project files stay ready for another session.

• Switch back to the Command Prompt terminal running Uvicorn.
• Press Ctrl+C to stop the server.
• Confirm that the command line is ready for input again.
• Leave the secure-task-api folder on your computer.

The returned command line confirms that Uvicorn has stopped. Your code plus saved records remain available without creating charges.

Delete - I don't want to use this again

Deleting local work is permanent, so this option deserves a careful choice. You can reset only the saved data or remove the complete project.

Reset only the database

Use this option when you want fresh users plus tasks while keeping your source code and installed packages.

• Switch back to the Command Prompt terminal running Uvicorn.
• Press Ctrl+C to stop the server.
• Return to Visual Studio Code.
• Select task_api.db in the file tree.
• Press the Delete key.
• Approve the confirmation dialog if Visual Studio Code displays one.
• Confirm that task_api.db no longer appears in the file tree.

Your source files plus virtual environment remain in place. The application creates a new empty database the next time the server starts.

Delete the complete project

Full removal deletes the source code, virtual environment, installed project packages, and local database.

• Switch back to the Command Prompt terminal running Uvicorn.
• Press Ctrl+C to stop the server.
• Close Visual Studio Code.
• Press the Windows key to open system search.
• Type secure-task-api into the search field.
• Select the secure-task-api folder in the results.
• Press Shift+Delete to remove the folder permanently.
• Confirm the permanent deletion in the dialog.

Search for secure-task-api again. You should no longer see the folder in the results.

Nice Work!

Nice Work!

You did it! Your Secure Team Task API now runs as a layered FastAPI backend with persistent SQLite storage.

You've learned how to:

• Built a FastAPI task API with complete CRUD routes. Exercised each route through Swagger UI.
• Persisted user records plus task records in SQLite. Separated Pydantic API schemas from SQLAlchemy database models. Organized each application responsibility in a focused module.
• Stored password hashes for every account. Issued JWT bearer tokens for authenticated requests. Enforced owner-only authorization for task changes.
• Secret Mission: Added an administrator policy for GET /tasks/admin/all. Enabled administrators to manage cross-user tasks. Kept members owner-scoped with visible 403 responses.

Ready to quiz yourself?