Build QueueForge Job API

Build a durable API with SQLite-backed jobs, workers, and burst-test metrics.

Introduction

30 Second Summary

Slow tasks can leave people staring at a loading screen with no proof their request was saved. A service restart makes that uncertainty worse when submitted work disappears.

In this project, you will build QueueForge, a REST API that turns number lists into trackable jobs. You will use asynchronous job processing to calculate each result while SQLite preserves every status across restarts.

What You'll Build

You watch a number list become a trackable job that survives an API restart before finishing with the correct sum.

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

  • A job submission flow with an immediate trackable ID for watching each request move from queued to completed.
  • A durability demo where a job remains available with its original payload after an API restart.
  • A public GitHub repository that presents your 20-job burst-test evidence plus documented scaling trade-offs.
  • Secret Mission: Run a one-worker versus two-worker benchmark to measure queue drain time against SQLite's scaling boundary.

Do I need anything before starting?

No backend experience or local development tools are required. The guide includes free personal account setup on GitHub.

Your workspace runs in GitHub Codespaces through the browser.

Before We Start

A clear goal gives every decision in QueueForge a purpose. This step is your commitment to proving your backend, persistence, reliability, and system-design skills before the hands-on work begins.

Launch Your Browser Workspace

Building QueueForge requires a consistent Python environment. Local setup varies by device, so this browser workspace removes that blocker.

A verified GitHub Free account gives you access to GitHub Codespaces. Codespaces runs the project through the Visual Studio Code web client in your browser.

The workspace includes Python for running the project. You will install FastAPI at version 0.162.0 from requirements.txt.

In this step, get ready to:
  • Create a verified GitHub Free account without adding a payment method.
  • Launch a browser-based Codespaces workspace.
  • Pin FastAPI to version 0.162.0 through requirements.txt.
Create and verify your GitHub account

Your GitHub account gives Codespaces somewhere to store your cloud workspace. Email verification confirms that you control the address attached to the account.

Keeping payment details unset protects your free-only budget. GitHub blocks Codespaces usage after the included allowance runs out when no valid payment method is on file.

  • Visit the GitHub signup page.
  • Enter your email address in the account form.
  • Create a unique password in the account form.
  • Choose a username for your account.
  • Complete the on-screen verification check if GitHub presents one.
  • Submit the completed account form.

GitHub sends a verification message to the email address you entered.

  • Open the verification message in your email inbox.
  • Select the verification link in the message.
  • Leave payment details unset on your GitHub Free account.

Your verified account is ready for the cloud workspace. You have also kept this project on the free-only path.

Launch the Codespaces workspace

The blank template starts with an empty directory inside GitHub's default development container. That container provides the terminal plus the runtime tools needed for QueueForge.

The first launch can take a few minutes while GitHub prepares the container. A quiet loading screen during that time is expected.

  • Visit the Your codespaces page.
  • Find the blank template under Explore quick start templates.
  • Select Use this template beneath the blank template.
  • Wait for the Visual Studio Code web client to finish loading.

You should see the Visual Studio Code web client with an integrated terminal at the bottom. This browser tab is now your development workspace.

  • Check the preinstalled Python version by running this command:
python -V

What does this command do?

The python command selects the workspace's Python runtime. The -V option prints its version so you can confirm that FastAPI supports it.

✔️ I see Python 3.10 or higher

Your workspace has Python 3.10 or newer. The runtime is ready for QueueForge.

ⓧ I see an older version

The blank template should provide a current Python runtime. An older result means this workspace did not load the expected default environment.

The workspace is still empty. Replacing it removes no QueueForge files.

  • Return to the Your codespaces page.
  • Open the menu for the current Codespace.
  • Select Delete.
  • Find the blank template under Explore quick start templates.
  • Select Use this template.
  • Wait for the replacement workspace to load.
  • Check the replacement runtime by running this command:
python -V

What does the replacement check prove?

The -V option reports the Python runtime in the replacement container. A result of 3.10 or higher confirms that the environment meets FastAPI's requirement.

ⓧ Command not found

The default Codespaces image includes Python. A missing command means this workspace did not load the expected runtime.

The workspace contains no QueueForge files yet. Replacing it is safe at this point.

  • Return to the Your codespaces page.
  • Open the menu for the current Codespace.
  • Select Delete.
  • Find the blank template under Explore quick start templates.
  • Select Use this template.
  • Wait for the replacement workspace to load.
  • Check the replacement runtime by running this command:
python -V

What does the replacement check prove?

The -V option asks the replacement container to report its Python runtime. A version of 3.10 or higher confirms that Python is available.

Your browser now contains the editor plus a working terminal. That is the complete development environment you need for QueueForge.

Install the pinned FastAPI version

A requirements file records the exact package version used by the project. This pin keeps your QueueForge environment repeatable.

  • Select the new-file icon at the top of the left file sidebar.
  • Enter requirements.txt as the file name.
  • Press Enter to create the file.

You should see an empty requirements.txt editor tab. The file is ready for its dependency pin.

  • Paste this exact dependency pin into requirements.txt:
fastapi[standard]==0.162.0

What does this dependency pin do?

  • The package name selects FastAPI.
  • The standard extra includes the FastAPI command-line tools used later.
  • The ==0.162.0 pin installs the exact project version.
  • Press Cmd+S (macOS) or Ctrl+S (Windows) to save requirements.txt.
  • Return to the terminal panel from earlier.
  • Install the project dependency by running this command:
python -m pip install -r requirements.txt

What does this installation command do?

The -r option tells pip to read the dependency list from requirements.txt. Pip then installs the pinned FastAPI package plus its standard dependencies.

The terminal downloads the required packages before returning to the prompt. A completed command without an installation error means the dependency is present.

Installation did not finish?

  • Save requirements.txt before retrying the installation.
  • Confirm that the dependency line contains version 0.162.0.
  • Run the installation command above again after correcting the file.

Ask for help with the terminal output using help me diagnose why FastAPI did not install from requirements.txt in my GitHub Codespace.

Before you run the final check, which FastAPI version do you expect the package metadata to report?

  • Verify the installed FastAPI version by running this command:
python -m pip show fastapi

What does the package check show?

Pip displays metadata for the installed FastAPI package. The version field lets you compare the installed package against the pin in requirements.txt.

You should see package metadata that includes Version: 0.162.0. Your browser workspace now has the exact FastAPI version QueueForge needs.

Seeing a different version?

  • Confirm that requirements.txt contains the 0.162.0 pin.
  • Save requirements.txt after correcting the version.
  • Run the installation command above again.

Ask for help using help me fix a FastAPI version mismatch in my GitHub Codespace.

✔️ Awesome, I've got everything!

Your requirements.txt file contains the exact FastAPI dependency pin.

ⓧ I'd like to double check the full code

  • Compare your complete requirements.txt file against this reference:
fastapi[standard]==0.162.0

What should match?

Your file should contain this single dependency line. The exact pin keeps every later QueueForge step on FastAPI version 0.162.0.

Your browser workspace is ready for development. Next up, you will build QueueForge's first in-memory API and expose the reliability problem it cannot solve.

Build the Volatile API

Your FastAPI installation is ready inside the browser workspace. QueueForge now needs a REST API that can accept jobs.

An in-memory dictionary gives you the fastest path to a working result. You will restart the API to test whether its job records remain available.

In this step, get ready to:
  • Create the QueueForge application shell.
  • Add request validation plus job endpoints.
  • Test whether a submitted job survives an API restart.
Create the API shell

The app object is the entry point that FastAPI serves. The jobs dictionary holds each submitted record inside the running process.

  • Use the new-file control in the left file sidebar to create main.py beside requirements.txt.
  • Build the application shell by pasting this code into main.py:
from datetime import datetime, timezone
from uuid import uuid4

from fastapi import FastAPI, HTTPException

app = FastAPI(
    title="QueueForge",
    description="A volatile asynchronous job processing API.",
    version="0.1.0",
)

jobs = {}


def utc_now():
    return datetime.now(timezone.utc).isoformat()

What Does This Code Do?

  • The datetime import supports timestamp creation.
  • The uuid4() function creates a unique ID for each job.
  • The app object sets the name shown in the interactive documentation.
  • The jobs dictionary stores records in process memory.
  • The utc_now() helper produces a timestamp with timezone information.
  • Save main.py.

