Build a Priority Task Queue API

Build a tested FastAPI task API with stable heap-based priority scheduling.

Introduction

30 Second Summary

A list feels fair when everyone waits their turn. The same rule becomes risky when an urgent job arrives behind routine work.

In this project, you will build a tested local task-processing API with FastAPI that processes urgent work first while preserving arrival order for equal priorities. A deliberate FIFO failure will show why a heap-based priority queue fits that scheduling rule.

What You'll Build

You'll demo a browser-driven task service where a later urgent task moves ahead of routine work while equal priorities stay in arrival order.

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

  • A browser-ready task API that turns a submitted title plus priority into a visible JSON task.
  • A priority scheduler that returns a later urgent task before older routine work.
  • An automated test suite that reports 4 passed for the API's required behavior.
  • Secret Mission: Add a read-only queue preview that shows processing order without removing tasks.

Are there any prerequisites?

You only need basic Python knowledge covering functions, classes, lists, and dictionaries. Step 1 guides you through every project-specific package.

Before We Start

This step is your moment to commit to the local Priority Task Queue API you are about to build. A clear purpose will guide every design choice that follows.

Set Up the Python API Workspace

Your local API depends on a predictable Python toolchain. System packages or changing dependency releases can make identical code behave differently.

An isolated virtual environment keeps this project separate from the rest of your computer. Exact package pins make every later test reproducible.

In this step, get ready to:
  • Confirm that Python 3.10 or newer is available.
  • Create the project workspace with an activated virtual environment.
  • Install a verified dependency set with FastAPI, Pydantic, plus pytest.
Confirm Python and create the workspace

Every pinned package in this project requires Python 3.10 or newer. Checking the interpreter first prevents compatibility errors during installation.

  • Use your Linux application launcher to open Terminal.
  • Check your available Python version by running this command:
python3 --version

What does this command check?

The command asks the system Python interpreter to report its version. That result determines whether it can run the pinned packages used by this project.

✔️ I see version 3.10 or higher

Your interpreter meets the project requirement. That compatibility check removes the first major source of setup errors.

ⓧ I see an older version

The available interpreter is too old for the pinned dependencies. Your virtual environment needs Python 3.10 or newer.

  • Install Python 3.10 or newer using the documented package process for your Linux distribution.
  • Return to the terminal after the installation finishes.
  • Repeat the version check above.
  • Continue when the reported version is 3.10 or newer.

ⓧ Command not found

Your shell cannot find a Python 3 interpreter. Install Python 3.10 or newer before creating the environment.

  • Install Python using the documented package process for your Linux distribution.
  • Return to the terminal after the installation finishes.
  • Repeat the version check above.
  • Continue when the shell reports Python 3.10 or newer.
  • Move to your Desktop by running this command:
cd ~/Desktop

What does this command do?

The command changes the terminal location to your Desktop. This gives the new project a predictable home that is easy to find.

  • Create the priority-task-api folder by running these commands:
mkdir priority-task-api
cd priority-task-api

What do these commands do?

  • The first command creates the priority-task-api folder on your Desktop.
  • The second command moves the terminal into that folder.

Your terminal is now working inside the new project folder.

  • Create the project environment by running this command:
python3 -m venv .venv

What does this command do?

The command creates an isolated Python environment inside .venv. Packages installed there stay separate from your system installation.

  • Activate the new environment by running this command:
source .venv/bin/activate

What does activation change?

Activation points the terminal at the environment's Python interpreter. Future package installations now belong to this project.

The environment name now appears at the start of your shell prompt. This confirms that the project environment is active.

The next command opens the current folder as a VS Code workspace. VS Code may ask whether you trust the folder because workspace files can run commands.

  • Choose Yes, I trust the authors if the workspace trust prompt appears.
  • Open the current priority-task-api folder in VS Code by running this command:
code .

What does this command do?

The period represents the terminal's current folder. VS Code opens priority-task-api as the workspace for this project.

