Build a Background Job Processing API

Build a REST API that queues and processes text analysis jobs in the background.

Introduction

30 Second Summary

When you upload a video to YouTube or ask ChatGPT a complex question, the platform doesn't make you wait on a loading screen until it's done. It hands you a ticket, processes your request in the background, and lets you check back for the results.

In this project, you will build a background job processing API that works exactly like those systems. You'll submit text for analysis, receive a job ID instantly, and poll back seconds later to find structured results waiting for you.

What You'll Build

You'll have a live API running on your machine where you paste any text into an interactive browser interface, get back a job ticket, and watch it transition from "pending" to "processing" to "completed" with rich analysis results.

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

  • A background job processing API that accepts any text, queues it for analysis, and returns word count, character frequency, sentence statistics, and readability metrics when you poll back.
  • Auto-generated interactive Swagger documentation you can demo live in your browser, submitting jobs and watching the full lifecycle from submission to completion.
  • A production-grade project structure with separate routers, services, models, and schemas that mirrors how professional backend teams organize real codebases.
  • Secret Mission: Add job priority levels and a cancel endpoint so high-priority jobs get processed first.

Are there any prerequisites?

This project assumes you have Python 3.11.9, PostgreSQL running locally, and Cursor installed on your machine. Step 1 walks you through creating your virtual environment and database, but the tools themselves should already be set up.

Before We Start

Before We Start...

Before we dive in, let's lock in what you're about to build and why it matters. This is your moment to commit to the goal so every step that follows has clear purpose.

Set Up Your Environment

Before writing any API code, you need a working Python environment with the right packages installed and a PostgreSQL database ready to accept connections.

In this step, you'll create an isolated project folder, install the exact package versions your API needs, and spin up a PostgreSQL database that will store your jobs.

In this step, get ready to:
  • Create a virtual environment and install project dependencies.
  • Create a PostgreSQL database for storing jobs.
  • Confirm everything works by running a minimal FastAPI server.
Create your project folder and virtual environment

A virtual environment gives your project its own isolated set of Python packages. This prevents version conflicts with other projects on your machine.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir job_processor
cd job_processor
  • Confirm you're in the right folder by running:
pwd

You should see a path ending in job_processor.

  • Create a virtual environment by running:
python -m venv venv

This creates a venv folder inside your project that holds isolated Python packages.

  • Activate the virtual environment:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt. This confirms you're working inside the virtual environment.

Don't see (venv) in your prompt?

On Windows, you may need to allow script execution first. Run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned in PowerShell, then try the activation command again.

On macOS, make sure you typed source before the path. Without it, the command won't activate anything.

help me activate my Python virtual environment

Now you'll create the file that lists your project's dependencies with exact version pins.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Cursor and press Enter to open it.
  • In Cursor, click File in the top menu bar.
  • Click Open Folder.
  • Navigate to Desktop, select the job_processor folder, and click Open.
  • Create a new file by clicking File then New File.
  • Save it immediately as requirements.txt by pressing Cmd+S (macOS) or Ctrl+S (Windows).
  • Paste the following content into requirements.txt:
fastapi[standard]==0.140.0
sqlalchemy==2.0.51
psycopg2-binary==2.9.12

What are these packages?

  • fastapi[standard] is the web framework that runs your API. The [standard] extra includes uvicorn, the server that actually serves your app.
  • sqlalchemy is the ORM (Object Relational Mapper) that lets you work with database tables as Python objects.
  • psycopg2-binary is the PostgreSQL driver that SQLAlchemy uses to talk to your database.
  • Save requirements.txt with Cmd+S (macOS) or Ctrl+S (Windows).
  • Open Cursor's integrated terminal by pressing Ctrl+` (backtick key, both platforms).
  • If your virtual environment is not active in this terminal (no (venv) in the prompt), activate it again using the command from earlier.
  • Install the packages by running:
pip install -r requirements.txt

You should see output ending with Successfully installed followed by the package names and versions.

Seeing an installation error?

Make sure (venv) appears in your terminal prompt. If it doesn't, you're installing to your system Python instead of the virtual environment.

If you see an error about psycopg2-binary, confirm that your Python version is 3.9 or higher by running python --version.

help me fix my pip install error

✔️ Awesome, I've got everything!

Great. Make sure you've saved requirements.txt and all three packages installed successfully.

ⓧ I'd like to double check the full code

Your requirements.txt should contain exactly this:

fastapi[standard]==0.140.0
sqlalchemy==2.0.51
psycopg2-binary==2.9.12
Create the PostgreSQL database

Your API needs a database to persist jobs. You'll create a dedicated PostgreSQL database called jobprocessor that will hold all job records.

  • In Cursor's integrated terminal, create the database by running:
createdb jobprocessor
  • Verify the database exists and accepts connections by running:
psql -d jobprocessor -c "SELECT 1;"

You should see a small table with a single column and the value 1. This confirms PostgreSQL is running and your database is ready.

Using a PostgreSQL password?

If your local PostgreSQL requires a username and password, note those credentials. You'll need them in a later step when you configure the database connection string.

Seeing 'command not found' or connection refused?

If createdb is not found, your PostgreSQL installation may not be in your system PATH. Check that PostgreSQL is installed and its bin directory is accessible.

If you see a connection error, PostgreSQL may not be running. Start the service using your system's service manager or the PostgreSQL app.

help me connect to my local PostgreSQL database

Verify the full setup

Everything is installed. Time to confirm that Python can import your packages and that FastAPI can actually serve a response.

  • Confirm FastAPI is installed by running:
python -c "import fastapi; print(fastapi.__version__)"

You should see 0.140.0 printed in your terminal.

  • Confirm SQLAlchemy is installed by running:
python -c "import sqlalchemy; print(sqlalchemy.__version__)"