The development server keeps this terminal occupied while it runs. You can stop it later with Ctrl+C.

  • Start the application by running this command in the terminal:
fastapi dev

What Does This Command Do?

The command finds the app object in main.py. It starts a development server for your API.

  • Click the forwarded http://127.0.0.1:8000/docs link in the terminal output.
  • Confirm the documentation page shows QueueForge with version 0.1.0.

Good progress. Your QueueForge application shell is live inside the Codespace.

Documentation Page Not Loading?

  • Confirm main.py appears beside requirements.txt in the file sidebar.
  • Check the terminal for a Python syntax error mentioning a line in main.py.
  • Compare the spelling of app against the code above.

Ask for help with the startup failure: help me troubleshoot why my QueueForge FastAPI documentation page is not loading

Add request checks and health

A public job endpoint needs to reject malformed requests before storing them. A health endpoint gives you a quick signal that the running process can answer requests.

  • Stop the development server by pressing Ctrl+C in the terminal.
  • Add the following code below utc_now() in main.py:
def validate_job_request(job):
    if job.get("task_type") != "sum":
        raise HTTPException(status_code=400, detail="task_type must be 'sum'")

    payload = job.get("payload")
    if not isinstance(payload, dict):
        raise HTTPException(status_code=400, detail="payload must be an object")

    numbers = payload.get("numbers")
    if not isinstance(numbers, list) or not numbers:
        raise HTTPException(
            status_code=400,
            detail="payload.numbers must be a non-empty list",
        )

    valid_numbers = all(
        isinstance(number, (int, float)) and not isinstance(number, bool)
        for number in numbers
    )
    if not valid_numbers:
        raise HTTPException(
            status_code=400,
            detail="payload.numbers must contain only numbers",
        )


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

How Do These Checks Protect the API?

  • The task_type check limits this version of QueueForge to sum jobs.
  • The payload check requires an object that contains numbers.
  • The number check rejects an empty list.
  • The final check rejects values that cannot be summed safely.
  • The /health route returns a small response that proves the process is answering requests.
  • Save main.py.
  • Restart the development server by running:
fastapi dev

What Does This Run Prove?

FastAPI reloads the saved application code. The documentation now includes the /health route.

  • Return to the existing documentation browser tab.
  • Refresh the documentation page.
  • Expand the GET /health route.
  • Enable the request using the control inside the expanded route.
  • Send the health request using the execution control.

You will see a response body containing {"status":"ok"}. Your first QueueForge endpoint is responding.

Health Route Missing?

  • Confirm you saved main.py before restarting the server.
  • Confirm @app.get("/health") appears directly above health().
  • Refresh the documentation page after the terminal reports that the server is running.

Ask for help with the missing route: help me find why GET /health is missing from my FastAPI documentation

Create job endpoints and test a restart

The job creation endpoint returns immediately with a trackable ID. The retrieval endpoint uses that ID to read the current record from jobs.

  • Stop the development server by pressing Ctrl+C in the terminal.
  • Add the following code below health() in main.py:
@app.post("/jobs", status_code=202)
def create_job(job: dict):
    validate_job_request(job)
    record = {
        "id": str(uuid4()),
        "task_type": job["task_type"],
        "payload": job["payload"],
        "status": "queued",
        "created_at": utc_now(),
    }
    jobs[record["id"]] = record
    return record


@app.get("/jobs/{job_id}")
def read_job(job_id: str):
    job = jobs.get(job_id)
    if job is None:
        raise HTTPException(status_code=404, detail="Job not found")
    return job

How Does Job Tracking Work?

  • The POST /jobs route validates the request before creating a record.
  • The generated id becomes the lookup key inside jobs.
  • A new record starts with the queued status.
  • The GET /jobs/{job_id} route returns the matching record.
  • A missing ID produces the stored Job not found response.
  • Save main.py.
  • Start the completed volatile API by running:
fastapi dev

What Changes in the Documentation?

The restarted server loads both job routes from main.py. You can now create a record through the browser.

  • Return to the documentation browser tab.
  • Refresh the documentation page.
  • Expand the POST /jobs route.
  • Enable the request body editor inside the expanded route.
  • Replace the example request body with this JSON request:
{"task_type":"sum","payload":{"numbers":[4,8,15,16,23,42]}}

What Does This Request Submit?

The request asks QueueForge to accept one sum job. Its payload contains six numbers for later processing.

  • Send the request using the execution control.

You will see a response containing an id plus the queued status. The response also includes the submitted payload.

  • Copy the returned id value into your returned job ID.
  • Expand the GET /jobs/{job_id} route.
  • Enter your returned job ID in the job_id field.
  • Send the retrieval request using the execution control.

You will see the same record with its original ID. Its status remains queued because no worker exists yet.

Before you restart the API, do you think the same job ID will still return its record?

  • Stop the development server by pressing Ctrl+C in the terminal.
  • Restart the API process by running:
fastapi dev

What Does This Restart Test?

This command launches a fresh API process. You can now request the same ID from the newly started process.

  • Return to the existing /docs browser tab.
  • Expand the GET /jobs/{job_id} route.
  • Enter your returned job ID in the job_id field.
  • Send the retrieval request using the execution control.

You will see a 404 response with Job not found. The fresh process created a new empty jobs dictionary.

You found the reliability gap on purpose. QueueForge can accept jobs now, but process memory cannot preserve them across restarts.

Job Still Appears After Restart?

  • Confirm you stopped the original server with Ctrl+C before starting it again.
  • Confirm the terminal shows a fresh server startup after the stop.
  • Confirm you used the ID copied before the restart.

Ask for help checking the restart test: help me verify that my QueueForge in-memory job disappears after a full API restart

✔️ Awesome, I've got everything!

Great. Double check that you saved main.py after adding all three routes.

ⓧ I'd like to double check the full code

Compare your complete main.py file against this reference.

from datetime import datetime, timezone
from uuid import uuid4

from fastapi import FastAPI, HTTPException

app = FastAPI(
    title="QueueForge",
    description="A volatile asynchronous job processing API.",
    version="0.1.0",
)

jobs = {}


def utc_now():
    return datetime.now(timezone.utc).isoformat()


def validate_job_request(job):
    if job.get("task_type") != "sum":
        raise HTTPException(status_code=400, detail="task_type must be 'sum'")

    payload = job.get("payload")
    if not isinstance(payload, dict):
        raise HTTPException(status_code=400, detail="payload must be an object")

    numbers = payload.get("numbers")
    if not isinstance(numbers, list) or not numbers:
        raise HTTPException(
            status_code=400,
            detail="payload.numbers must be a non-empty list",
        )

    valid_numbers = all(
        isinstance(number, (int, float)) and not isinstance(number, bool)
        for number in numbers
    )
    if not valid_numbers:
        raise HTTPException(
            status_code=400,
            detail="payload.numbers must contain only numbers",
        )


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


@app.post("/jobs", status_code=202)
def create_job(job: dict):
    validate_job_request(job)
    record = {
        "id": str(uuid4()),
        "task_type": job["task_type"],
        "payload": job["payload"],
        "status": "queued",
        "created_at": utc_now(),
    }
    jobs[record["id"]] = record
    return record


@app.get("/jobs/{job_id}")
def read_job(job_id: str):
    job = jobs.get(job_id)
    if job is None:
        raise HTTPException(status_code=404, detail="Job not found")
    return job

How to Compare the File

Check the order of each function against your file. Pay close attention to indentation inside validate_job_request() plus both job routes.

Your API accepts trackable jobs, and you have evidence of its volatile-state weakness. Next, you will move those records into durable storage so a restart cannot erase them.

Persist Jobs with SQLite

Your API can accept jobs through its interactive documentation. The restart test exposed the reliability gap because every job vanished with the API process.

This step moves job state into SQLite. The database file outlives the API process, so a restart no longer erases the queue.

In this step, get ready to:
  • Create the database schema and connection helpers.
  • Replace the in-memory dictionary with database operations.
  • Restart the API and retrieve a persisted job.
Create the SQLite storage layer

A durable database stores records on disk beyond the lifetime of one process. QueueForge uses a queueforge.db file with a jobs table for every submitted job.

  • Return to the terminal running the API from earlier.
  • Stop the API by pressing Ctrl+C.
  • Create db.py inside the current Codespace by selecting the new-file control in the Explorer sidebar.
  • Define the database path and table schema by pasting this code into db.py:
import json
import sqlite3
from contextlib import closing