You should see priority-task-api at the top of the VS Code Explorer. Your Linux terminal remains inside the same folder with the environment active.

VS Code did not open?

  • Confirm that VS Code is installed before retrying the command.
  • Use File followed by Open Folder... in VS Code if the terminal cannot find its command.
  • Select ~/Desktop/priority-task-api as the folder.

Ask for targeted help with opening the project folder in VS Code. If your Linux installation behaves differently, share the error in the NextWork community.

Pin and install the dependencies

A requirements file records the exact package releases that belong to the project. Installing from that file gives you the same tested dependency set each time.

  • Select New File... in the VS Code Explorer.
  • Enter requirements.txt as the file name.
  • Press Enter to create the file.
  • Add the exact dependency pins by pasting this content:
fastapi[standard]==0.162.0
pydantic==2.14.3
pytest==9.2.0

What do these pins install?

  • The fastapi[standard] package supplies the web framework plus the command-line and testing support used by this project.
  • The pydantic package validates task titles and priority values.
  • The pytest package runs the automated tests you add later.
  • Save requirements.txt with Ctrl+S.
  • Confirm that requirements.txt appears under priority-task-api in the Explorer.

Does the file look different?

  • Check that the file is named requirements.txt with no extra file extension.
  • Compare every package name and version with the reference below.
  • Remove any blank spaces from the package lines.

Use this prompt for help with checking the requirements file.

✔️ Awesome, I've got everything!

Great. Double-check that requirements.txt is saved before installing from it.

ⓧ I'd like to double check the full code

fastapi[standard]==0.162.0
pydantic==2.14.3
pytest==9.2.0
  • Switch back to the Linux terminal from earlier.
  • Install the pinned dependencies into the active environment by running this command:
python -m pip install -r requirements.txt

What does this command do?

The active environment's Python interpreter runs pip against requirements.txt. Each exact pin prevents a future release from changing the project underneath you.

The terminal downloads the packages before returning to the prompt without an installation error. Your isolated dependency set is now ready for the API.

Did the installation fail?

  • Confirm that the environment name still appears at the start of your terminal prompt.
  • Reactivate .venv from the priority-task-api terminal if the environment is inactive.
  • Repeat the installation command after confirming your internet connection.

Ask for help with debugging the dependency installation. You can also share the non-sensitive error output in the NextWork community.

Verify the toolchain

A successful installation is only useful when the terminal is using the expected interpreter and releases. These final checks prove that the active environment contains the complete pinned toolchain.

Before you run these checks, which version should each command report?

  • Verify the active Python interpreter and installed packages by running these commands:
python --version
python -c "import fastapi, pydantic; print(fastapi.__version__, pydantic.__version__)"
pytest --version

What do these checks prove?

  • The first command confirms which Python interpreter the active environment uses.
  • The second command imports both API packages before printing their installed versions.
  • The third command confirms that the test runner is installed and callable.

You should see Python 3.10 or newer.

The package check prints 0.162.0 2.14.3. The last command reports pytest 9.2.0.

That confirms the workspace is reproducible. Your project now uses the intended interpreter plus every tested package release.

Seeing unexpected versions?

  • Confirm that the environment name appears at the start of your terminal prompt.
  • Return to ~/Desktop/priority-task-api before reactivating the environment.
  • Repeat the dependency installation if an import is missing.

Use this prompt for diagnosing a version mismatch.

Your Python workspace is isolated and verified. Next, you will turn it into a working FIFO task API that you can call from the browser.

Build the FIFO Task API

Your isolated Python workspace is ready. It now needs a service that can accept work from a browser.

FastAPI turns typed Python functions into HTTP endpoints. This first version stores work in a FIFO queue so its scheduling rule stays easy to see.

In this step, get ready to:
  • Define validated task models.
  • Build the FIFO endpoints.
  • Confirm the API through interactive browser documentation.
Define the task models

An API needs a precise shape for the data it accepts. Pydantic models define that shape while checking each incoming value.