You should see 2.0.51 printed.

Now you'll create a minimal API file to confirm the server can start and respond to requests.

  • In Cursor, create a new file called main.py by clicking File then New File, and saving it as main.py.
  • Paste the following code into main.py:
from fastapi import FastAPI

# Create the FastAPI application instance
app = FastAPI()

# A simple health check endpoint to confirm the server is running
@app.get("/health")
def health_check():
    return {"status": "ok"}

What does this code do?

  • FastAPI() creates your application object. Every endpoint you build later attaches to this.
  • @app.get("/health") registers a GET endpoint at the /health path. When someone hits this URL, FastAPI runs the function below it.
  • The function returns a dictionary, which FastAPI automatically converts to a JSON response.
  • Save main.py with Cmd+S (macOS) or Ctrl+S (Windows).

Before you run the server, what do you think you'll see in the terminal output? Will it show an error or a success message?

  • Start the FastAPI development server by running:
fastapi dev main.py

You should see output indicating the server is running at http://127.0.0.1:8000. The dev server auto-reloads when you change code, so you can leave it running.

  • Open your browser and navigate to http://127.0.0.1:8000/health.

You should see the JSON response {"status":"ok"} displayed in the browser. Your API is alive.

Server not starting?

Make sure (venv) is active in your terminal. The fastapi command comes from the package you installed in the virtual environment.

If you see a port conflict error, another process may be using port 8000. Stop any other dev servers you have running.

help me start my FastAPI development server

✔️ Awesome, I've got everything!

Great. Make sure main.py is saved and your server is running.

ⓧ I'd like to double check the full code

Your main.py should contain exactly this:

from fastapi import FastAPI

app = FastAPI()

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

Your environment is ready. You have a running FastAPI server, a connected PostgreSQL database, and all the packages you need. Next up, you'll build your first real API endpoint that accepts text analysis jobs.

Build Your First API Endpoint

Your environment is set up and your FastAPI app can serve a health check. Now it's time to build the real thing: endpoints that accept text input and hand back job tickets.

Before connecting a database, you'll use simple in-memory storage to confirm your routing and data validation work correctly. This way you'll see results immediately and catch structural issues early.

In this step, get ready to:
  • Define Pydantic models to validate incoming requests and shape outgoing responses.
  • Create a router with POST and GET endpoints using temporary in-memory storage.
  • Wire the router into your app and test both endpoints through Swagger UI.
Create the Pydantic schemas

FastAPI uses Pydantic models to validate request data and format responses automatically. You define the shape of what goes in and what comes out as Python classes.

  • Create a new file called schemas.py in your job_processor folder by pasting the following code:
from datetime import datetime
from pydantic import BaseModel


# Defines what the client sends when creating a job
class JobCreate(BaseModel):
    text_input: str


# Defines what the API returns for any job
class JobResponse(BaseModel):
    id: str
    text_input: str
    status: str
    created_at: datetime

What does this code do?

You've defined two Pydantic models that act as contracts for your API.

  • JobCreate defines what a client must send: just a text_input string. Pydantic rejects any request missing this field.
  • JobResponse defines what the API sends back: the job's ID, the original text, a status, and a timestamp. FastAPI uses this to format every response consistently.
  • Save schemas.py.
  • Confirm the file exists by running:
ls schemas.py

You should see schemas.py printed in the output.

Don't see schemas.py?

  • Make sure you saved the file in the job_processor directory, not a subfolder or your Desktop.
  • Check that your terminal is in the job_processor folder by running pwd.
  • help me find where schemas.py was saved

✔️ Awesome, I've got everything!

Great. Make sure you've saved schemas.py before moving on.

ⓧ I'd like to double check the full code

from datetime import datetime
from pydantic import BaseModel


class JobCreate(BaseModel):
    text_input: str


class JobResponse(BaseModel):
    id: str
    text_input: str
    status: str
    created_at: datetime
Create the router with your first endpoints

Routers group related endpoints together so your app stays organized. Your jobs router will handle all requests under the /jobs/ path.

  • Create a new file called router.py in your job_processor folder and paste the following code for the POST endpoint:
from fastapi import APIRouter, HTTPException
from datetime import datetime
import uuid
from schemas import JobCreate, JobResponse

# Creates a router that groups all /jobs/ endpoints together
router = APIRouter(prefix="/jobs", tags=["Jobs"])

# Temporary in-memory storage (replaced by PostgreSQL later)
jobs_db = {}


@router.post("/", response_model=JobResponse, status_code=201)
def create_job(job_data: JobCreate):
    # Generate a unique ID for this job
    job_id = str(uuid.uuid4())
    job = {
        "id": job_id,
        "text_input": job_data.text_input,
        "status": "pending",
        "created_at": datetime.now(),
    }
    # Store the job in our temporary dictionary
    jobs_db[job_id] = job
    return job

What does this code do?

  • APIRouter(prefix="/jobs") means every endpoint in this file automatically gets /jobs/ prepended to its path.
  • jobs_db is a plain Python dictionary acting as temporary storage. Jobs live here until you wire up PostgreSQL in the next step.
  • The create_job function generates a UUID, builds a job dictionary with a "pending" status, stores it, and returns it. FastAPI automatically validates the input against JobCreate and formats the response using JobResponse.
  • Now add the GET endpoint below the create_job function in the same file:
@router.get("/{job_id}", response_model=JobResponse)
def get_job(job_id: str):
    # Look up the job by its ID
    job = jobs_db.get(job_id)
    if not job:
        raise HTTPException(status_code=404, detail="Job not found")
    return job

What does this code do?

The get_job function takes a job_id from the URL path, looks it up in the dictionary, and returns it. If the ID doesn't exist, it raises a 404 error with a clear message.

  • Save router.py.
  • Confirm the file exists by running:
ls router.py

You should see router.py printed in the output.

Don't see router.py?

  • Make sure you created the file inside the job_processor directory.
  • Confirm your terminal's location matches the file location by running pwd.
  • help me find where router.py was saved

✔️ Awesome, I've got everything!

Great. Make sure you've saved router.py before moving on.

ⓧ I'd like to double check the full code

from fastapi import APIRouter, HTTPException
from datetime import datetime
import uuid
from schemas import JobCreate, JobResponse

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

jobs_db = {}


@router.post("/", response_model=JobResponse, status_code=201)
def create_job(job_data: JobCreate):
    job_id = str(uuid.uuid4())
    job = {
        "id": job_id,
        "text_input": job_data.text_input,
        "status": "pending",
        "created_at": datetime.now(),
    }
    jobs_db[job_id] = job
    return job


@router.get("/{job_id}", response_model=JobResponse)
def get_job(job_id: str):
    job = jobs_db.get(job_id)
    if not job:
        raise HTTPException(status_code=404, detail="Job not found")
    return job
Wire the router into your app and test

Your router exists, but FastAPI doesn't know about it yet. You need to register it with your app so the endpoints become reachable.

  • Open main.py in Cursor.
  • Find this line at the top:
from fastapi import FastAPI
  • Add a new import line directly below it, so the top of your file reads:
from fastapi import FastAPI
from router import router
  • Now find the line app = FastAPI() and add the router registration directly below it:
app = FastAPI()

app.include_router(router)

What does this do?

app.include_router(router) tells FastAPI to register all the endpoints defined in router.py. After this, your app responds to POST /jobs/ and GET /jobs/{job_id} in addition to the health check.

  • Save main.py.
  • Start the development server by running:
fastapi dev main.py

You should see output indicating the server is running at http://127.0.0.1:8000.

Before you open the browser, predict: what endpoints do you think will appear in the Swagger docs? How many should there be?

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

You should see three endpoints listed: POST /jobs/, GET /jobs/{job_id}, and GET /health. The Swagger UI is auto-generated by FastAPI from your code.

  • Click the POST /jobs/ endpoint to expand it.
  • Click Try it out.
  • Replace the request body with this JSON:
{
  "text_input": "FastAPI is a modern web framework for building APIs with Python."
}
  • Click Execute.

You should see a 201 response with a JSON body containing your job's unique id, the text_input you submitted, a "pending" status, and a created_at timestamp.

Seeing an error instead?

  • If you see a ModuleNotFoundError, confirm that schemas.py and router.py are in the same folder as main.py.
  • If the server won't start, check that from router import router matches the filename router.py exactly.
  • If you see a 422 validation error, make sure your request body JSON has the exact key text_input (not text or input).
  • my FastAPI server won't start or I'm getting an import error

✔️ Awesome, I've got everything!

Great. Make sure you've saved all three files before moving on.

ⓧ I'd like to double check the full code

from fastapi import FastAPI
from router import router

app = FastAPI()

app.include_router(router)


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

Your API can now accept text and hand back job tickets. Next up, you'll replace that temporary dictionary with a real PostgreSQL database so jobs survive server restarts.

Wire Up PostgreSQL

Your API works, but it has a critical weakness. Every time you stop the server, all submitted jobs vanish.

Real job systems need persistent storage so jobs survive crashes and restarts. In this step, you'll connect your API to PostgreSQL so data lives in a real database. This reveals an interesting gap: jobs persist now, but they sit at "pending" forever because nothing processes them yet.

In this step, get ready to:
  • Create the database connection layer with SQLAlchemy.
  • Define a Job model that maps to a PostgreSQL table.
  • Replace in-memory storage with real database queries.
Create the database layer

Your API needs a way to talk to PostgreSQL. SQLAlchemy handles this through an engine (the connection), a session factory (one session per request), and a base class (the foundation for all your models).

  • Create a new file called database.py in your job_processor folder and add the following code:
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

DATABASE_URL = "postgresql://postgres:your-password-here@localhost:5432/jobprocessor"

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)


class Base(DeclarativeBase):
    pass


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


def create_tables():
    Base.metadata.create_all(bind=engine)

What does this code do?

  • DATABASE_URL tells SQLAlchemy where your PostgreSQL database lives. Replace your-password-here with your actual PostgreSQL password, or remove it entirely if your local PostgreSQL doesn't require one.
  • SessionLocal creates database sessions. Each API request gets its own session so operations don't interfere with each other.
  • get_db() is a dependency function that FastAPI calls automatically. It opens a session, hands it to your endpoint, then closes it when the request finishes.
  • create_tables() reads all your model definitions and creates matching tables in PostgreSQL.

No password on local PostgreSQL?

If your local PostgreSQL uses trust authentication (common on macOS with Homebrew), change the URL to postgresql://postgres@localhost:5432/jobprocessor with no password portion.

  • Save database.py.
  • Verify the module loads correctly by running this in your terminal:
python -c "from database import get_db, create_tables; print('Database module ready')"

You should see Database module ready printed with no errors.

Seeing an import error?

  • Make sure your virtual environment is activated. You should see (venv) in your terminal prompt.
  • Check that database.py is saved in the same job_processor folder where your other files live.
  • If you see a connection error, your PostgreSQL credentials or database name may be wrong. Verify you can connect with psql -d jobprocessor.

help me fix my database.py import error

✔️ Awesome, I've got everything!

Great. Make sure you've saved database.py before moving on.

ⓧ I'd like to double check the full code

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

DATABASE_URL = "postgresql://postgres:your-password-here@localhost:5432/jobprocessor"

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)


class Base(DeclarativeBase):
    pass


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


def create_tables():
    Base.metadata.create_all(bind=engine)
Create the Job model