DB_PATH = "queueforge.db"

SCHEMA = """
CREATE TABLE IF NOT EXISTS jobs (
    id TEXT PRIMARY KEY,
    task_type TEXT NOT NULL,
    payload TEXT NOT NULL,
    status TEXT NOT NULL CHECK (status IN ('queued', 'processing', 'completed', 'failed')),
    result TEXT,
    error TEXT,
    attempts INTEGER NOT NULL DEFAULT 0,
    created_at TEXT NOT NULL,
    started_at TEXT,
    completed_at TEXT,
    duration_ms REAL
)
"""

What does this schema store?

  • The id column gives each job its unique lookup key.
  • The payload column stores the submitted input as JSON text.
  • The status column restricts each record to one supported job state.
  • The remaining columns reserve space for processing attempts, results, errors, timestamps, and duration.
  • Save db.py.
  • Confirm that db.py now appears beside main.py in the Explorer sidebar.

Do not see db.py?

Check that the file name is exactly db.py. Confirm that you created it beside main.py in the current Codespace.

Ask for help if the file appears in a different folder: help me place db.py beside main.py in my GitHub Codespaces workspace

Each database operation uses a fresh connection. The connection receives named-column access through sqlite3.Row before it is closed after the operation.

  • Add the connection and initialization functions below SCHEMA in db.py by pasting:
def connect():
    connection = sqlite3.connect(DB_PATH, timeout=5.0)
    connection.row_factory = sqlite3.Row
    return connection


def init_db():
    with closing(connect()) as connection:
        connection.execute(SCHEMA)
        connection.execute(
            "CREATE INDEX IF NOT EXISTS idx_jobs_status_created "
            "ON jobs(status, created_at)"
        )
        connection.commit()

How does initialization work?

  • The connect() function opens queueforge.db with a five-second lock timeout.
  • The init_db() function creates the table when it is missing.
  • The status index prepares the database to find queued jobs efficiently.
  • The explicit commit makes the schema changes durable before the connection closes.
  • Save db.py.
  • Create the database file by running the initialization helper in the terminal:
python -c "from db import init_db; init_db(); print('queueforge.db created')"

What does this check prove?

Python imports init_db() from your new module. The helper creates the database schema before the command prints its confirmation.

You should see queueforge.db created in the terminal. You should also see queueforge.db appear in the Explorer sidebar.

Database file missing?

Check db.py for mismatched quotes or indentation. Confirm that init_db() includes connection.commit().

Ask for help with the terminal output: help me diagnose why init_db did not create queueforge.db

Connect the API to durable storage

Database writes need safe value binding. Parameterized SQL sends the statement separately from its values, which prevents submitted data from changing the SQL command.

  • Add the job insertion function below init_db() in db.py by pasting:
def insert_job(job):
    with closing(connect()) as connection:
        connection.execute(
            """
            INSERT INTO jobs (id, task_type, payload, status, created_at)
            VALUES (?, ?, ?, ?, ?)
            """,
            (
                job["id"],
                job["task_type"],
                json.dumps(job["payload"]),
                job["status"],
                job["created_at"],
            ),
        )
        connection.commit()

How are jobs written safely?

  • The question-mark placeholders receive job values through parameter binding.
  • The json.dumps() call converts the payload dictionary into text that SQLite can store.
  • The commit preserves the inserted row after the short-lived connection closes.
  • Save db.py.
  • Add the row-decoding function directly below insert_job() by pasting:
def decode_job(row):
    if row is None:
        return None
    job = dict(row)
    job["payload"] = json.loads(job["payload"])
    job["result"] = json.loads(job["result"]) if job["result"] else None
    return job

Why decode each row?

SQLite returns the stored payload as text. The decode_job() function restores that text to a Python dictionary before the API returns it.

  • Save db.py.
  • Confirm that both write helpers load without a syntax error by running:
python -c "from db import insert_job, decode_job; print('write helpers loaded')"

What does this import check show?

Python parses db.py before importing the two helpers. The confirmation appears only when both function definitions load successfully.

You should see write helpers loaded in the terminal.

  • Add the lookup function below decode_job() in db.py by pasting:
def get_job(job_id):
    with closing(connect()) as connection:
        row = connection.execute(
            "SELECT * FROM jobs WHERE id = ?",
            (job_id,),
        ).fetchone()
    return decode_job(row)

How does job lookup work?

The query selects one row through a bound job ID. The function returns a decoded job dictionary or None when no row matches.

  • Save db.py.
  • Switch back to main.py in the Explorer sidebar.
  • Replace everything from the first import through jobs = {} with this database-backed setup:
from datetime import datetime, timezone
from uuid import uuid4

from fastapi import FastAPI, HTTPException

from db import get_job, init_db, insert_job

app = FastAPI(
    title="QueueForge",
    description="A durable asynchronous job processing API.",
    version="1.0.0",
)

init_db()

What changes at startup?

  • The API imports its storage operations from db.py.
  • The init_db() call confirms the database schema whenever the API starts.
  • The API description now records that QueueForge uses durable storage.
  • Save main.py.
  • Start the API with its database initialization by running:
fastapi dev

What happens when the API starts?

FastAPI loads the app object from main.py. The startup code opens the existing database or creates its schema before serving requests.

  • Return to the forwarded interactive documentation at http://127.0.0.1:8000/docs.
  • Send a request through GET /health to confirm the API loaded successfully.

You should receive a successful response containing the ok status.

API failed to load?

Check that db.py sits beside main.py. Compare the imported names with get_job, init_db, and insert_job.

Ask for help with the terminal error: help me fix the database import or startup error in QueueForge

The endpoints still need to call the new storage helpers. These final edits remove every remaining dependency on the deleted in-memory dictionary.

  • In create_job(), replace the dictionary assignment and direct return with these lines:
    insert_job(record)
    return get_job(record["id"])

How does submission change?

The endpoint commits the new record through insert_job(). It reads the stored row back so the response matches the database representation.

  • Replace the existing read_job() function with this database-backed version:
@app.get("/jobs/{job_id}")
def read_job(job_id: str):
    job = get_job(job_id)
    if job is None:
        raise HTTPException(status_code=404, detail="Job not found")
    return job

How does retrieval change?

The endpoint now searches queueforge.db through get_job(). The existing missing-job response remains unchanged.

  • Save main.py.
  • Return to the terminal to confirm the development server reloaded the updated API.

You should see the server complete its reload without a Python error.

Server reload shows an error?

Check that jobs[record["id"]] = record and jobs.get(job_id) no longer remain in main.py.

Ask for help with the failing line: help me replace the in-memory jobs references with SQLite helpers

Prove the job survives a restart

A durable record must remain retrievable after the API process stops. You will submit one job, restart the server, and request the same ID again.

  • Return to the forwarded interactive documentation at http://127.0.0.1:8000/docs.
  • Expand POST /jobs.
  • Place this request in the request body editor:
{"task_type":"sum","payload":{"numbers":[4,8,15,16,23,42]}}

What does this request create?

The request creates a sum job with six numbers. The API stores it immediately with a generated ID and the queued status.

  • Send the POST /jobs request.
  • Record the returned ID here: your returned job ID.
  • Expand GET /jobs/{job_id}.
  • Enter your returned job ID as the path value.
  • Send the retrieval request.

You should see the submitted payload with the queued status.

  • Return to the terminal running the API.
  • Stop the API by pressing Ctrl+C.
  • Restart the API by running:
fastapi dev

What should persist through this restart?

The API process starts with fresh memory. The job remains in queueforge.db because the earlier insert was committed to disk.

Before you retrieve the job again, do you expect the API to return the record or a missing-job response?

  • Return to GET /jobs/{job_id} in the interactive documentation.
  • Enter your returned job ID as the path value.
  • Send the retrieval request after the restart.

You should see the original job with its queued status and six-number payload. The same ID now survives a full API restart.

That reliability gap is closed. QueueForge now keeps its job records beyond the lifetime of the API process.

Job missing after the restart?

Confirm that you copied the ID from the database-backed submission made in this step. A job created before this step existed only in memory.

Check that create_job() calls insert_job(record) before returning the saved row.

Ask for help with the saved record: help me find why my SQLite job disappeared after restarting FastAPI

✔️ Awesome, I've got db.py

Great. Double check that you saved db.py in the Codespace.

ⓧ I'd like to double check db.py