Why FastAPI for this API?

FastAPI uses type declarations to validate request data. It also generates OpenAPI-based interactive documentation.

Flask can serve APIs too. Matching this validation flow requires more manual wiring.

  • Create main.py inside the open priority-task-api folder using the file sidebar in VS Code.
  • Add the task models to main.py by pasting this code:
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field

class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=80)
    priority: int = Field(ge=1, le=5)

class Task(TaskCreate):
    id: int

What do these models define?

  • The FastAPI imports prepare the application for routes plus API errors. The Pydantic imports provide typed models with field constraints.
  • TaskCreate accepts titles from 1 to 80 characters. It accepts priorities from 1 to 5.
  • Task inherits those validated fields. It adds the integer ID returned by the API.
  • Save main.py.
  • Check main.py for red syntax underlines.

You should see no syntax warnings in the model definitions.

Seeing import or syntax warnings?

Confirm that VS Code is using the Python interpreter from .venv. Compare every bracket with the code block if the warning points inside a model.

Help me fix the warnings in my FastAPI model code.

The application object connects your Python code to the development server. The temporary list keeps submitted tasks in memory while an integer tracks the next ID.

  • Add the application state plus the root endpoint below the Task model by pasting this code:
app = FastAPI(title="Priority Task Queue API")
tasks: list[Task] = []
next_id = 1

@app.get("/")
def root() -> dict[str, str]:
    return {"message": "Priority Task Queue API is running"}

What does this code do?

  • app creates the FastAPI service with a title for its documentation.
  • tasks holds submitted task objects in insertion order.
  • next_id starts the ID sequence at 1.
  • root() provides a small response that proves the service is running.
  • Save main.py.

Before you start the server, what response do you expect from the root address? Keep that prediction in mind.

  • Start the development server from the activated terminal by running this command:
fastapi dev

What does this command do?

The command detects main.py before starting a local development server. It keeps the terminal session occupied while the service runs.

  • Enter http://127.0.0.1:8000 in your browser address bar.

You should see the message Priority Task Queue API is running. Your first live API response is working.

Page not loading?

Confirm that the development server is still running in the Linux terminal. Check that the address uses port 8000.

Help me debug my local FastAPI server.

Build the FIFO endpoints

FIFO processes the oldest queued item first. Appending each new task to a list preserves that insertion order.

  • Add the task creation endpoint below root() by pasting this code:
@app.post("/tasks", response_model=Task, status_code=status.HTTP_201_CREATED)
def create_task(task: TaskCreate) -> Task:
    global next_id
    created_task = Task(id=next_id, **task.model_dump())
    tasks.append(created_task)
    next_id += 1
    return created_task

How does task creation work?

  • POST /tasks accepts data shaped like TaskCreate.
  • model_dump() turns the validated request into values for a new Task.
  • tasks.append() places the new task at the end of the list.
  • next_id increases after each successful creation.
  • Save main.py.
  • Switch back to the browser tab from earlier.
  • Enter http://127.0.0.1:8000/docs in the address bar.

You should see POST /tasks in the generated documentation. This proves the server reloaded your new endpoint.

Task endpoint missing?

Confirm that you saved main.py. Check that the @app.post("/tasks", response_model=Task, status_code=status.HTTP_201_CREATED) decorator sits directly above create_task().

Help me find why POST /tasks is missing from my FastAPI docs.

Processing needs one more endpoint. Removing index 0 returns the oldest task from the list.

  • Add the processing endpoint below create_task() by pasting this code:
@app.post("/tasks/next", response_model=Task)
def process_next_task() -> Task:
    if not tasks:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="No tasks queued",
        )
    return tasks.pop(0)

How does FIFO processing work?

  • process_next_task() checks whether the in-memory list contains a task.
  • HTTPException returns a 404 response when the queue is empty.
  • tasks.pop(0) removes the first submitted task from the list.
  • Save main.py.
  • Refresh the interactive documentation in your browser.