With the database connection ready, you need a model that tells SQLAlchemy what your jobs table looks like. Each column maps to a Python attribute on the Job class.

  • Create a new file called models.py in your job_processor folder and add the following code:
import uuid
from datetime import datetime
from sqlalchemy import String, Text, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column
from database import Base


class Job(Base):
    __tablename__ = "jobs"

    id: Mapped[str] = mapped_column(
        String(36), primary_key=True, default=lambda: str(uuid.uuid4())
    )
    text_input: Mapped[str] = mapped_column(Text, nullable=False)
    status: Mapped[str] = mapped_column(String(20), default="pending")
    result: Mapped[str | None] = mapped_column(Text, nullable=True)
    error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now()
    )
    completed_at: Mapped[datetime | None] = mapped_column(
        DateTime, nullable=True
    )

What does this code do?

  • __tablename__ = "jobs" tells SQLAlchemy to create a table called jobs in PostgreSQL.
  • id generates a unique UUID string for each job automatically. This becomes the job ticket your users get back.
  • status defaults to "pending" when a new job is created. Later, a processor will move it through "processing" to "completed".
  • server_default=func.now() makes PostgreSQL set the creation timestamp on the database side, so the time is always accurate.
  • result and error_message are nullable because they only get filled once a job finishes processing.
  • Save models.py.
  • Verify the model loads correctly by running:
python -c "from models import Job; print(Job.__tablename__)"

You should see jobs printed in your terminal.

Seeing a circular import error?

  • Make sure models.py imports Base from database (not the other way around). The import direction is models.py → database.py.
  • Double check that database.py does not import anything from models.py.

help me fix my models.py import error

✔️ Awesome, I've got everything!

Great. Make sure you've saved models.py before moving on.

ⓧ I'd like to double check the full code

import uuid
from datetime import datetime
from sqlalchemy import String, Text, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column
from database import Base


class Job(Base):
    __tablename__ = "jobs"

    id: Mapped[str] = mapped_column(
        String(36), primary_key=True, default=lambda: str(uuid.uuid4())
    )
    text_input: Mapped[str] = mapped_column(Text, nullable=False)
    status: Mapped[str] = mapped_column(String(20), default="pending")
    result: Mapped[str | None] = mapped_column(Text, nullable=True)
    error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now()
    )
    completed_at: Mapped[datetime | None] = mapped_column(
        DateTime, nullable=True
    )
Update the router to use the real database

You now have the database layer and the model. It's time to swap out the in-memory dictionary for real PostgreSQL queries. This requires changes to three files: main.py (to create tables on startup), schemas.py (to support ORM objects), and router.py (to use actual database sessions).

  • Replace the entire contents of main.py with the following code:
from contextlib import asynccontextmanager
from fastapi import FastAPI
from database import create_tables
from router import router


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


app = FastAPI(lifespan=lifespan)

app.include_router(router)

What is the lifespan pattern?

The lifespan function runs setup code (before yield) when the server starts and cleanup code (after yield) when it stops. This is the officially recommended pattern in FastAPI for startup and shutdown logic.

Here, create_tables() runs at startup and ensures your jobs table exists in PostgreSQL before any requests arrive.

  • Save main.py.
  • Start the server to confirm tables get created by running:
fastapi dev main.py

The server starts and your jobs table is now created in PostgreSQL. The API still works with the in-memory router for now.

  • Stop the server with Ctrl+C.
  • Now replace the entire contents of schemas.py with the following code:
from datetime import datetime
from pydantic import BaseModel


class JobCreate(BaseModel):
    text_input: str


class JobResponse(BaseModel):
    id: str
    text_input: str
    status: str
    result: str | None = None
    error_message: str | None = None
    created_at: datetime
    completed_at: datetime | None = None

    model_config = {"from_attributes": True}

What changed in schemas.py?

  • model_config = {"from_attributes": True} tells Pydantic to read data directly from SQLAlchemy model attributes instead of requiring a plain dictionary. Without this, FastAPI would crash when trying to serialize your database objects.
  • The new optional fields (result, error_message, completed_at) match the columns you defined in your Job model. They default to None until a job finishes processing.
  • Save schemas.py.
  • Now replace the entire contents of router.py with the following code:
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db
from models import Job
from schemas import JobCreate, JobResponse

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


@router.post("/", response_model=JobResponse, status_code=201)
def create_job(job_data: JobCreate, db: Session = Depends(get_db)):
    job = Job(text_input=job_data.text_input)
    db.add(job)
    db.commit()
    db.refresh(job)
    return job


@router.get("/{job_id}", response_model=JobResponse)
def get_job(job_id: str, db: Session = Depends(get_db)):
    job = db.query(Job).filter(Job.id == job_id).first()
    if not job:
        raise HTTPException(status_code=404, detail="Job not found")
    return job

What changed in router.py?

  • db: Session = Depends(get_db) uses FastAPI's dependency injection to automatically provide a fresh database session for each request. When the request ends, the session closes itself.
  • Instead of storing jobs in a Python dictionary, db.add(job) inserts a row into the jobs table. db.commit() saves it permanently. db.refresh(job) reloads the row so you get the server-generated values (like created_at).
  • db.query(Job).filter(Job.id == job_id).first() translates to a SQL SELECT query that finds one job by its UUID.
  • Save router.py.
  • Start the server again by running:
fastapi dev main.py

The server should start without errors. Open http://127.0.0.1:8000/docs in your browser to see the Swagger UI.

Server won't start?

  • If you see a connection error mentioning PostgreSQL, double check that your DATABASE_URL in database.py has the correct password and that the jobprocessor database exists.
  • If you see an import error, confirm all five files (database.py, models.py, schemas.py, router.py, main.py) are saved in the same job_processor directory.

help me fix my FastAPI server startup error with SQLAlchemy and PostgreSQL