Compare your complete db.py file with this reference.

import json
import sqlite3
from contextlib import closing

DB_PATH = "queueforge.db"

SCHEMA = """
CREATE TABLE IF NOT EXISTS jobs (
    id TEXT PRIMARY KEY,
    task_type TEXT NOT NULL,
    payload TEXT NOT NULL,
    status TEXT NOT NULL CHECK (status IN ('queued', 'processing', 'completed', 'failed')),
    result TEXT,
    error TEXT,
    attempts INTEGER NOT NULL DEFAULT 0,
    created_at TEXT NOT NULL,
    started_at TEXT,
    completed_at TEXT,
    duration_ms REAL
)
"""


def connect():
    connection = sqlite3.connect(DB_PATH, timeout=5.0)
    connection.row_factory = sqlite3.Row
    return connection


def init_db():
    with closing(connect()) as connection:
        connection.execute(SCHEMA)
        connection.execute(
            "CREATE INDEX IF NOT EXISTS idx_jobs_status_created "
            "ON jobs(status, created_at)"
        )
        connection.commit()


def insert_job(job):
    with closing(connect()) as connection:
        connection.execute(
            """
            INSERT INTO jobs (id, task_type, payload, status, created_at)
            VALUES (?, ?, ?, ?, ?)
            """,
            (
                job["id"],
                job["task_type"],
                json.dumps(job["payload"]),
                job["status"],
                job["created_at"],
            ),
        )
        connection.commit()


def decode_job(row):
    if row is None:
        return None
    job = dict(row)
    job["payload"] = json.loads(job["payload"])
    job["result"] = json.loads(job["result"]) if job["result"] else None
    return job


def get_job(job_id):
    with closing(connect()) as connection:
        row = connection.execute(
            "SELECT * FROM jobs WHERE id = ?",
            (job_id,),
        ).fetchone()
    return decode_job(row)

How to compare this file

Check each function name, indentation level, SQL placeholder, and commit. Your file should end with return decode_job(row).

✔️ Awesome, I've got main.py

Great. Double check that you saved main.py after the server reloaded.

ⓧ I'd like to double check main.py

Compare your complete main.py file with this database-backed reference.

from datetime import datetime, timezone
from uuid import uuid4

from fastapi import FastAPI, HTTPException

from db import get_job, init_db, insert_job

app = FastAPI(
    title="QueueForge",
    description="A durable asynchronous job processing API.",
    version="1.0.0",
)

init_db()


def utc_now():
    return datetime.now(timezone.utc).isoformat()


def validate_job_request(job):
    if job.get("task_type") != "sum":
        raise HTTPException(status_code=400, detail="task_type must be 'sum'")

    payload = job.get("payload")
    if not isinstance(payload, dict):
        raise HTTPException(status_code=400, detail="payload must be an object")

    numbers = payload.get("numbers")
    if not isinstance(numbers, list) or not numbers:
        raise HTTPException(
            status_code=400,
            detail="payload.numbers must be a non-empty list",
        )

    valid_numbers = all(
        isinstance(number, (int, float)) and not isinstance(number, bool)
        for number in numbers
    )
    if not valid_numbers:
        raise HTTPException(
            status_code=400,
            detail="payload.numbers must contain only numbers",
        )


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


@app.post("/jobs", status_code=202)
def create_job(job: dict):
    validate_job_request(job)
    record = {
        "id": str(uuid4()),
        "task_type": job["task_type"],
        "payload": job["payload"],
        "status": "queued",
        "created_at": utc_now(),
    }
    insert_job(record)
    return get_job(record["id"])


@app.get("/jobs/{job_id}")
def read_job(job_id: str):
    job = get_job(job_id)
    if job is None:
        raise HTTPException(status_code=404, detail="Job not found")
    return job

How to compare this file

Confirm that the in-memory jobs dictionary is gone. Both job endpoints should use the imported database helpers.

Your API now remembers every queued job after a restart. Next up, you will separate processing from request handling with a dedicated worker.

Run a Separate Worker

Your durable FastAPI API accepts jobs quickly. Processing those jobs inside the request handler would make every client wait for the calculation to finish.

A separate worker can claim queued work while the API continues handling requests. Shared SQLite records let both processes coordinate without tying their lifecycles together.

In this step, get ready to:
  • Add transactional helpers that claim jobs and store terminal outcomes.
  • Create a worker process that calculates sums outside the API.
  • Prove that normal jobs complete while controlled failures remain inspectable.
Add transactional job updates

Two workers could read the same queued row at almost the same moment. A write transaction makes each worker secure its claim before processing begins.

  • Leave fastapi dev running in the first terminal from earlier.
  • Return to db.py in the Codespaces editor.
  • Append the job-claiming helper below get_job() by adding this code:
def claim_next_job(started_at):
    with closing(connect()) as connection:
        connection.execute("BEGIN IMMEDIATE")
        row = connection.execute(
            """
            SELECT * FROM jobs
            WHERE status = 'queued'
            ORDER BY created_at
            LIMIT 1
            """
        ).fetchone()

        if row is None:
            connection.commit()
            return None

        updated = connection.execute(
            """
            UPDATE jobs
            SET status = 'processing', started_at = ?, attempts = attempts + 1
            WHERE id = ? AND status = 'queued'
            """,
            (started_at, row["id"]),
        ).rowcount
        connection.commit()

        if updated != 1:
            return None
        return decode_job(row)

How Does a Worker Claim One Job?

  • The BEGIN IMMEDIATE transaction reserves the database write path before the worker selects a job.
  • The query selects the oldest record whose status is queued.
  • The conditional update changes that record to processing only while it remains queued.
  • The affected row count confirms that this worker secured the claim.
  • Save db.py.
  • Check the first terminal for a successful FastAPI reload with no Python traceback.

Did the API Reload Fail?

Compare the indentation inside claim_next_job() with the snippet above. The transaction and both queries must remain inside the connection block.

Check that every triple-quoted SQL string has matching opening and closing quotes.

Help me fix the new claim_next_job function in db.py.

Claiming work only records that processing began. The database also needs terminal updates for successful and failed jobs.

  • Append the successful completion helper below claim_next_job() by adding this code:
def complete_job(job_id, result, completed_at, duration_ms):
    with closing(connect()) as connection:
        connection.execute(
            """
            UPDATE jobs
            SET status = 'completed', result = ?, error = NULL,
                completed_at = ?, duration_ms = ?
            WHERE id = ?
            """,
            (json.dumps(result), completed_at, duration_ms, job_id),
        )
        connection.commit()

What Does This Helper Store?

  • The result becomes JSON text so the existing decode_job() helper can reconstruct it later.
  • The status becomes completed.
  • The completion time and measured duration preserve evidence about the finished work.
  • Save db.py.
  • Confirm the first terminal reloads the API without a traceback.

Did the Completion Helper Break Reloading?

Check the commas after the SQL string and parameter tuple. A missing comma can turn the database call into invalid Python.

Help me debug complete_job in db.py.

  • Append the failed outcome helper below complete_job() by adding this code:
def fail_job(job_id, error, completed_at, duration_ms):
    with closing(connect()) as connection:
        connection.execute(
            """
            UPDATE jobs
            SET status = 'failed', error = ?, completed_at = ?, duration_ms = ?
            WHERE id = ?
            """,
            (error, completed_at, duration_ms, job_id),
        )
        connection.commit()

Why Store Failures?

The worker converts an exception into a durable failed record. The API stays available while the stored error explains what happened to that job.

  • Save db.py.
  • Confirm the first terminal reloads the API without a traceback.

Did the Failure Helper Break Reloading?

Check that the SQL parameter order matches error, completed_at, duration_ms, and job_id.

Help me debug fail_job in db.py.

✔️ Awesome, I've got everything!

Your db.py file can now claim queued jobs and store both terminal outcomes.

ⓧ I'd like to double check the full code

Compare your complete db.py file with this version.

import json
import sqlite3
from contextlib import closing

DB_PATH = "queueforge.db"

SCHEMA = """
CREATE TABLE IF NOT EXISTS jobs (
    id TEXT PRIMARY KEY,
    task_type TEXT NOT NULL,
    payload TEXT NOT NULL,
    status TEXT NOT NULL CHECK (status IN ('queued', 'processing', 'completed', 'failed')),
    result TEXT,
    error TEXT,
    attempts INTEGER NOT NULL DEFAULT 0,
    created_at TEXT NOT NULL,
    started_at TEXT,
    completed_at TEXT,
    duration_ms REAL
)
"""