You should now see both POST /tasks and POST /tasks/next.

Processing endpoint missing?

Check that process_next_task() is aligned with create_task(). Confirm that its decorator uses the exact path /tasks/next.

Help me fix my POST /tasks/next endpoint.

✔️ Awesome, I've got everything!

Great. Save main.py once more before testing the complete FIFO flow.

ⓧ I'd like to double check the full code

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field

class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=80)
    priority: int = Field(ge=1, le=5)

class Task(TaskCreate):
    id: int

app = FastAPI(title="Priority Task Queue API")
tasks: list[Task] = []
next_id = 1

@app.get("/")
def root() -> dict[str, str]:
    return {"message": "Priority Task Queue API is running"}

@app.post("/tasks", response_model=Task, status_code=status.HTTP_201_CREATED)
def create_task(task: TaskCreate) -> Task:
    global next_id
    created_task = Task(id=next_id, **task.model_dump())
    tasks.append(created_task)
    next_id += 1
    return created_task

@app.post("/tasks/next", response_model=Task)
def process_next_task() -> Task:
    if not tasks:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="No tasks queued",
        )
    return tasks.pop(0)

How to use this reference

Compare this reference with your saved main.py from top to bottom.

Create and process a task

Swagger UI lets you call each endpoint from the browser. You can now watch one task enter the list before FIFO processing removes it.

  • Switch back to the browser documentation at http://127.0.0.1:8000/docs.
  • Expand POST /tasks.
  • Click Try it out.
  • Replace the example title with Write weekly report.
  • Replace the example priority with 1.
  • Click Execute.

You should see response code 201. The response body should show ID 1 with your task title plus priority.

Before you process the task, which title do you expect the queue to return? Hold that answer while you run the request.

  • Expand POST /tasks/next.
  • Click Try it out.
  • Click Execute.

You should see response code 200. The response body should return Write weekly report as the next task.

That is the full request cycle working. Your API accepted a validated task before returning the same task from its FIFO queue.

Request not returning the task?

Create the task before calling POST /tasks/next. Confirm that the creation response contains the title you entered.

Help me debug the FIFO request flow in my FastAPI docs.

  • Return to the Linux terminal from earlier.
  • Stop the development server by pressing Ctrl+C.

You should see your terminal prompt return. The local server is now stopped.

Your FIFO API can accept validated work through a browser. Next, you will write an automated test that reveals why submission order fails when a newer task is more urgent.

Expose the FIFO Scheduling Bug

Your FastAPI service already accepts tasks. The browser check proved it can process one task from start to finish.

A FIFO queue processes the oldest item first. This step uses an automated test to capture the requirement that urgency wins when arrival order conflicts with priority.

In this step, get ready to:
  • Create a test that records the required priority order.
  • Model a routine task arriving before an urgent task.
  • Run the test suite against the current FIFO scheduler.
Write the priority-order test

An automated test turns the scheduling requirement into an executable example. pytest runs the example against the same FastAPI app that you tested in the browser.

  • Switch back to VS Code from your earlier API work.
  • Use the new-file icon in the file sidebar to create test_main.py beside main.py.
  • Add the priority-order test by pasting this code into test_main.py:
from fastapi.testclient import TestClient

from main import app, tasks

client = TestClient(app)

def test_higher_priority_processed_first():
    tasks.clear()
    client.post("/tasks", json={"title": "Write weekly report", "priority": 1})
    client.post("/tasks", json={"title": "Fix production outage", "priority": 5})

    response = client.post("/tasks/next")

    assert response.status_code == 200
    assert response.json()["title"] == "Fix production outage"

What does this test prove?

  • The TestClient sends requests directly to the existing app.
  • The tasks.clear() call gives the test an empty queue.
  • The first request queues Write weekly report with priority 1.
  • The second request queues Fix production outage with priority 5.
  • The response.status_code assertion confirms that processing succeeds.
  • The response.json()["title"] assertion records which task should be processed first.
  • Save test_main.py.
  • Confirm test_main.py appears beside main.py in the file sidebar.