Now let's prove that jobs actually persist across server restarts.

  • In the Swagger UI, expand the POST /jobs/ endpoint.
  • Click Try it out.
  • Replace the request body with this JSON:
{
  "text_input": "This is a test job to prove persistence works across restarts."
}
  • Click Execute.
  • Copy the id value from the response. It looks like a UUID such as a1b2c3d4-e5f6-7890-abcd-ef1234567890.
  • Go back to your terminal and stop the server with Ctrl+C.
  • Start the server again by running:
fastapi dev main.py

Before you test, what do you think will happen when you request the same job ID after a full server restart? Will the job still be there, or will it be gone like before?

  • In the Swagger UI, expand the GET /jobs/{job_id} endpoint.
  • Click Try it out.
  • Paste the job ID you copied earlier into the job_id field.
  • Click Execute.

Your job is still there. The data survived the restart because it lives in PostgreSQL now, not in Python's memory. But look at the status field. It says "pending". It will stay "pending" forever because nothing in your system actually processes the job.

Why is the job stuck at pending?

You've built the storage layer, but not the processing engine. Right now, your API is like a restaurant that takes orders but has no kitchen. The order ticket exists (your job row in PostgreSQL), but nobody picks it up and cooks it.

This is exactly how production job systems start. The next step adds the background processor that moves jobs from "pending" through "processing" to "completed".

✔️ Awesome, I've got everything!

Great. Confirm all files are saved and the server is running with no errors.

ⓧ I'd like to double check the full code

Here are all 5 files as they should look now. Compare each one:

database.py:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

DATABASE_URL = "postgresql://postgres:your-password-here@localhost:5432/jobprocessor"

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)


class Base(DeclarativeBase):
    pass


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


def create_tables():
    Base.metadata.create_all(bind=engine)

models.py:

import uuid
from datetime import datetime
from sqlalchemy import String, Text, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column
from database import Base


class Job(Base):
    __tablename__ = "jobs"

    id: Mapped[str] = mapped_column(
        String(36), primary_key=True, default=lambda: str(uuid.uuid4())
    )
    text_input: Mapped[str] = mapped_column(Text, nullable=False)
    status: Mapped[str] = mapped_column(String(20), default="pending")
    result: Mapped[str | None] = mapped_column(Text, nullable=True)
    error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now()
    )
    completed_at: Mapped[datetime | None] = mapped_column(
        DateTime, nullable=True
    )

schemas.py:

from datetime import datetime
from pydantic import BaseModel


class JobCreate(BaseModel):
    text_input: str


class JobResponse(BaseModel):
    id: str
    text_input: str
    status: str
    result: str | None = None
    error_message: str | None = None
    created_at: datetime
    completed_at: datetime | None = None

    model_config = {"from_attributes": True}

router.py:

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db
from models import Job
from schemas import JobCreate, JobResponse

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


@router.post("/", response_model=JobResponse, status_code=201)
def create_job(job_data: JobCreate, db: Session = Depends(get_db)):
    job = Job(text_input=job_data.text_input)
    db.add(job)
    db.commit()
    db.refresh(job)
    return job


@router.get("/{job_id}", response_model=JobResponse)
def get_job(job_id: str, db: Session = Depends(get_db)):
    job = db.query(Job).filter(Job.id == job_id).first()
    if not job:
        raise HTTPException(status_code=404, detail="Job not found")
    return job

main.py:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from database import create_tables
from router import router


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


app = FastAPI(lifespan=lifespan)

app.include_router(router)

Your jobs now survive server restarts. That's real persistence. But every job is stuck at "pending" forever because nothing processes them. In the next step, you'll build the background processing engine that picks up pending jobs, analyzes the text, and fills in the results.

Add Background Processing

Your jobs survive server restarts now, but there's an obvious problem. Every job you submit sits at "pending" forever because nothing actually processes it.

FastAPI's BackgroundTasks lets you kick off processing after returning the response to the client. The API returns HTTP 201 immediately with the job ticket, then does the heavy work asynchronously. In this step, you'll build the processing engine that transitions jobs through their states.

In this step, get ready to:
  • Write a text analysis function that calculates word count, character frequency, and sentence statistics.
  • Build a background job processor that manages the job state machine.
  • Wire the processor into your router so jobs process automatically after submission.
Create the text analysis function

The processing engine needs a function that actually does something useful with the submitted text. You'll write a pure Python function that extracts word count, character frequency, sentence statistics, and readability metrics.

  • Create a new file called services.py in your job_processor folder.
  • Add the imports and the analyze_text function by pasting this code:
import json
import time
from datetime import datetime
from database import SessionLocal
from models import Job


def analyze_text(text: str) -> dict:
    # Split text into words and count basics
    words = text.split()
    word_count = len(words)
    char_count = len(text)
    sentence_count = text.count(".") + text.count("!") + text.count("?")
    if sentence_count == 0:
        sentence_count = 1

    # Calculate averages for readability metrics
    avg_word_length = round(char_count / max(word_count, 1), 2)
    avg_sentence_length = round(word_count / sentence_count, 2)

    # Count frequency of each alphabetic character
    char_frequency = {}
    for char in text.lower():
        if char.isalpha():
            char_frequency[char] = char_frequency.get(char, 0) + 1

    # Keep only the top 5 most frequent characters
    top_chars = sorted(
        char_frequency.items(), key=lambda x: x[1], reverse=True
    )[:5]

What does this code do?

The imports at the top include json, time, and SessionLocal. You'll use these in the background processor you write next.