def connect():
    connection = sqlite3.connect(DB_PATH, timeout=5.0)
    connection.row_factory = sqlite3.Row
    return connection


def init_db():
    with closing(connect()) as connection:
        connection.execute(SCHEMA)
        connection.execute(
            "CREATE INDEX IF NOT EXISTS idx_jobs_status_created "
            "ON jobs(status, created_at)"
        )
        connection.commit()


def insert_job(job):
    with closing(connect()) as connection:
        connection.execute(
            """
            INSERT INTO jobs (id, task_type, payload, status, created_at)
            VALUES (?, ?, ?, ?, ?)
            """,
            (
                job["id"],
                job["task_type"],
                json.dumps(job["payload"]),
                job["status"],
                job["created_at"],
            ),
        )
        connection.commit()


def decode_job(row):
    if row is None:
        return None
    job = dict(row)
    job["payload"] = json.loads(job["payload"])
    job["result"] = json.loads(job["result"]) if job["result"] else None
    return job


def get_job(job_id):
    with closing(connect()) as connection:
        row = connection.execute(
            "SELECT * FROM jobs WHERE id = ?",
            (job_id,),
        ).fetchone()
    return decode_job(row)


def claim_next_job(started_at):
    with closing(connect()) as connection:
        connection.execute("BEGIN IMMEDIATE")
        row = connection.execute(
            """
            SELECT * FROM jobs
            WHERE status = 'queued'
            ORDER BY created_at
            LIMIT 1
            """
        ).fetchone()

        if row is None:
            connection.commit()
            return None

        updated = connection.execute(
            """
            UPDATE jobs
            SET status = 'processing', started_at = ?, attempts = attempts + 1
            WHERE id = ? AND status = 'queued'
            """,
            (started_at, row["id"]),
        ).rowcount
        connection.commit()

        if updated != 1:
            return None
        return decode_job(row)


def complete_job(job_id, result, completed_at, duration_ms):
    with closing(connect()) as connection:
        connection.execute(
            """
            UPDATE jobs
            SET status = 'completed', result = ?, error = NULL,
                completed_at = ?, duration_ms = ?
            WHERE id = ?
            """,
            (json.dumps(result), completed_at, duration_ms, job_id),
        )
        connection.commit()


def fail_job(job_id, error, completed_at, duration_ms):
    with closing(connect()) as connection:
        connection.execute(
            """
            UPDATE jobs
            SET status = 'failed', error = ?, completed_at = ?, duration_ms = ?
            WHERE id = ?
            """,
            (error, completed_at, duration_ms, job_id),
        )
        connection.commit()
Build the worker process

The worker polls for one queued record at a time. Its processing function either calculates a sum or raises a controlled exception that the worker can store.

  • Select the new-file icon in the Codespaces file sidebar.
  • Name the file worker.py.
  • Add the worker configuration and processing function by pasting this code:
import time
from datetime import datetime, timezone

from db import claim_next_job, complete_job, fail_job, init_db

POLL_INTERVAL_SECONDS = 0.5
SIMULATED_WORK_SECONDS = 0.4


def utc_now():
    return datetime.now(timezone.utc).isoformat()


def process_job(job):
    payload = job["payload"]
    if payload.get("force_failure") is True:
        raise RuntimeError("Forced failure requested for demonstration")

    time.sleep(SIMULATED_WORK_SECONDS)
    numbers = payload["numbers"]
    return {"sum": sum(numbers), "count": len(numbers)}

What Does the Processing Function Do?

  • The polling interval limits how often an idle worker checks the database.
  • The simulated work interval makes each job take a measurable amount of time.
  • A payload containing force_failure set to true raises the demonstration error.
  • Every other valid payload produces a result containing its sum and item count.
  • Save worker.py.
  • Confirm worker.py appears beside main.py in the file sidebar.

Is worker.py Missing?

Check the file sidebar for a file named exactly worker.py. A different extension prevents Python from treating it as the intended script.

Help me locate or recreate worker.py in my Codespaces workspace.

The worker loop connects the claim and outcome helpers. It keeps polling until you stop the process.

  • Append the worker loop below process_job() by adding this code:
def run_worker():
    init_db()
    print("QueueForge worker started. Press Ctrl+C to stop.")

    while True:
        job = claim_next_job(utc_now())
        if job is None:
            time.sleep(POLL_INTERVAL_SECONDS)
            continue

        print(f"Processing {job['id']}")
        started = time.perf_counter()
        try:
            result = process_job(job)
            duration_ms = (time.perf_counter() - started) * 1000
            complete_job(job["id"], result, utc_now(), duration_ms)
            print(f"Completed {job['id']}")
        except Exception as error:
            duration_ms = (time.perf_counter() - started) * 1000
            fail_job(job["id"], str(error), utc_now(), duration_ms)
            print(f"Failed {job['id']}: {error}")


if __name__ == "__main__":
    run_worker()

How Does the Worker Loop Behave?

  • An idle worker sleeps briefly before checking for another queued record.
  • A claimed job moves through process_job() while the loop measures its duration.
  • Successful work calls complete_job().
  • An exception calls fail_job() and leaves the worker running.
  • Save worker.py.
  • Create a second terminal with the new-terminal control in the terminal panel.
  • Start the separate worker process by running this command:
python worker.py

What Should You See?

The second terminal prints QueueForge worker started. Press Ctrl+C to stop. The process keeps that terminal busy while it polls for jobs.

The first terminal continues running the API. The second terminal now owns background processing.

Did the Worker Stop Immediately?

Check that the bottom of worker.py calls run_worker() inside the __main__ condition.

Check that all four database helpers in the import line match the names in db.py.

Help me start the QueueForge worker process.

✔️ Awesome, I've got everything!

Your second terminal is now running the worker independently from the API.

ⓧ I'd like to double check the full code

Compare your complete worker.py file with this version.

import time
from datetime import datetime, timezone

from db import claim_next_job, complete_job, fail_job, init_db

POLL_INTERVAL_SECONDS = 0.5
SIMULATED_WORK_SECONDS = 0.4


def utc_now():
    return datetime.now(timezone.utc).isoformat()


def process_job(job):
    payload = job["payload"]
    if payload.get("force_failure") is True:
        raise RuntimeError("Forced failure requested for demonstration")

    time.sleep(SIMULATED_WORK_SECONDS)
    numbers = payload["numbers"]
    return {"sum": sum(numbers), "count": len(numbers)}


def run_worker():
    init_db()
    print("QueueForge worker started. Press Ctrl+C to stop.")

    while True:
        job = claim_next_job(utc_now())
        if job is None:
            time.sleep(POLL_INTERVAL_SECONDS)
            continue

        print(f"Processing {job['id']}")
        started = time.perf_counter()
        try:
            result = process_job(job)
            duration_ms = (time.perf_counter() - started) * 1000
            complete_job(job["id"], result, utc_now(), duration_ms)
            print(f"Completed {job['id']}")
        except Exception as error:
            duration_ms = (time.perf_counter() - started) * 1000
            fail_job(job["id"], str(error), utc_now(), duration_ms)
            print(f"Failed {job['id']}: {error}")


if __name__ == "__main__":
    run_worker()
Inspect completed and failed jobs

The queued job from the previous step gives you the first end-to-end test. Its record should now contain a terminal result because the worker claimed it from the shared database.

  • Record the queued job ID from earlier here: your queued job ID.
  • Return to the forwarded http://127.0.0.1:8000/docs page from earlier.

Before you check, consider whether the queued job is still waiting or already completed.

  • Expand the GET /jobs/{job_id} route.
  • Enter your queued job ID in the job ID field.
  • Submit the retrieval request with the route panel's execution control.

You should see a completed record. Its result contains the correct sum and count for the submitted numbers.

A reliable queue also preserves failures as data. The next request deliberately asks the worker to follow its controlled failure path.

  • Expand the POST /jobs route in the interactive documentation.
  • Reuse the valid sum request body from earlier.
  • Add force_failure inside the payload object.
  • Set the force_failure value to true.
  • Submit the forced-failure job with the route panel's execution control.
  • Record its returned ID here: your failed job ID.