Can't Find the Test File?

  • Confirm test_main.py sits inside priority-task-api beside main.py.
  • Save the file before checking the file sidebar.

Help me check why pytest cannot find my test file.

✔️ Awesome, I've got everything!

Good work. Your test now records the priority behavior that the API needs to provide.

ⓧ I'd like to double check the full code

from fastapi.testclient import TestClient

from main import app, tasks

client = TestClient(app)

def test_higher_priority_processed_first():
    tasks.clear()
    client.post("/tasks", json={"title": "Write weekly report", "priority": 1})
    client.post("/tasks", json={"title": "Fix production outage", "priority": 5})

    response = client.post("/tasks/next")

    assert response.status_code == 200
    assert response.json()["title"] == "Fix production outage"

What Should Match?

This reference shows the complete test_main.py file for this step. The imports, task titles, priorities, request paths, and assertions should match exactly.

Run the test against FIFO

The current service still removes the first item with pop(0). The test now checks whether that behavior satisfies the priority requirement.

Before you run the test, which task title do you think the current scheduler returns?

  • Test the current FIFO behavior by running this command:
pytest

What Does This Command Do?

The command discovers the test in test_main.py. It executes each request against the imported app.

You'll see one failed test. The current response contains Write weekly report.

The assertion expects Fix production outage. The pop(0) operation chooses the oldest item.

You found the exact scheduling gap. The failing test now protects the urgent-first behavior that users need.

Test Not Running?

  • Confirm your terminal is inside the priority-task-api folder.
  • Confirm the terminal prompt still shows the active .venv environment.
  • Check that the filename is exactly test_main.py.
  • Check that the test function is named test_higher_priority_processed_first().

Help me debug why pytest is not running the intended failing test.

  • Leave test_main.py unchanged because it records the required behavior.
  • Keep main.py on its current FIFO implementation.

That is the scheduling bug pinned down. Next, you'll replace the FIFO list with a priority heap while preserving arrival order for ties.

Replace FIFO with a Priority Heap

Your failing FastAPI test now proves that the FIFO scheduler sends older work ahead of urgent work. That test gives this refactor a precise target.

You will move scheduling into a priority queue built with a heap. A sequence counter will keep equal-priority tasks in submission order.

In this step, get ready to:
  • Build a heap-based task scheduler.
  • Connect the existing API endpoints to the scheduler.
  • Expand the automated tests to cover ordering plus validation.
Create the priority queue

Python's heapq module removes the smallest entry first. Storing each priority as a negative number makes priority 5 sort ahead of priority 1.

Each heap entry also contains a unique sequence number. That number preserves FIFO behavior when two tasks have the same priority.

  • Use the new-file control in the VS Code Explorer sidebar to create scheduler.py inside priority-task-api.
  • Add the scheduler imports plus its shared types by pasting this code:
import heapq
from itertools import count

TaskData = dict[str, int | str]
HeapEntry = tuple[int, int, TaskData]

What does this code set up?

  • The heapq module maintains the task ordering rule.
  • The count iterator supplies increasing sequence values plus task IDs.
  • The TaskData and HeapEntry aliases describe the dictionaries plus tuples stored by the scheduler.
  • Save scheduler.py.
  • Confirm that scheduler.py is listed beside main.py in the Explorer sidebar.

Can't find scheduler.py?

Check that the file sits directly inside priority-task-api. A nested copy will not match the import used by main.py.

Ask for help if the file appears in the wrong folder. Help me place scheduler.py beside main.py in my priority-task-api folder.

The PriorityTaskQueue class owns the heap plus both counters. Its methods give the API a small interface for adding tasks or removing the correct next task.

  • Add the complete queue class below HeapEntry by pasting this code:
class PriorityTaskQueue:
    def __init__(self) -> None:
        self._heap: list[HeapEntry] = []
        self._sequence = count()
        self._next_id = count(1)

    def push(self, title: str, priority: int) -> TaskData:
        task: TaskData = {
            "id": next(self._next_id),
            "title": title,
            "priority": priority,
        }
        entry = (-priority, next(self._sequence), task)
        heapq.heappush(self._heap, entry)
        return task

    def pop(self) -> TaskData | None:
        if not self._heap:
            return None
        return heapq.heappop(self._heap)[2]

    def clear(self) -> None:
        self._heap.clear()
        self._sequence = count()
        self._next_id = count(1)

How does the queue work?

  • The push() method creates a task with an increasing ID.
  • Each heap entry stores -priority first. This puts larger priorities at the front.
  • The sequence value breaks priority ties according to submission order.
  • The pop() method returns None for an empty queue. Otherwise it removes the first heap entry.
  • The clear() method resets the state so each test starts with a fresh queue.
  • Save scheduler.py.

Before you run the test suite, do you expect the existing FIFO failure to change while main.py still uses tasks?

  • Confirm that the new module has not changed the current API behavior by running:
pytest

What should I see?

The test suite should still report the same assertion failure. This confirms that the scheduler exists without changing the API before you connect it.

Seeing a scheduler error?

Check the indentation inside PriorityTaskQueue. Each method must align at the same class level.

Check that scheduler.py contains both imports above the type aliases. Help me debug scheduler.py using the traceback from pytest.

✔️ Awesome, I've got everything!

Your scheduler now contains the complete heap implementation.

ⓧ I'd like to double check the full code

import heapq
from itertools import count

TaskData = dict[str, int | str]
HeapEntry = tuple[int, int, TaskData]

class PriorityTaskQueue:
    def __init__(self) -> None:
        self._heap: list[HeapEntry] = []
        self._sequence = count()
        self._next_id = count(1)

    def push(self, title: str, priority: int) -> TaskData:
        task: TaskData = {
            "id": next(self._next_id),
            "title": title,
            "priority": priority,
        }
        entry = (-priority, next(self._sequence), task)
        heapq.heappush(self._heap, entry)
        return task

    def pop(self) -> TaskData | None:
        if not self._heap:
            return None
        return heapq.heappop(self._heap)[2]

    def clear(self) -> None:
        self._heap.clear()
        self._sequence = count()
        self._next_id = count(1)
Connect the API to the heap

The request models plus endpoints stay in main.py. The scheduler now owns task IDs plus processing order.

This separation keeps the HTTP layer focused on requests and responses. The data structure can change without changing the API paths.

  • Switch back to main.py from earlier.
  • Select the existing contents of main.py.
  • Replace the selected contents by pasting this code:
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field

from scheduler import PriorityTaskQueue

class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=80)
    priority: int = Field(ge=1, le=5)

class Task(TaskCreate):
    id: int

app = FastAPI(title="Priority Task Queue API")
queue = PriorityTaskQueue()

@app.get("/")
def root() -> dict[str, str]:
    return {"message": "Priority Task Queue API is running"}

@app.post("/tasks", response_model=Task, status_code=status.HTTP_201_CREATED)
def create_task(task: TaskCreate) -> Task:
    return Task(**queue.push(task.title, task.priority))

@app.post("/tasks/next", response_model=Task)
def process_next_task() -> Task:
    task = queue.pop()
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="No tasks queued",
        )
    return Task(**task)

What changed in the API?

  • The API imports PriorityTaskQueue from the new scheduler module.
  • The queue instance replaces the FIFO tasks list plus the separate ID counter.
  • The creation endpoint passes validated values into queue.push().
  • The processing endpoint receives the next task from queue.pop().
  • The existing Pydantic models still control input validation plus response serialization.
  • Save main.py.
  • Confirm that the import from scheduler has no editor diagnostic.

Is PriorityTaskQueue unresolved?