The analyze_text function takes a string and extracts six metrics from it: word count, character count, sentence count, average word length, average sentence length, and the top 5 most frequent alphabetic characters.

  • Now add the return statement directly below the code you just pasted:
    # Return all metrics as a dictionary
    return {
        "word_count": word_count,
        "character_count": char_count,
        "sentence_count": sentence_count,
        "average_word_length": avg_word_length,
        "average_sentence_length": avg_sentence_length,
        "top_characters": dict(top_chars),
    }
  • Save services.py.
  • Test the function works by running this command in your terminal:
python -c "from services import analyze_text; print(analyze_text('Hello world. How are you today?'))"

You should see a dictionary printed with keys like word_count, character_count, and top_characters with numeric values.

Seeing an ImportError?

  • Make sure your virtual environment is activated. You should see (venv) in your terminal prompt.
  • Confirm you're running the command from inside the job_processor folder where services.py lives.
  • Check that services.py is saved with no typos in the function name analyze_text.

Why can't Python find my analyze_text function?

Create the background job processor

The analysis function works, but it needs something to call it on actual jobs from the database. The process_job function will manage the full lifecycle: fetch the job, update its status, run the analysis, and save the results.

  • In services.py, add the process_job function below your analyze_text function:
def process_job(job_id: str):
    # Open a fresh DB session (the request session is already closed)
    db = SessionLocal()
    try:
        job = db.query(Job).filter(Job.id == job_id).first()
        if not job:
            return

        # Mark the job as processing so clients can see the transition
        job.status = "processing"
        db.commit()

        # Simulate slow processing so you can observe the state change
        time.sleep(2)

        # Run the text analysis and save results as JSON
        result = analyze_text(job.text_input)
        job.result = json.dumps(result)
        job.status = "completed"
        job.completed_at = datetime.now()
        db.commit()
    except Exception as e:
        db.rollback()
        job = db.query(Job).filter(Job.id == job_id).first()
        if job:
            job.status = "failed"
            job.error_message = str(e)
            db.commit()
    finally:
        db.close()

What does this code do?

  • process_job creates its own database session with SessionLocal(). Background tasks run after the HTTP response is sent, so the request-scoped session is already closed by then.
  • time.sleep(2) simulates a slow process. Without it, jobs complete instantly and you'd never see the "processing" state when polling.
  • The try/except/finally block ensures that if anything goes wrong, the job gets marked as "failed" with the error message saved. The session always closes cleanly.

Why does process_job open its own database session?

Your router's get_db dependency creates a session that lives only for the duration of the HTTP request. By the time the background task starts running, that request is done and the session is closed.

That's why process_job calls SessionLocal() directly to get its own independent session.

  • Save services.py.

✔️ Awesome, I've got everything!

Great. Confirm you've saved services.py with both analyze_text and process_job present.

ⓧ I'd like to double check the full code

Your services.py should have, in order:

  • The import block at the top: json, time, datetime, SessionLocal, and Job.
  • The analyze_text function (splits text, counts words/characters/sentences, calculates averages, finds top 5 characters, returns a dictionary).
  • The process_job function (opens a session, sets status to "processing", sleeps 2 seconds, runs analyze_text, saves results, handles errors).

Scroll up and confirm each chunk is present in your file.

Wire BackgroundTasks into the router

The processing engine is ready, but the router doesn't know about it yet. You need to tell FastAPI to call process_job in the background every time a new job is created.

  • In router.py, find the imports at the top of the file:
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db
from models import Job
from schemas import JobCreate, JobResponse
  • Replace the imports with this updated version that adds BackgroundTasks and imports process_job:
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db
from models import Job
from schemas import JobCreate, JobResponse
from services import process_job
  • Now find the create_job function:
@router.post("/", response_model=JobResponse, status_code=201)
def create_job(job_data: JobCreate, db: Session = Depends(get_db)):
    job = Job(text_input=job_data.text_input)
    db.add(job)
    db.commit()
    db.refresh(job)
    return job
  • Replace the entire create_job function with this updated version that adds background processing:
@router.post("/", response_model=JobResponse, status_code=201)
def create_job(
    job_data: JobCreate,
    background_tasks: BackgroundTasks,
    db: Session = Depends(get_db),
):
    job = Job(text_input=job_data.text_input)
    db.add(job)
    db.commit()
    db.refresh(job)

    # Queue the job for background processing
    background_tasks.add_task(process_job, job.id)

    return job

What changed here?

  • background_tasks: BackgroundTasks is a parameter that FastAPI injects automatically. You don't need to create it yourself.
  • background_tasks.add_task(process_job, job.id) schedules process_job to run after the response is sent back to the client. The client gets their job ID immediately without waiting.
  • Save router.py.

✔️ Awesome, I've got everything!

Great. Confirm you've saved router.py with the updated imports and the new create_job function.

ⓧ I'd like to double check the full code

Your router.py should have, in order:

  • The updated imports with BackgroundTasks and from services import process_job.
  • The router creation line: router = APIRouter(prefix="/jobs", tags=["Jobs"]).
  • The updated create_job with the background_tasks parameter and add_task call.
  • The unchanged get_job function at the bottom.

Scroll up and confirm each section is present in your file.

Now it's time to test the full job lifecycle. Before you run this, think about what you expect to happen. If you POST a job and immediately GET it, will it be "completed" already, or still in progress?

  • Start the development server by running:
fastapi dev main.py
  • Open the Swagger UI at http://127.0.0.1:8000/docs in your browser.
  • Expand the POST /jobs/ endpoint and click Try it out.
  • Replace the request body with this text:
{
  "text_input": "FastAPI makes it easy to build modern APIs. Background tasks handle the heavy lifting. This is a test of the job processing system."
}
  • Click Execute.

You'll see a 201 response with your job. The status shows "pending" because the response returns before processing starts.

  • Copy the id value from the response.
  • Immediately expand GET /jobs/{job_id}, click Try it out, paste the ID, and click Execute.