Before you retrieve the record, consider whether the worker exception will stop the API or become a stored job outcome.

  • Return to the GET /jobs/{job_id} route.
  • Enter your failed job ID in the job ID field.
  • Submit the retrieval request with the route panel's execution control.
  • Repeat the retrieval request until the stored status becomes failed.

You should see the stored error Forced failure requested for demonstration. The API remains available because the worker converted the exception into a durable failure record.

Is the Job Still Queued?

Check the second terminal for the worker startup message. A stopped worker leaves submitted jobs in the queue.

Check the first terminal separately to confirm the API is still running. Both processes must stay active during this test.

Help me troubleshoot a QueueForge job that remains queued.

You have separated fast request handling from background processing. QueueForge now completes valid work and preserves controlled failures without crashing the API.

Test the Burst and Publish

Your FastAPI API now accepts durable jobs. Your worker records completed results or stored failures in SQLite.

A scalability claim needs measurable evidence. In this step, you will expose metrics, run a concurrent burst test, document the result, and publish QueueForge in a public GitHub repository.

In this step, get ready to:
  • Measure queue completion with 20 jobs across five clients.
  • Document the measured result in a project README.
  • Publish QueueForge as a public repository.
Build the metrics and burst test

A burst test needs a finish line. The API will report status counts plus the average duration of completed jobs.

  • In db.py, add the following get_metrics() function below fail_job():
def get_metrics():
    counts = {
        "queued": 0,
        "processing": 0,
        "completed": 0,
        "failed": 0,
    }

    with closing(connect()) as connection:
        rows = connection.execute(
            "SELECT status, COUNT(*) AS count FROM jobs GROUP BY status"
        ).fetchall()
        average_row = connection.execute(
            """
            SELECT AVG(duration_ms) AS average_duration_ms
            FROM jobs
            WHERE status = 'completed'
            """
        ).fetchone()

    for row in rows:
        counts[row["status"]] = row["count"]

    average = average_row["average_duration_ms"] or 0
    return {
        **counts,
        "total": sum(counts.values()),
        "average_completed_duration_ms": round(average, 2),
    }

What does this function measure?

  • The counts dictionary gives every possible job state a zero-value starting point.
  • The grouped query counts the stored rows for each status.
  • The average query measures completed jobs with recorded durations.
  • The returned object combines status counts with a rounded average duration.
  • Save db.py.
  • In main.py, replace the existing database import with this line:
from db import get_job, get_metrics, init_db, insert_job

Why import get_metrics?

The API route needs direct access to the database summary. This import connects that route to get_metrics().

  • At the bottom of main.py, add the metrics route:
@app.get("/metrics")
def metrics():
    return get_metrics()

What does this route expose?

The /metrics route returns the current queue counts. It also exposes the average duration recorded for completed work.

  • Save main.py.

Before you check the new route, which job states do you expect to find in the response?

  • Return to the browser connected to your running API.
  • Open http://127.0.0.1:8000/metrics through the forwarded port.

You will see an object containing queued, processing, completed, failed, total, and average_completed_duration_ms.

Metrics route unavailable?

Check that the API terminal still shows the development server running. Confirm that get_metrics appears in the database import inside main.py.

Compare the route path with /metrics. A missing slash or spelling difference creates a different route.

Help me troubleshoot the QueueForge metrics route.

The load test uses ThreadPoolExecutor to send five requests at a time. It then polls the metrics route until no queued or processing work remains.

  • Use the Explorer sidebar in the existing Codespaces workspace to create load_test.py.
  • Add the request helpers plus the burst settings with this code:
import json
import time
import urllib.request
from concurrent.futures import ThreadPoolExecutor

BASE_URL = "http://127.0.0.1:8000"
JOB_COUNT = 20
MAX_CLIENTS = 5


def read_json(url):
    with urllib.request.urlopen(url, timeout=5) as response:
        return json.loads(response.read().decode("utf-8"))


def submit_job(index):
    body = {
        "task_type": "sum",
        "payload": {"numbers": [index, index + 1, index + 2]},
    }
    request = urllib.request.Request(
        url=f"{BASE_URL}/jobs",
        data=json.dumps(body).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=5) as response:
        return json.loads(response.read().decode("utf-8"))

How are jobs submitted concurrently?

  • The constants target the running API with 20 jobs across five clients.
  • The read_json() helper reads metrics responses.
  • The submit_job() helper sends one numbered sum job.
  • Each response becomes a Python dictionary for the final accepted-job count.
  • Save load_test.py.
  • Confirm that load_test.py now appears in the Explorer sidebar.
  • Add the following main() function below submit_job():
def main():
    started = time.perf_counter()
    with ThreadPoolExecutor(max_workers=MAX_CLIENTS) as executor:
        jobs = list(executor.map(submit_job, range(JOB_COUNT)))
    submitted_at = time.perf_counter()

    deadline = time.perf_counter() + 30
    while True:
        metrics = read_json(f"{BASE_URL}/metrics")
        if metrics["queued"] == 0 and metrics["processing"] == 0:
            break
        if time.perf_counter() >= deadline:
            raise TimeoutError("The queue did not drain within 30 seconds")
        time.sleep(0.25)

    finished = time.perf_counter()
    print(
        json.dumps(
            {
                "accepted_jobs": len(jobs),
                "submission_seconds": round(submitted_at - started, 3),
                "completed_in_seconds": round(finished - started, 3),
                "final_metrics": metrics,
            },
            indent=2,
        )
    )

How does the burst finish?

  • The executor maps submit_job() across all 20 job indexes.
  • The polling loop waits for both active queue counts to reach zero.
  • The deadline prevents the test from waiting forever.
  • The final object reports submission speed plus total queue completion time.
  • Add the script entry point at the bottom of load_test.py:
if __name__ == "__main__":
    main()

Why add an entry point?

The entry point calls main() when Python runs the file directly. This starts the burst from the terminal.

  • Save load_test.py.
Run the burst and record evidence

The burst separates API acceptance speed from worker completion speed. Your API can accept requests concurrently while one worker drains the durable queue at its own pace.

Before you run the burst, do you expect submission time or queue completion time to be larger?

  • Use the terminal panel's add control to create a third terminal.
  • Confirm that the API remains active in the first terminal.
  • Confirm that the worker remains active in the second terminal.
  • Submit the 20-job burst from the third terminal by running:
python load_test.py

What should I see?

The final object shows accepted_jobs as 20. It shows queued and processing as 0 inside final_metrics.

That is the full queue lifecycle measured in one run. The API accepted the burst before the worker finished processing every job.

Burst test did not finish?

Confirm that the API terminal remains active. Confirm that the worker terminal continues printing processing messages.

Check that JOB_COUNT is 20. Check that MAX_CLIENTS is 5.

Help me diagnose my QueueForge burst test.

A public repository should contain the source plus enough evidence to explain the design. The local database stays outside version control because it contains generated job records.

  • Use the Explorer sidebar to create .gitignore inside the current Codespaces workspace.
  • Add the generated-file exclusions with this content:
__pycache__/
*.py[cod]
.venv/
queueforge.db

What stays out of Git?

  • Python cache files stay local because the interpreter regenerates them.
  • Virtual environment files stay local because dependencies come from requirements.txt.
  • The queueforge.db file stays local because runtime data does not belong in the source repository.
  • Save .gitignore.
  • Return to the Source Control view to confirm that queueforge.db is absent from the pending files.
  • Use the Explorer sidebar to create README.md inside the current Codespaces workspace.
  • Add the project summary plus the architecture section with this content:
# QueueForge

QueueForge is a durable job-processing API built to demonstrate backend development, persistence, failure handling, burst testing, and system-design trade-offs.

## Demo

1. Submit a `sum` job through the interactive API documentation.
2. Receive a job ID immediately with status `queued`.
3. Run the worker and retrieve the job until it becomes `completed` or `failed`.
4. Restart the API and retrieve the same record from SQLite.
5. Run the burst test and inspect final metrics.

## Architecture

```text
Client requests
      |
      v
FastAPI API  ---- reads and writes ---->  SQLite jobs table
                                              ^
                                              |
                                  claims and updates
                                              |
                                       Worker process
```

The API and worker are separate processes. The API accepts work without waiting for processing to finish. The SQLite table is both the durable state store and the small-project queue. `BEGIN IMMEDIATE` serializes job claims so two local workers do not claim the same queued row.

What does this section communicate?

The opening gives readers a demo path they can follow. The diagram shows that the API plus worker share the SQLite jobs table.