Confirm that main.py and scheduler.py sit in the same priority-task-api folder. Check that the class name is spelled PriorityTaskQueue in both files.

Ask for help if the import remains unresolved. Help me debug the PriorityTaskQueue import in main.py.

✔️ Awesome, I've got everything!

Your API now delegates task ordering to the priority queue.

ⓧ I'd like to double check the full code

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field

from scheduler import PriorityTaskQueue

class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=80)
    priority: int = Field(ge=1, le=5)

class Task(TaskCreate):
    id: int

app = FastAPI(title="Priority Task Queue API")
queue = PriorityTaskQueue()

@app.get("/")
def root() -> dict[str, str]:
    return {"message": "Priority Task Queue API is running"}

@app.post("/tasks", response_model=Task, status_code=status.HTTP_201_CREATED)
def create_task(task: TaskCreate) -> Task:
    return Task(**queue.push(task.title, task.priority))

@app.post("/tasks/next", response_model=Task)
def process_next_task() -> Task:
    task = queue.pop()
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="No tasks queued",
        )
    return Task(**task)
Expand the test suite

The original test already captures the priority requirement. A second test now protects the sequence counter by checking equal-priority FIFO behavior.

  • Switch back to test_main.py from earlier.
  • Select the existing contents of test_main.py.
  • Replace the selected contents with the imports plus ordering tests below:
from fastapi.testclient import TestClient

from main import app, queue

client = TestClient(app)

def test_higher_priority_processed_first():
    queue.clear()
    client.post("/tasks", json={"title": "Write weekly report", "priority": 1})
    client.post("/tasks", json={"title": "Fix production outage", "priority": 5})

    response = client.post("/tasks/next")

    assert response.status_code == 200
    assert response.json()["title"] == "Fix production outage"

def test_equal_priority_preserves_fifo_order():
    queue.clear()
    client.post("/tasks", json={"title": "First task", "priority": 3})
    client.post("/tasks", json={"title": "Second task", "priority": 3})

    response = client.post("/tasks/next")

    assert response.json()["title"] == "First task"

What do these tests prove?

  • The first test keeps the requirement that exposed the FIFO bug.
  • The second test gives two tasks the same priority.
  • Calling queue.clear() prevents one test from leaving tasks or IDs behind for another test.
  • Save test_main.py.

Before you run the suite, how many passing tests do you expect from the file you have now?

  • Check both ordering rules by running:
pytest

What should I see?

Both ordering tests should pass. That's the original scheduling bug repaired plus equal-priority behavior protected.

Is an ordering test failing?

Check that each heap entry uses -priority before the sequence value. Reversing those fields changes the scheduling rule.

Check that both tests call queue.clear() before posting tasks. Help me debug a failing priority-order test.

Ordering is only part of the API contract. The remaining tests protect the empty-queue response plus the existing priority range.

  • Add the empty-queue plus invalid-priority tests below the existing tests:
def test_empty_queue_returns_404():
    queue.clear()

    response = client.post("/tasks/next")

    assert response.status_code == 404
    assert response.json() == {"detail": "No tasks queued"}

def test_invalid_priority_returns_validation_error():
    queue.clear()

    response = client.post(
        "/tasks",
        json={"title": "Impossible priority", "priority": 6},
    )

    assert response.status_code == 422

What do the final tests protect?

  • The empty-queue test confirms that processing returns a 404 response when no task is available.
  • The validation test submits priority 6.
  • The expected 422 response proves that the supported range remains 1 through 5.
  • Save test_main.py.

Before you run the final check, do you expect every test to pass now that priority ordering plus tie ordering are both implemented?

  • Verify the complete API behavior by running:
pytest

What should I see?

You should see 4 passed. That result proves the API processes urgent work first while preserving its other response rules.

Not seeing four passing tests?

Read the first failing test name before changing code. An ordering failure points toward scheduler.py. A response failure points toward main.py.

Compare the failing assertion with the matching full-file reference below. Help me diagnose why pytest is not reporting four passing tests.