If you're fast enough (within 2 seconds), you'll catch the job with status "processing". This is the state machine in action.

  • Wait about 3 seconds, then click Execute on the GET endpoint again with the same job ID.

The status now shows "completed" and the result field contains a JSON string with your text analysis: word count, character count, sentence count, averages, and the top 5 characters.

Job stuck at "pending" forever?

  • Check your terminal for error messages. A common cause is a typo in services.py that crashes process_job before it can update the database.
  • Make sure from services import process_job appears in your router.py imports. A missing import means the background task never gets registered.
  • If the status jumps straight to "failed", check the error_message field in the GET response. It tells you exactly what went wrong inside process_job.

My job is stuck at pending or showing as failed. Help me debug.

Your background processing engine is alive. Jobs now transition through pending, processing, and completed automatically. Next up, you'll add proper validation and error handling so your API responds cleanly when clients send bad input.

Add Validation and Error Handling

Your API processes jobs end-to-end, but it has a blind spot. Right now, anyone can submit an empty string and your system happily creates a job, processes it, and returns meaningless results.

Production APIs return structured, helpful error responses that tell the client exactly what went wrong and how to fix it. In this step, you will add input validation, descriptive error messages, and a global safety net for unhandled exceptions.

In this step, get ready to:
  • Add input validation to the JobCreate schema so empty or oversized text is rejected.
  • Improve error responses with descriptive messages and a global exception handler.
  • Polish the API metadata so Swagger UI shows an organized, professional interface.
Add input validation to schemas

Before you fix anything, try the problem yourself. What happens when you submit empty text to your API?

  • Make sure your dev server is running. Start it by running:
fastapi dev main.py
  • Open Swagger UI at http://127.0.0.1:8000/docs in your browser.
  • Expand the POST /jobs/ endpoint and click Try it out.
  • Set the request body to:
{"text_input": ""}

Before you click Execute, predict what will happen. Will the API reject the empty string, or accept it?

  • Click Execute.

You get a 201 Created response. The API accepted the empty string and created a job with no meaningful text to analyze. That is the gap you are about to close.

Why is this a problem?

Without validation, garbage input flows through your entire system. The job gets queued, processed, and stored, wasting database space and compute time on results like a word count of 0.

Production APIs reject bad input at the gate with a clear error message so the client can fix it immediately.

Pydantic's Field lets you declare constraints directly on your model attributes. When input violates a constraint, FastAPI automatically returns a structured 422 response with field-level details.

  • In Cursor, open schemas.py and replace its entire contents with the following:
from datetime import datetime
from pydantic import BaseModel, Field


class JobCreate(BaseModel):
    # Reject empty strings and cap input at 10,000 characters
    text_input: str = Field(
        min_length=1,
        max_length=10000,
        description="The text content to analyze",
    )


class JobResponse(BaseModel):
    id: str
    text_input: str
    status: str
    result: str | None = None
    error_message: str | None = None
    created_at: datetime
    completed_at: datetime | None = None

    model_config = {"from_attributes": True}

What does this code do?

  • Field(min_length=1) rejects empty strings. The text must have at least one character.
  • Field(max_length=10000) caps input at 10,000 characters, preventing someone from submitting a massive payload.
  • The description parameter appears in Swagger UI, documenting what the field expects.
  • Save schemas.py.
  • Return to Swagger UI in your browser and refresh the page.
  • Try submitting the same empty string request body again:
{"text_input": ""}
  • Click Execute.

This time you get a 422 Unprocessable Entity response with a structured JSON body. The response includes the field location, an error message, and the type of validation that failed.

Still seeing 201 Created?

  • Make sure you saved schemas.py. The dev server auto-reloads on save.
  • Check that you imported Field from pydantic in your import line.
  • Refresh the Swagger UI page fully with Cmd+Shift+R (macOS) or Ctrl+Shift+R (Windows) to pick up the new schema.

Why is my validation not working?

Add proper HTTP error responses in the router

The validation handles bad input, but what about requests for jobs that do not exist? Your current 404 message just says "Job not found" with no context about which ID failed.

  • In Cursor, open router.py and find this line inside the get_job function:
        raise HTTPException(status_code=404, detail="Job not found")
  • Replace that line with a descriptive f-string message so the line now looks like this:
        raise HTTPException(
            status_code=404,
            detail=f"Job with id '{job_id}' not found",
        )

Why include the job ID in the error?

When a client receives a 404, knowing which ID failed helps them debug immediately. They can check for typos or confirm whether they are hitting the right endpoint. Generic error messages force the client to guess what went wrong.

  • Save router.py.
  • In Swagger UI, expand GET /jobs/{job_id} and click Try it out.
  • Enter a fake ID like not-a-real-id and click Execute.

You should see a 404 response with the message: Job with id 'not-a-real-id' not found.

Seeing a 500 error instead of 404?

  • Check your terminal for a syntax error in router.py. The f-string needs the f prefix before the opening quote.
  • Make sure the indentation matches the original raise statement exactly.

Why am I getting a 500 error instead of 404 when I request a non-existent job?

Polish the API metadata and add a global exception handler

Your validation catches bad input and your 404 is descriptive. The last piece is a safety net for unexpected crashes and professional metadata that makes your Swagger docs look polished.

  • In Cursor, open main.py and replace its entire contents with the following:
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from database import create_tables
from router import router


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


# Add title, description, and version for Swagger UI
app = FastAPI(
    title="Job Processing API",
    description="A background job processing system for text analysis",
    version="1.0.0",
    lifespan=lifespan,
)

app.include_router(router)

What does this code do?

  • The title, description, and version parameters populate Swagger UI's header, giving your API a professional appearance.
  • The Request and JSONResponse imports are needed for the global exception handler you will add next.
  • Save main.py and refresh Swagger UI in your browser.

The header now shows Job Processing API with the description and version 1.0.0 displayed.