The architecture note connects transactional claims to duplicate-work prevention. That makes the reliability decision visible to an employer.

  • Save README.md.
  • Confirm that the editor shows the QueueForge title plus the architecture diagram.
  • Append the run instructions with this content:
## Run

```text
python -m pip install -r requirements.txt
fastapi dev
```

Open a second terminal:

```text
python worker.py
```

Open a third terminal for the burst test:

```text
python load_test.py
```

Interactive API documentation is available at `http://127.0.0.1:8000/docs` inside the Codespace through its forwarded port.

Why include run instructions?

The run section gives readers the same three-process workflow you tested. They can start the API, worker, and burst test in separate terminals.

  • Save README.md.
  • Confirm that the editor shows commands for the API, worker, and burst test.
  • Append the evidence plus the design decisions with this content:
## Evidence to Record

Record the output from `python load_test.py` here after your run:

- Accepted jobs: 20
- Submission time: use your observed `submission_seconds`
- Queue completion time: use your observed `completed_in_seconds`
- Final queued jobs: 0
- Final processing jobs: 0

## Reliability Decisions

- Job state is stored outside the API process, so an API restart does not erase it.
- Each database operation uses parameterized SQL.
- A worker records terminal success or failure instead of crashing the API.
- The database transaction protects a job claim from competing local workers.

## Scaling Trade-offs

SQLite keeps this project free and easy to run, but concurrent writes are serialized. A production evolution would move queue delivery to a dedicated message broker or managed queue and move durable state to a server database before distributing workers across machines.

## Resume Talking Point

Built a durable asynchronous job-processing API with separate API and worker processes, transactional job claiming, persisted failure states, concurrent burst testing, and documented scaling trade-offs.

How does this turn output into evidence?

The evidence section records the accepted workload plus the final queue state. The reliability section explains how persistence plus transactions protect job tracking.

The scaling section names SQLite's serialized-write boundary. The resume line condenses the project into one demonstrable backend accomplishment.

  • Save README.md.
  • Confirm that the editor shows the Scaling Trade-offs section plus the Resume Talking Point.

README sections missing?

Check that each heading begins with ## on its own line. Confirm that each fenced block has an opening marker plus a closing marker.

Help me compare my QueueForge README with the required project evidence.

✔️ Awesome, I've got everything!

Your source files are saved. Your burst evidence plus repository documentation are ready for publication.

ⓧ I'd like to double check the full code

Compare these complete files with the versions in your Codespaces workspace.

__pycache__/
*.py[cod]
.venv/
queueforge.db

# QueueForge

QueueForge is a durable job-processing API built to demonstrate backend development, persistence, failure handling, burst testing, and system-design trade-offs.

## Demo

1. Submit a `sum` job through the interactive API documentation.
2. Receive a job ID immediately with status `queued`.
3. Run the worker and retrieve the job until it becomes `completed` or `failed`.
4. Restart the API and retrieve the same record from SQLite.
5. Run the burst test and inspect final metrics.

## Architecture

```text
Client requests
      |
      v
FastAPI API  ---- reads and writes ---->  SQLite jobs table
                                              ^
                                              |
                                  claims and updates
                                              |
                                       Worker process
```

The API and worker are separate processes. The API accepts work without waiting for processing to finish. The SQLite table is both the durable state store and the small-project queue. `BEGIN IMMEDIATE` serializes job claims so two local workers do not claim the same queued row.

## Run

```text
python -m pip install -r requirements.txt
fastapi dev
```

Open a second terminal:

```text
python worker.py
```

Open a third terminal for the burst test:

```text
python load_test.py
```

Interactive API documentation is available at `http://127.0.0.1:8000/docs` inside the Codespace through its forwarded port.

## Evidence to Record

Record the output from `python load_test.py` here after your run:

- Accepted jobs: 20
- Submission time: use your observed `submission_seconds`
- Queue completion time: use your observed `completed_in_seconds`
- Final queued jobs: 0
- Final processing jobs: 0

## Reliability Decisions

- Job state is stored outside the API process, so an API restart does not erase it.
- Each database operation uses parameterized SQL.
- A worker records terminal success or failure instead of crashing the API.
- The database transaction protects a job claim from competing local workers.

## Scaling Trade-offs

SQLite keeps this project free and easy to run, but concurrent writes are serialized. A production evolution would move queue delivery to a dedicated message broker or managed queue and move durable state to a server database before distributing workers across machines.

## Resume Talking Point

Built a durable asynchronous job-processing API with separate API and worker processes, transactional job claiming, persisted failure states, concurrent burst testing, and documented scaling trade-offs.

import json
import sqlite3
from contextlib import closing

DB_PATH = "queueforge.db"

SCHEMA = """
CREATE TABLE IF NOT EXISTS jobs (
    id TEXT PRIMARY KEY,
    task_type TEXT NOT NULL,
    payload TEXT NOT NULL,
    status TEXT NOT NULL CHECK (status IN ('queued', 'processing', 'completed', 'failed')),
    result TEXT,
    error TEXT,
    attempts INTEGER NOT NULL DEFAULT 0,
    created_at TEXT NOT NULL,
    started_at TEXT,
    completed_at TEXT,
    duration_ms REAL
)
"""


def connect():
    connection = sqlite3.connect(DB_PATH, timeout=5.0)
    connection.row_factory = sqlite3.Row
    return connection


def init_db():
    with closing(connect()) as connection:
        connection.execute(SCHEMA)
        connection.execute(
            "CREATE INDEX IF NOT EXISTS idx_jobs_status_created "
            "ON jobs(status, created_at)"
        )
        connection.commit()


def insert_job(job):
    with closing(connect()) as connection:
        connection.execute(
            """
            INSERT INTO jobs (id, task_type, payload, status, created_at)
            VALUES (?, ?, ?, ?, ?)
            """,
            (
                job["id"],
                job["task_type"],
                json.dumps(job["payload"]),
                job["status"],
                job["created_at"],
            ),
        )
        connection.commit()


def decode_job(row):
    if row is None:
        return None
    job = dict(row)
    job["payload"] = json.loads(job["payload"])
    job["result"] = json.loads(job["result"]) if job["result"] else None
    return job


def get_job(job_id):
    with closing(connect()) as connection:
        row = connection.execute(
            "SELECT * FROM jobs WHERE id = ?",
            (job_id,),
        ).fetchone()
    return decode_job(row)


def claim_next_job(started_at):
    with closing(connect()) as connection:
        connection.execute("BEGIN IMMEDIATE")
        row = connection.execute(
            """
            SELECT * FROM jobs
            WHERE status = 'queued'
            ORDER BY created_at
            LIMIT 1
            """
        ).fetchone()

        if row is None:
            connection.commit()
            return None

        updated = connection.execute(
            """
            UPDATE jobs
            SET status = 'processing', started_at = ?, attempts = attempts + 1
            WHERE id = ? AND status = 'queued'
            """,
            (started_at, row["id"]),
        ).rowcount
        connection.commit()

        if updated != 1:
            return None
        return decode_job(row)


def complete_job(job_id, result, completed_at, duration_ms):
    with closing(connect()) as connection:
        connection.execute(
            """
            UPDATE jobs
            SET status = 'completed', result = ?, error = NULL,
                completed_at = ?, duration_ms = ?
            WHERE id = ?
            """,
            (json.dumps(result), completed_at, duration_ms, job_id),
        )
        connection.commit()


def fail_job(job_id, error, completed_at, duration_ms):
    with closing(connect()) as connection:
        connection.execute(
            """
            UPDATE jobs
            SET status = 'failed', error = ?, completed_at = ?, duration_ms = ?
            WHERE id = ?
            """,
            (error, completed_at, duration_ms, job_id),
        )
        connection.commit()


def get_metrics():
    counts = {
        "queued": 0,
        "processing": 0,
        "completed": 0,
        "failed": 0,
    }

    with closing(connect()) as connection:
        rows = connection.execute(
            "SELECT status, COUNT(*) AS count FROM jobs GROUP BY status"
        ).fetchall()
        average_row = connection.execute(
            """
            SELECT AVG(duration_ms) AS average_duration_ms
            FROM jobs
            WHERE status = 'completed'
            """
        ).fetchone()

    for row in rows:
        counts[row["status"]] = row["count"]

    average = average_row["average_duration_ms"] or 0
    return {
        **counts,
        "total": sum(counts.values()),
        "average_completed_duration_ms": round(average, 2),
    }