✔️ Awesome, I've got everything!

Your complete test suite now covers priority ordering plus the API's edge cases.

ⓧ I'd like to double check the full code

from fastapi.testclient import TestClient

from main import app, queue

client = TestClient(app)

def test_higher_priority_processed_first():
    queue.clear()
    client.post("/tasks", json={"title": "Write weekly report", "priority": 1})
    client.post("/tasks", json={"title": "Fix production outage", "priority": 5})

    response = client.post("/tasks/next")

    assert response.status_code == 200
    assert response.json()["title"] == "Fix production outage"

def test_equal_priority_preserves_fifo_order():
    queue.clear()
    client.post("/tasks", json={"title": "First task", "priority": 3})
    client.post("/tasks", json={"title": "Second task", "priority": 3})

    response = client.post("/tasks/next")

    assert response.json()["title"] == "First task"

def test_empty_queue_returns_404():
    queue.clear()

    response = client.post("/tasks/next")

    assert response.status_code == 404
    assert response.json() == {"detail": "No tasks queued"}

def test_invalid_priority_returns_validation_error():
    queue.clear()

    response = client.post(
        "/tasks",
        json={"title": "Impossible priority", "priority": 6},
    )

    assert response.status_code == 422

Why does the heap fit this scheduler?

Removing index 0 from a list shifts every later item forward. That work grows with the number of queued tasks.

Heap insertion plus removal restore ordering along a tree path. This trade-off matches a scheduler that repeatedly adds work before selecting the most urgent task.

  • Explain aloud why pop(0) shifts later list items.
  • Explain aloud how the heap restores its ordering along a tree path.
  • Explain aloud why the sequence counter preserves submission order for equal priorities.

Secret mission

Preview the Queue Without Changing It

Operators need to see which tasks are waiting before they process the next one. Build a read-only queue preview that reveals the processing order without consuming any tasks.

Clean Up Your Resources

Clean Up Your Resources

Choose whether to keep your local project ready, pause your work, or delete it entirely. Everything runs on your Linux computer, so keeping the project creates no ongoing service charges.

Resources you used:

  • Your local priority-task-api folder stores the complete project. Its source files are main.py, scheduler.py, test_main.py, plus requirements.txt. Its .venv contains the pinned project packages.

Keep everything running

No action is needed. Choose this option if you want the tested API available for more experiments.

  • Keep the priority-task-api folder on your Linux computer.
  • Leave .venv inside the folder for future runs.

Pause - I'll come back to this later

The pause option ends your current shell session. Your source code and installed packages stay on disk.

  • Close the Linux terminal window that contains the activated .venv.
  • Leave the priority-task-api folder in place for your next session.

Delete - I don't want to use this again

Deletion permanently removes the local project. Your files stay safe unless you run the cleanup command below.

  • Switch back to the Linux terminal session with the active .venv.
  • Leave the priority-task-api folder before removing it by running these commands:
cd ..
rm -rf priority-task-api

What Does This Cleanup Do?

The first line moves your terminal into the directory that contains priority-task-api.

The second line permanently removes priority-task-api plus every file inside it. This includes the local .venv.

Once the prompt returns without an error, the folder has been removed. That is your local cleanup complete.

Nice Work!

Nice Work!

You made it! You built a tested FastAPI service that schedules work through a stable priority queue.

What you learned:

  • Built a typed JSON API with FastAPI. Used interactive browser documentation to submit tasks. Processed the next task through the same interface.
  • Used pytest to capture the FIFO urgency bug in a failing automated test before changing the implementation. Finished with five passing tests covering normal requests. Added edge-case coverage for rejected priorities.
  • Replaced the FIFO list with a heap-based scheduler that processes higher-priority tasks first. Preserved stable tie ordering with a sequence counter.
  • Secret Mission: Added a non-destructive queue preview that returns tasks in processing order. Proved the preview leaves the highest-priority task available for processing.

Ready to quiz yourself?