Now add the global exception handler. This catches any unhandled exception and returns a clean JSON response instead of a raw stack trace.

  • In main.py, add the following code below the app.include_router(router) line:
# Catch any unhandled exception and return a clean 500 response
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={"detail": "An internal error occurred. Please try again later."},
    )

What does the exception handler do?

If any unhandled exception bubbles up during a request, this handler catches it and returns a clean 500 JSON response. Clients get a consistent error format without seeing your internal code or stack trace.

  • Save main.py.

Now run through the full demo flow to confirm everything works together. Before you test, predict: will the API handle all these cases cleanly?

  • In Swagger UI, submit a valid job with this request body:
{"text_input": "FastAPI makes building APIs fast and enjoyable. Validation keeps your data clean."}
  • Click Execute. You should see a 201 response with a job ID and pending status.
  • Wait 3 seconds, then use GET /jobs/{job_id} with that job ID. You should see completed status with a result field containing the text analysis.
  • Try POST /jobs/ with {"text_input": ""}. You should see a 422 validation error with a clear message about minimum length.
  • Try GET /jobs/{job_id} with a fake ID. You should see a clean 404 JSON response.

Server not reloading after saving?

  • Check your terminal for import errors. A typo in from fastapi.responses import JSONResponse will prevent the server from restarting.
  • If the server crashed, stop it with Ctrl+C and restart with fastapi dev main.py.

Why is my FastAPI server crashing after I updated main.py?

✔️ Awesome, I've got everything!

Great. Make sure you have saved schemas.py, router.py, and main.py.

ⓧ I'd like to double check the full code

Here is the full schemas.py:

from datetime import datetime
from pydantic import BaseModel, Field


class JobCreate(BaseModel):
    text_input: str = Field(
        min_length=1,
        max_length=10000,
        description="The text content to analyze",
    )


class JobResponse(BaseModel):
    id: str
    text_input: str
    status: str
    result: str | None = None
    error_message: str | None = None
    created_at: datetime
    completed_at: datetime | None = None

    model_config = {"from_attributes": True}

Here is the full router.py:

from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db
from models import Job
from schemas import JobCreate, JobResponse
from services import process_job

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


@router.post("/", response_model=JobResponse, status_code=201)
def create_job(
    job_data: JobCreate,
    background_tasks: BackgroundTasks,
    db: Session = Depends(get_db),
):
    job = Job(text_input=job_data.text_input)
    db.add(job)
    db.commit()
    db.refresh(job)

    background_tasks.add_task(process_job, job.id)

    return job
@router.get("/{job_id}", response_model=JobResponse)
def get_job(job_id: str, db: Session = Depends(get_db)):
    job = db.query(Job).filter(Job.id == job_id).first()
    if not job:
        raise HTTPException(
            status_code=404,
            detail=f"Job with id '{job_id}' not found",
        )
    return job

Here is the full main.py:

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from database import create_tables
from router import router


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


app = FastAPI(
    title="Job Processing API",
    description="A background job processing system for text analysis",
    version="1.0.0",
    lifespan=lifespan,
)

app.include_router(router)
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={"detail": "An internal error occurred. Please try again later."},
    )

Your API now rejects bad input with clear 422 errors, returns descriptive 404 messages, and has a global safety net for unexpected crashes. Next up, you will extend this into a Secret Mission with job priorities and cancellation.

Secret mission

Add Job Priority and Cancellation

Real job processing systems don't treat every job equally. Some need to jump the queue. Others need to be stopped before they start. Can you extend your API to handle both?

Clean Up Your Resources

Clean Up Your Resources

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

Resources you used:

  • The job_processor project folder on your Desktop (includes all Python files and the virtual environment).
  • The jobprocessor PostgreSQL database on your local machine.
  • pip packages installed inside the virtual environment (FastAPI, SQLAlchemy, psycopg2-binary).

Keep everything running

No action needed. Choose this if you're still actively building or want to experiment further with the API.

  • Your project folder, virtual environment, and database all stay in place.
  • To resume work later, open the job_processor folder in Cursor, activate the virtual environment, and run fastapi dev main.py.

Pause - I'll come back to this later

Shut down running processes to free up memory. Your files and database stay intact for when you return.

  • Stop the FastAPI dev server by pressing Ctrl+C in the terminal where it is running.
  • Deactivate the virtual environment by running:
deactivate
  • Your project folder and PostgreSQL database remain on disk. No data is lost.

Delete - I don't want to use this again

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

  • Stop the FastAPI dev server by pressing Ctrl+C if it is still running.
  • Deactivate the virtual environment by running:
deactivate
  • Drop the PostgreSQL database by running:
dropdb jobprocessor
  • Delete the project folder. On macOS/Linux, run:
rm -rf ~/Desktop/job_processor
  • On Windows, run:
rmdir /s /q %USERPROFILE%\Desktop\job_processor
  • That removes all Python files, the virtual environment, and installed packages in one step.

Nice Work!

Nice Work!

You just built a production-grade background job processing API from scratch. This is the exact architecture pattern behind systems like ChatGPT's async processing, video transcoding platforms, and data pipeline services.

You've learned how to:

  • Build a complete REST API with FastAPI that accepts text analysis jobs, processes them in the background with BackgroundTasks, and returns structured results through a job state machine.
  • Wire up PostgreSQL persistence with SQLAlchemy 2.0 using modern Mapped columns, dependency injection for database sessions, and the lifespan pattern for startup logic.
  • Add production-grade Pydantic v2 validation with structured error responses, global exception handling, and auto-generated Swagger documentation that serves as a live interactive demo.
  • Secret Mission: Added job priority levels, a cancel endpoint for pending jobs, and a filtered list endpoint with query parameters for status and priority.

Ready to quiz yourself?