import json
import time
import urllib.request
from concurrent.futures import ThreadPoolExecutor

BASE_URL = "http://127.0.0.1:8000"
JOB_COUNT = 20
MAX_CLIENTS = 5


def read_json(url):
    with urllib.request.urlopen(url, timeout=5) as response:
        return json.loads(response.read().decode("utf-8"))


def submit_job(index):
    body = {
        "task_type": "sum",
        "payload": {"numbers": [index, index + 1, index + 2]},
    }
    request = urllib.request.Request(
        url=f"{BASE_URL}/jobs",
        data=json.dumps(body).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=5) as response:
        return json.loads(response.read().decode("utf-8"))


def main():
    started = time.perf_counter()
    with ThreadPoolExecutor(max_workers=MAX_CLIENTS) as executor:
        jobs = list(executor.map(submit_job, range(JOB_COUNT)))
    submitted_at = time.perf_counter()

    deadline = time.perf_counter() + 30
    while True:
        metrics = read_json(f"{BASE_URL}/metrics")
        if metrics["queued"] == 0 and metrics["processing"] == 0:
            break
        if time.perf_counter() >= deadline:
            raise TimeoutError("The queue did not drain within 30 seconds")
        time.sleep(0.25)

    finished = time.perf_counter()
    print(
        json.dumps(
            {
                "accepted_jobs": len(jobs),
                "submission_seconds": round(submitted_at - started, 3),
                "completed_in_seconds": round(finished - started, 3),
                "final_metrics": metrics,
            },
            indent=2,
        )
    )


if __name__ == "__main__":
    main()

from datetime import datetime, timezone
from uuid import uuid4

from fastapi import FastAPI, HTTPException

from db import get_job, get_metrics, init_db, insert_job

app = FastAPI(
    title="QueueForge",
    description="A durable asynchronous job processing API.",
    version="1.0.0",
)

init_db()


def utc_now():
    return datetime.now(timezone.utc).isoformat()


def validate_job_request(job):
    if job.get("task_type") != "sum":
        raise HTTPException(status_code=400, detail="task_type must be 'sum'")

    payload = job.get("payload")
    if not isinstance(payload, dict):
        raise HTTPException(status_code=400, detail="payload must be an object")

    numbers = payload.get("numbers")
    if not isinstance(numbers, list) or not numbers:
        raise HTTPException(
            status_code=400,
            detail="payload.numbers must be a non-empty list",
        )

    valid_numbers = all(
        isinstance(number, (int, float)) and not isinstance(number, bool)
        for number in numbers
    )
    if not valid_numbers:
        raise HTTPException(
            status_code=400,
            detail="payload.numbers must contain only numbers",
        )


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


@app.post("/jobs", status_code=202)
def create_job(job: dict):
    validate_job_request(job)
    record = {
        "id": str(uuid4()),
        "task_type": job["task_type"],
        "payload": job["payload"],
        "status": "queued",
        "created_at": utc_now(),
    }
    insert_job(record)
    return get_job(record["id"])


@app.get("/jobs/{job_id}")
def read_job(job_id: str):
    job = get_job(job_id)
    if job is None:
        raise HTTPException(status_code=404, detail="Job not found")
    return job


@app.get("/metrics")
def metrics():
    return get_metrics()

fastapi[standard]==0.162.0

import time
from datetime import datetime, timezone

from db import claim_next_job, complete_job, fail_job, init_db

POLL_INTERVAL_SECONDS = 0.5
SIMULATED_WORK_SECONDS = 0.4


def utc_now():
    return datetime.now(timezone.utc).isoformat()


def process_job(job):
    payload = job["payload"]
    if payload.get("force_failure") is True:
        raise RuntimeError("Forced failure requested for demonstration")

    time.sleep(SIMULATED_WORK_SECONDS)
    numbers = payload["numbers"]
    return {"sum": sum(numbers), "count": len(numbers)}


def run_worker():
    init_db()
    print("QueueForge worker started. Press Ctrl+C to stop.")

    while True:
        job = claim_next_job(utc_now())
        if job is None:
            time.sleep(POLL_INTERVAL_SECONDS)
            continue

        print(f"Processing {job['id']}")
        started = time.perf_counter()
        try:
            result = process_job(job)
            duration_ms = (time.perf_counter() - started) * 1000
            complete_job(job["id"], result, utc_now(), duration_ms)
            print(f"Completed {job['id']}")
        except Exception as error:
            duration_ms = (time.perf_counter() - started) * 1000
            fail_job(job["id"], str(error), utc_now(), duration_ms)
            print(f"Failed {job['id']}: {error}")


if __name__ == "__main__":
    run_worker()
Publish the public repository

Publishing makes the source plus the measured design available from one link. The ignored database remains local inside the Codespace.

Before you publish, which project files should appear on GitHub after the upload completes?

  • Return to the Source Control view in the Visual Studio Code web client.
  • Confirm that the pending source includes README.md, .gitignore, db.py, main.py, worker.py, load_test.py, and requirements.txt.
  • Confirm that queueforge.db is absent from the pending source.
  • Select Publish to GitHub public repository in the Source Control view.
  • Enter queueforge when GitHub asks for the repository name.
  • Follow the repository link shown after publication finishes.

You will see the public queueforge repository with the README rendered on its main page. The source files plus requirements.txt are visible while queueforge.db remains excluded.

Repository publication failed?

Confirm that your GitHub session is still active. Return to the Source Control view after signing in again.

Check that the repository name is queueforge. Confirm that you selected the public repository publishing option.

Help me troubleshoot publishing my Codespace as a public QueueForge repository.

Strong finish. QueueForge now has measurable burst-test evidence plus a public explanation of its reliability decisions and scaling boundary.

Secret mission

Benchmark Two Workers

Your first burst measured QueueForge with one worker. Run the same 20-job, five-client test with a second worker. Use the timing difference to document SQLite's horizontal scaling limit.

Clean Up Your Resources

Clean Up Your Resources

QueueForge remains at $0 while your GitHub Codespaces usage stays within the GitHub Free allowance. Decide whether to keep the workspace available, stop it to preserve included compute, or delete it to remove future storage use.

Resources you used:

  • The GitHub Codespaces workspace that contains your QueueForge source files.
  • The ignored queueforge.db SQLite file stored inside that Codespace.

Your GitHub account and public queueforge repository stay in place in every option below.

Keep everything running

No action needed if you are actively testing QueueForge or extending its worker design. The existing Codespace continues to use included storage while it remains available.

  • Leave the Codespace available with queueforge.db intact.
  • Keep your public queueforge repository available with the latest benchmark evidence.

Pause - I'll come back to this later

Stopping the Codespace ends active compute while keeping its files ready for your return. Closing the browser tab alone does not stop it.

  • Visit the Your codespaces page.
  • Open the Codespace menu for the workspace that contains QueueForge.
  • Select Stop codespace.

Your workspace is paused now. Its source files and queueforge.db remain available for your next session.

Delete - I don't want to use this again

Deleting the Codespace removes the browser workspace plus its local queueforge.db. Your public queueforge repository remains on GitHub with the source and benchmark evidence.

  • Return to your public queueforge repository on GitHub to confirm the updated README.md comparison is visible.
  • Visit the Your codespaces page.
  • Open the Codespace menu for the workspace that contains QueueForge.
  • Select Delete.

You should no longer see that Codespace on the Your codespaces page.

Cleanup complete. The removed workspace can no longer consume included compute or storage.

Nice Work!

Nice Work!

Outstanding work! You've built QueueForge as a durable job-processing system backed by an independent worker process.

You've learned how to:

  • Built a REST API with FastAPI that returns trackable job IDs immediately. Replaced volatile memory with durable SQLite storage so jobs survive API restarts.
  • Separated request handling from background processing with a dedicated worker process. Used transactions with BEGIN IMMEDIATE to keep two local workers from claiming the same queued row. Recorded terminal success or failure without crashing the API.
  • Measured burst-test evidence from 20 jobs across five concurrent clients. Published the results in a public GitHub repository with the next scaling bottleneck clearly documented.
  • Secret Mission: Benchmarked one worker against two workers without changing JOB_COUNT or MAX_CLIENTS. Compared the observed completed_in_seconds results to test horizontal worker scaling against SQLite's shared-write limit.

Ready to quiz yourself?