Real-Time Service Queue API

Build a real-time queue API with WebSockets, background tasks, and state machines.

Introduction

30 Second Summary

Think about the last time you waited at a bank or clinic. Behind the counter, someone is managing a queue system that tracks who's next, who's being served, and who gave up and left.

In this project, you will build a real-time service queue API using FastAPI, PostgreSQL, and WebSockets. You'll create REST endpoints for managing tickets, broadcast live queue updates to connected clients, and run a background worker that detects abandoned tickets automatically.

What You'll Build

A production-patterned async API that combines real-time event broadcasting, periodic background processing, and domain state management.

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

  • REST endpoints for creating tickets, calling the next customer, and completing service.
  • A WebSocket endpoint that broadcasts live queue position updates to all connected clients.
  • A background worker running inside the FastAPI lifespan that automatically detects and marks abandoned tickets as no-shows.
  • A state machine enforcing valid ticket transitions that rejects invalid state changes with clear errors.
  • Secret Mission: An in-memory caching layer with TTL and cache invalidation for a dashboard statistics endpoint.

Are there any prerequisites?

This project assumes you're familiar with FastAPI basics like routing, Pydantic models, and dependency injection. You'll also need PostgreSQL running locally on your machine and experience using Python virtual environments.

Set Up the Project

Before writing any application logic, you need a solid foundation. This step sets up the project directory, installs all dependencies, and wires up the async database connection your service queue will rely on.

You will use FastAPI as the web framework, SQLAlchemy for async database access, and asyncpg as the native PostgreSQL driver.

In this step, get ready to:
  • Create the project directory and install dependencies.
  • Set up the file structure for the service queue app.
  • Configure the async database connection and verify the server starts.
Create the project and install dependencies

Every Python project starts with an isolated environment so your dependencies do not collide with other projects on your machine.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir service-queue
cd service-queue
  • Confirm the folder exists by running:
ls

You should see the service-queue directory listed in the output.

  • Create a Python virtual environment by running this command:
python3 -m venv venv
  • Activate the virtual environment:

macOS

source venv/bin/activate

Windows

.\venv\Scripts\Activate

You should see (venv) appear at the beginning of your terminal prompt.

Don't see (venv) in your prompt?

Make sure you ran the activate command from inside the service-queue folder.

On Windows, if you get a permissions error, run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned in PowerShell first, then try activating again.

help me activate my Python virtual environment

  • Install the project dependencies by running:
pip install "fastapi[standard]" "sqlalchemy[asyncio]" asyncpg pydantic-settings PyJWT

What are these packages?

  • fastapi[standard] installs FastAPI plus uvicorn (the dev server) and other standard utilities.
  • sqlalchemy[asyncio] adds async support to SQLAlchemy for non-blocking database queries.
  • asyncpg is a high-performance native PostgreSQL driver built for async Python.
  • pydantic-settings lets you manage configuration (like database URLs) through environment variables or defaults.
  • PyJWT handles JSON Web Token encoding and decoding for potential auth extensions.

Wait for the install to finish. You should see Successfully installed followed by a list of package names in the terminal output.

Install failing with connection or build errors?

Confirm you have (venv) in your prompt. Installing without the venv active can cause permission errors.

If asyncpg fails to build, ensure you have PostgreSQL development headers installed on your system. On macOS with Homebrew, run brew install postgresql. On Ubuntu, run sudo apt install libpq-dev.

help me fix my pip install errors

Create the project file structure

Your service queue app uses a modular layout that separates concerns into distinct files. This makes it easier to navigate as the codebase grows.

  • Create the app directory and its subdirectories by running:

macOS

mkdir -p app/routers app/services

Windows

mkdir app\routers, app\services
  • Create the project files by running:

macOS

touch app/main.py app/models.py app/schemas.py app/database.py app/websocket.py app/routers/.gitkeep app/services/.gitkeep

Windows

New-Item -ItemType File -Path app\main.py, app\models.py, app\schemas.py, app\database.py, app\websocket.py, app\routers\.gitkeep, app\services\.gitkeep
  • Confirm the file structure by running:

macOS

find app -type f

Windows

Get-ChildItem app -Recurse -File

You should see app/main.py, app/models.py, app/schemas.py, app/database.py, app/websocket.py, and the two .gitkeep files listed.

What are .gitkeep files?

Git does not track empty directories. Placing a .gitkeep file inside app/routers/ and app/services/ ensures Git preserves these folders when you commit your project.

Configure the database connection

The database module sets up an async connection pool to PostgreSQL. Async connections let the server handle other requests while waiting for database responses instead of blocking.

  • Create the PostgreSQL database for this project by running:
createdb servicequeue

Getting a connection or role error?

If you see a "role does not exist" error, your PostgreSQL might require a specific username. Try running createdb -U postgres servicequeue instead.

If you see "connection refused", confirm your PostgreSQL server is running. On macOS, try brew services start postgresql. On Linux, try sudo systemctl start postgresql.

help me create my PostgreSQL database

  • Open app/main.py in VS Code and add the following code:
from fastapi import FastAPI

app = FastAPI(title="Real-Time Service Queue API")
  • Save app/main.py.
  • Open app/database.py in VS Code and add the following code:
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
from pydantic_settings import BaseSettings

# Load configuration from environment variables or defaults
class Settings(BaseSettings):
    database_url: str = "postgresql+asyncpg://localhost/servicequeue"

settings = Settings()

# Create the async engine and session factory
engine = create_async_engine(settings.database_url, echo=True)
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

# Base class for all SQLAlchemy models
class Base(DeclarativeBase):
    pass

# Dependency that provides a database session to route handlers
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

What does this code do?

  • Settings uses pydantic-settings to read configuration from environment variables. If none are set, it falls back to the default database URL.
  • create_async_engine creates a connection pool to PostgreSQL using the asyncpg driver. The echo=True flag prints SQL statements to the console for debugging.
  • async_sessionmaker is a factory that produces database sessions. Setting expire_on_commit=False keeps object data accessible after a commit in async contexts.
  • get_db() is a FastAPI dependency that opens a session for each request and automatically closes it when the request finishes.

Using a password for PostgreSQL?

If your local PostgreSQL requires a username and password, update the database_url default to include credentials: postgresql+asyncpg://username:password@localhost/servicequeue.

  • Save app/database.py.

Seeing a red underline on imports?

Make sure VS Code is using the Python interpreter from your virtual environment. Press Cmd+Shift+P (macOS) or Ctrl+Shift+P (Windows) and search for "Python: Select Interpreter". Choose the one inside your venv folder.

help me fix import errors in VS Code

  • Verify the server starts by running this command:
fastapi dev app/main.py

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

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

You should see the Swagger UI page titled "Real-Time Service Queue API" with no endpoints listed yet.

Server not starting?

If you see "command not found", make sure your virtual environment is still active (look for (venv) in your prompt). The fastapi CLI is installed inside the venv.

If you see a module import error, confirm pip install completed successfully earlier. Re-run the install command from the first substep.

help me start my FastAPI dev server

✔️ Awesome, I've got everything!

Great. Double check you have saved both app/main.py and app/database.py.

ⓧ I'd like to double check the full code

Here is the complete code for each file at this point:

from fastapi import FastAPI

app = FastAPI(title="Real-Time Service Queue API")
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str = "postgresql+asyncpg://localhost/servicequeue"

settings = Settings()

engine = create_async_engine(settings.database_url, echo=True)
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

Your project skeleton is ready and the server starts cleanly. Next up, you will define the ticket model and build a state machine that enforces valid transitions.

Build the Ticket Model and State Machine

Your project skeleton is ready and the dev server is running. Now you need to define the core domain: a ticket that moves through a lifecycle of states.

In a real service queue (think bank counters or clinic waiting rooms), a ticket cannot jump from "waiting" directly to "completed". It must be called first. Enforcing those rules in code is called a state machine. In this step, you will build the data model, the transition rules, and the REST endpoints that expose the queue to clients.

In this step, get ready to:
  • Define the Ticket model and Pydantic schemas.
  • Implement a state machine that validates ticket transitions.
  • Create REST endpoints for creating, calling, and completing tickets.
Define the ticket model and schemas

Every ticket in the queue needs a status. You will represent the possible statuses as a Python enum, then define a SQLAlchemy model that maps to a tickets table in PostgreSQL.

  • Add the ticket status enum and model to app/models.py by pasting this code:
import enum
from datetime import datetime
from sqlalchemy import Column, Integer, String, DateTime, Enum as SAEnum
from app.database import Base

# Define the valid states a ticket can be in
class TicketStatus(str, enum.Enum):
    waiting = "waiting"
    being_served = "being_served"
    completed = "completed"
    no_show = "no_show"

# Map this class to a "tickets" table in the database
class Ticket(Base):
    __tablename__ = "tickets"

    id = Column(Integer, primary_key=True, index=True)
    ticket_number = Column(String, unique=True, nullable=False)
    status = Column(SAEnum(TicketStatus), default=TicketStatus.waiting, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
    called_at = Column(DateTime, nullable=True)
    completed_at = Column(DateTime, nullable=True)

What does this code do?

TicketStatus inherits from both str and enum.Enum. This lets SQLAlchemy store the value as a string in the database while Python treats it as a proper enum.

The Ticket model uses Column definitions to declare each field. The called_at and completed_at timestamps start as None and get filled in as the ticket progresses through states.

  • Save app/models.py.
  • You won't see a change yet. The next chunk defines the schemas that shape API responses.
  • Define the request and response schemas in app/schemas.py by pasting this code:
from pydantic import BaseModel
from datetime import datetime
from app.models import TicketStatus
from typing import Optional

# Empty body because tickets are auto-generated (no user input needed)
class TicketCreate(BaseModel):
    pass

# Shape the JSON response for ticket objects
class TicketResponse(BaseModel):
    id: int
    ticket_number: str
    status: TicketStatus
    created_at: datetime
    called_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None

    class Config:
        from_attributes = True

What does this code do?

TicketCreate is empty because customers don't provide data when joining a queue. The system auto-assigns a ticket number.

TicketResponse uses Pydantic to define the exact JSON shape returned by the API. The from_attributes = True config lets Pydantic read values directly from the SQLAlchemy model object.

  • Save app/schemas.py.
  • You won't see a change yet. The service layer you build next uses both files.
Implement the state machine service

The state machine is a function that checks whether moving from one status to another is allowed. If it is not, the function raises an HTTP 400 error immediately.

  • Create a new file app/services/ticket_service.py and paste the following code:
from fastapi import HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.models import Ticket, TicketStatus
from datetime import datetime
import uuid

# Define which transitions are legal for each status
VALID_TRANSITIONS = {
    TicketStatus.waiting: [TicketStatus.being_served, TicketStatus.no_show],
    TicketStatus.being_served: [TicketStatus.completed, TicketStatus.no_show],
    TicketStatus.completed: [],
    TicketStatus.no_show: [],
}

def validate_transition(current: TicketStatus, next_status: TicketStatus):
    """Raise a 400 error if the transition is not allowed."""
    if next_status not in VALID_TRANSITIONS[current]:
        raise HTTPException(
            status_code=400,
            detail=f"Invalid transition from {current} to {next_status}"
        )

async def create_ticket(db: AsyncSession) -> Ticket:
    """Create a new ticket with a unique number and add it to the queue."""
    ticket_number = f"T-{uuid.uuid4().hex[:6].upper()}"
    ticket = Ticket(ticket_number=ticket_number, status=TicketStatus.waiting)
    db.add(ticket)
    await db.commit()
    await db.refresh(ticket)
    return ticket

async def call_next_ticket(db: AsyncSession) -> Ticket:
    """Find the oldest waiting ticket and move it to being_served."""
    result = await db.execute(
        select(Ticket)
        .where(Ticket.status == TicketStatus.waiting)
        .order_by(Ticket.created_at)
        .limit(1)
    )
    ticket = result.scalar_one_or_none()
    if not ticket:
        raise HTTPException(status_code=404, detail="No waiting tickets")
    validate_transition(ticket.status, TicketStatus.being_served)
    ticket.status = TicketStatus.being_served
    ticket.called_at = datetime.utcnow()
    await db.commit()
    await db.refresh(ticket)
    return ticket

async def complete_ticket(ticket_id: int, db: AsyncSession) -> Ticket:
    """Mark a ticket as completed after service is done."""
    result = await db.execute(select(Ticket).where(Ticket.id == ticket_id))
    ticket = result.scalar_one_or_none()
    if not ticket:
        raise HTTPException(status_code=404, detail="Ticket not found")
    validate_transition(ticket.status, TicketStatus.completed)
    ticket.status = TicketStatus.completed
    ticket.completed_at = datetime.utcnow()
    await db.commit()
    await db.refresh(ticket)
    return ticket

What does this code do?

VALID_TRANSITIONS is a dictionary that maps each status to a list of statuses it can move to. completed and no_show are terminal states with empty lists.

validate_transition() checks the dictionary. If the requested move is not in the allowed list, it raises an HTTPException with status code 400.

The three async functions handle ticket creation, calling the next customer, and completing service. Each one commits the change to the database and refreshes the object so the returned data is up to date.

  • Save app/services/ticket_service.py.
  • You won't see a change yet. The router you create next exposes these functions as HTTP endpoints.

Why a state machine instead of plain if-statements?

A dictionary-based transition map centralizes all the business rules in one place. When you need to add a new state later (like "paused" or "transferred"), you update one dictionary instead of hunting through scattered conditionals across your codebase.

This is the pattern interviewers probe for in system design discussions. It shows you think about domain invariants explicitly.

✔️ Awesome, I've got everything!

Great. Make sure you have saved app/services/ticket_service.py.

ⓧ I'd like to double check the full code

from fastapi import HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.models import Ticket, TicketStatus
from datetime import datetime
import uuid

VALID_TRANSITIONS = {
    TicketStatus.waiting: [TicketStatus.being_served, TicketStatus.no_show],
    TicketStatus.being_served: [TicketStatus.completed, TicketStatus.no_show],
    TicketStatus.completed: [],
    TicketStatus.no_show: [],
}

def validate_transition(current: TicketStatus, next_status: TicketStatus):
    if next_status not in VALID_TRANSITIONS[current]:
        raise HTTPException(
            status_code=400,
            detail=f"Invalid transition from {current} to {next_status}"
        )

async def create_ticket(db: AsyncSession) -> Ticket:
    ticket_number = f"T-{uuid.uuid4().hex[:6].upper()}"
    ticket = Ticket(ticket_number=ticket_number, status=TicketStatus.waiting)
    db.add(ticket)
    await db.commit()
    await db.refresh(ticket)
    return ticket

async def call_next_ticket(db: AsyncSession) -> Ticket:
    result = await db.execute(
        select(Ticket)
        .where(Ticket.status == TicketStatus.waiting)
        .order_by(Ticket.created_at)
        .limit(1)
    )
    ticket = result.scalar_one_or_none()
    if not ticket:
        raise HTTPException(status_code=404, detail="No waiting tickets")
    validate_transition(ticket.status, TicketStatus.being_served)
    ticket.status = TicketStatus.being_served
    ticket.called_at = datetime.utcnow()
    await db.commit()
    await db.refresh(ticket)
    return ticket

async def complete_ticket(ticket_id: int, db: AsyncSession) -> Ticket:
    result = await db.execute(select(Ticket).where(Ticket.id == ticket_id))
    ticket = result.scalar_one_or_none()
    if not ticket:
        raise HTTPException(status_code=404, detail="Ticket not found")
    validate_transition(ticket.status, TicketStatus.completed)
    ticket.status = TicketStatus.completed
    ticket.completed_at = datetime.utcnow()
    await db.commit()
    await db.refresh(ticket)
    return ticket
Create the REST endpoints and wire everything up

The router file connects HTTP requests to the service functions. You will also update app/main.py to register the router and create the database tables on startup.

  • Create a new file app/routers/tickets.py and paste this code:
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.schemas import TicketResponse
from app.services import ticket_service

# Group all ticket endpoints under the /tickets prefix
router = APIRouter(prefix="/tickets", tags=["tickets"])

@router.post("/", response_model=TicketResponse)
async def create_ticket(db: AsyncSession = Depends(get_db)):
    return await ticket_service.create_ticket(db)

@router.post("/call-next", response_model=TicketResponse)
async def call_next(db: AsyncSession = Depends(get_db)):
    return await ticket_service.call_next_ticket(db)

@router.post("/{ticket_id}/complete", response_model=TicketResponse)
async def complete_ticket(ticket_id: int, db: AsyncSession = Depends(get_db)):
    return await ticket_service.complete_ticket(ticket_id, db)

What does this code do?

APIRouter groups related endpoints together. The prefix="/tickets" means every route in this router starts with /tickets.

Depends(get_db) injects a database session into each endpoint. FastAPI handles creating and closing the session automatically.

  • Save app/routers/tickets.py.

Next, you need to add two configuration settings to app/database.py that the background worker will use in a later step.

  • In app/database.py, find the Settings class. It currently looks like this:
class Settings(BaseSettings):
    database_url: str = "postgresql+asyncpg://localhost/servicequeue"
  • Add two new fields below database_url so the class looks like this:
class Settings(BaseSettings):
    database_url: str = "postgresql+asyncpg://localhost/servicequeue"
    no_show_threshold_minutes: int = 5
    worker_interval_seconds: int = 30
  • Save app/database.py.

Finally, update app/main.py to register the router and create database tables when the server starts.

  • Replace the entire contents of app/main.py with this code:
from fastapi import FastAPI
from app.routers import tickets
from app.database import engine, Base
from contextlib import asynccontextmanager

# Create tables on startup, clean up on shutdown
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield

app = FastAPI(title="Real-Time Service Queue API", lifespan=lifespan)

# Register the tickets router
app.include_router(tickets.router)

What does this code do?

The lifespan context manager runs code before the first request is served. Base.metadata.create_all tells SQLAlchemy to create any tables that do not already exist in the database.

app.include_router(tickets.router) mounts all the ticket endpoints onto the main application.

  • Save app/main.py.
  • Return to your terminal and start the development server by running:
fastapi dev app/main.py
  • Open http://127.0.0.1:8000/docs in your browser. You should see three new endpoints listed under the tickets tag: POST /tickets/, POST /tickets/call-next, and POST /tickets/{ticket_id}/complete.
  • Click POST /tickets/, then click Try it out, then click Execute. You should see a 200 response with a ticket in waiting status.
  • Click POST /tickets/call-next and execute it. You should see the ticket move to being_served status with a called_at timestamp.
  • Click POST /tickets/{ticket_id}/complete, enter the ticket id from the previous response, and execute. The ticket should move to completed status.

Seeing a connection error?

Make sure your local PostgreSQL server is running and a database named servicequeue exists. You can create it by running createdb servicequeue in a separate terminal.

If you see an import error mentioning app.services, check that your file is saved at the exact path app/services/ticket_service.py (not inside a nested subfolder).

help me debug a connection or import error with my FastAPI ticket service

Now test the state machine by attempting an invalid transition.

  • Create another ticket by executing POST /tickets/ again.
  • Try to complete this new ticket directly by executing POST /tickets/{ticket_id}/complete with the new ticket's id.
  • You should see a 400 response with the message Invalid transition from waiting to completed. The state machine rejected the illegal jump.

✔️ Awesome, I've got everything!

Great. Double check you have saved all files.

ⓧ I'd like to double check the full code

from fastapi import FastAPI
from app.routers import tickets
from app.database import engine, Base
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield

app = FastAPI(title="Real-Time Service Queue API", lifespan=lifespan)

app.include_router(tickets.router)
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str = "postgresql+asyncpg://localhost/servicequeue"
    no_show_threshold_minutes: int = 5
    worker_interval_seconds: int = 30

settings = Settings()

engine = create_async_engine(settings.database_url, echo=True)
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session
import enum
from datetime import datetime
from sqlalchemy import Column, Integer, String, DateTime, Enum as SAEnum
from app.database import Base

class TicketStatus(str, enum.Enum):
    waiting = "waiting"
    being_served = "being_served"
    completed = "completed"
    no_show = "no_show"

class Ticket(Base):
    __tablename__ = "tickets"

    id = Column(Integer, primary_key=True, index=True)
    ticket_number = Column(String, unique=True, nullable=False)
    status = Column(SAEnum(TicketStatus), default=TicketStatus.waiting, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
    called_at = Column(DateTime, nullable=True)
    completed_at = Column(DateTime, nullable=True)
from pydantic import BaseModel
from datetime import datetime
from app.models import TicketStatus
from typing import Optional

class TicketCreate(BaseModel):
    pass

class TicketResponse(BaseModel):
    id: int
    ticket_number: str
    status: TicketStatus
    created_at: datetime
    called_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None

    class Config:
        from_attributes = True
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.schemas import TicketResponse
from app.services import ticket_service

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

@router.post("/", response_model=TicketResponse)
async def create_ticket(db: AsyncSession = Depends(get_db)):
    return await ticket_service.create_ticket(db)

@router.post("/call-next", response_model=TicketResponse)
async def call_next(db: AsyncSession = Depends(get_db)):
    return await ticket_service.call_next_ticket(db)

@router.post("/{ticket_id}/complete", response_model=TicketResponse)
async def complete_ticket(ticket_id: int, db: AsyncSession = Depends(get_db)):
    return await ticket_service.complete_ticket(ticket_id, db)
from fastapi import HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.models import Ticket, TicketStatus
from datetime import datetime
import uuid

VALID_TRANSITIONS = {
    TicketStatus.waiting: [TicketStatus.being_served, TicketStatus.no_show],
    TicketStatus.being_served: [TicketStatus.completed, TicketStatus.no_show],
    TicketStatus.completed: [],
    TicketStatus.no_show: [],
}

def validate_transition(current: TicketStatus, next_status: TicketStatus):
    if next_status not in VALID_TRANSITIONS[current]:
        raise HTTPException(
            status_code=400,
            detail=f"Invalid transition from {current} to {next_status}"
        )

async def create_ticket(db: AsyncSession) -> Ticket:
    ticket_number = f"T-{uuid.uuid4().hex[:6].upper()}"
    ticket = Ticket(ticket_number=ticket_number, status=TicketStatus.waiting)
    db.add(ticket)
    await db.commit()
    await db.refresh(ticket)
    return ticket

async def call_next_ticket(db: AsyncSession) -> Ticket:
    result = await db.execute(
        select(Ticket)
        .where(Ticket.status == TicketStatus.waiting)
        .order_by(Ticket.created_at)
        .limit(1)
    )
    ticket = result.scalar_one_or_none()
    if not ticket:
        raise HTTPException(status_code=404, detail="No waiting tickets")
    validate_transition(ticket.status, TicketStatus.being_served)
    ticket.status = TicketStatus.being_served
    ticket.called_at = datetime.utcnow()
    await db.commit()
    await db.refresh(ticket)
    return ticket

async def complete_ticket(ticket_id: int, db: AsyncSession) -> Ticket:
    result = await db.execute(select(Ticket).where(Ticket.id == ticket_id))
    ticket = result.scalar_one_or_none()
    if not ticket:
        raise HTTPException(status_code=404, detail="Ticket not found")
    validate_transition(ticket.status, TicketStatus.completed)
    ticket.status = TicketStatus.completed
    ticket.completed_at = datetime.utcnow()
    await db.commit()
    await db.refresh(ticket)
    return ticket

Your ticket model, state machine, and REST endpoints are all working. Next up, you will add WebSocket broadcasting so connected clients see queue changes in real time.

Add WebSocket Broadcasting

Your REST endpoints handle ticket state changes, but clients have no way to know when the queue updates unless they manually refresh.

In this step, you will add a WebSocket endpoint that pushes live queue updates to every connected client the instant a ticket is created, called, or completed.

In this step, get ready to:
  • Build a ConnectionManager class that tracks and broadcasts to active WebSocket connections.
  • Create a WebSocket endpoint that keeps clients connected and listening.
  • Wire up broadcasting so every ticket state change pushes a live queue snapshot.
Implement the ConnectionManager class

The ConnectionManager is a singleton that holds a list of all active WebSocket connections and exposes methods for adding, removing, and broadcasting to them.

  • Open app/websocket.py in VS Code and add the following code:
from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from typing import List
import json

router = APIRouter()

class ConnectionManager:
    def __init__(self):
        # Track all currently connected WebSocket clients
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        # Accept the incoming connection and add to the list
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        # Remove a disconnected client from the list
        self.active_connections.remove(websocket)

    async def broadcast(self, message: dict):
        # Send a JSON message to every connected client
        for connection in self.active_connections:
            await connection.send_text(json.dumps(message))

# Single shared instance used across the application
manager = ConnectionManager()

What does this code do?

The ConnectionManager class follows FastAPI's official WebSocket pattern. It maintains a list of active connections and provides three methods.

  • connect() accepts a new WebSocket handshake and adds the client to the list.
  • disconnect() removes a client when they close the tab or lose connection.
  • broadcast() loops through every active connection and sends the same JSON message to all of them simultaneously.

The manager instance at the bottom is the single shared object that every part of the app uses to send messages.

  • Save app/websocket.py.

You will not see a change yet. The next chunk adds the endpoint that actually accepts client connections.

Create the WebSocket endpoint and register the router

The ConnectionManager handles the plumbing, but you still need an endpoint where clients connect. This endpoint keeps the connection open and listens for incoming messages (or disconnection).

  • Add the following endpoint below the manager = ConnectionManager() line in app/websocket.py:
@router.websocket("/ws/queue")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        # Keep the connection alive by waiting for messages
        while True:
            await websocket.receive_text()
    except WebSocketDisconnect:
        manager.disconnect(websocket)

What does this endpoint do?

When a client connects to /ws/queue, the endpoint accepts the connection and enters an infinite loop calling receive_text(). This loop keeps the connection open.

When the client disconnects (closes the browser tab or drops the network), a WebSocketDisconnect exception fires and the manager removes that client from the active list.

  • Save app/websocket.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved app/websocket.py before moving on.

ⓧ I'd like to double check the full code

from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from typing import List
import json

router = APIRouter()

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: dict):
        for connection in self.active_connections:
            await connection.send_text(json.dumps(message))

manager = ConnectionManager()

@router.websocket("/ws/queue")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            await websocket.receive_text()
    except WebSocketDisconnect:
        manager.disconnect(websocket)

Now register this WebSocket router in the main application file so FastAPI knows about the endpoint.

  • In app/main.py, find this import block at the top:
from fastapi import FastAPI
from app.routers import tickets
from app.database import engine, Base
from contextlib import asynccontextmanager
  • Add the WebSocket router import so the imports look like this:
from fastapi import FastAPI
from app.routers import tickets
from app.database import engine, Base
from app.websocket import router as ws_router
from contextlib import asynccontextmanager
  • Then find app.include_router(tickets.router) near the bottom of the file and add the WebSocket router below it:
app.include_router(tickets.router)
app.include_router(ws_router)
  • Save app/main.py.
  • In your terminal, restart the dev server by running:
fastapi dev app/main.py

The server should start without errors. You will not see the WebSocket endpoint in the Swagger docs page at http://127.0.0.1:8000/docs because WebSocket endpoints are not rendered in OpenAPI by default.

Seeing an ImportError?

  • Check that app/websocket.py defines router = APIRouter() at module level. The import from app.websocket import router as ws_router expects that name to exist.
  • Make sure there is no typo in the import path. The file is app/websocket.py, not app/websockets.py (no trailing s).

Why is my WebSocket router import failing?

Integrate broadcasting into the ticket service

The WebSocket endpoint can accept connections, but nothing broadcasts to those connections yet. You need to send a queue snapshot every time a ticket is created, called, or completed.

First, define the Pydantic schemas that describe the shape of the queue snapshot message.

  • Open app/schemas.py and find the Optional import at the top:
from typing import Optional
  • Update it to also import List:
from typing import Optional, List
  • Then add the following two classes at the bottom of app/schemas.py, below the TicketResponse class:
class QueueTicketPosition(BaseModel):
    position: int
    ticket_number: str
    ticket_id: int

class QueueSnapshot(BaseModel):
    waiting: List[QueueTicketPosition]
    being_served_count: int

What do these schemas represent?

QueueSnapshot is the message shape that gets broadcast to all WebSocket clients. It contains a list of waiting tickets (each with their position in line) and a count of how many tickets are currently being served.

Defining these as Pydantic models means the shape is documented and validated, even though the broadcast sends raw JSON.

  • Save app/schemas.py.

Next, update the ticket service to build and broadcast a queue snapshot after every state change.

  • Replace the entire contents of app/services/ticket_service.py with the following code:
from fastapi import HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.models import Ticket, TicketStatus
from datetime import datetime
import uuid

VALID_TRANSITIONS = {
    TicketStatus.waiting: [TicketStatus.being_served, TicketStatus.no_show],
    TicketStatus.being_served: [TicketStatus.completed, TicketStatus.no_show],
    TicketStatus.completed: [],
    TicketStatus.no_show: [],
}

def validate_transition(current: TicketStatus, next_status: TicketStatus):
    if next_status not in VALID_TRANSITIONS[current]:
        raise HTTPException(
            status_code=400,
            detail=f"Invalid transition from {current} to {next_status}"
        )

async def get_queue_snapshot(db: AsyncSession) -> dict:
    # Fetch all waiting tickets ordered by arrival time
    result = await db.execute(
        select(Ticket)
        .where(Ticket.status == TicketStatus.waiting)
        .order_by(Ticket.created_at)
    )
    waiting_tickets = result.scalars().all()
    # Count how many tickets are currently being served
    being_served_result = await db.execute(
        select(Ticket).where(Ticket.status == TicketStatus.being_served)
    )
    being_served_count = len(being_served_result.scalars().all())
    return {
        "waiting": [
            {"position": i + 1, "ticket_number": t.ticket_number, "ticket_id": t.id}
            for i, t in enumerate(waiting_tickets)
        ],
        "being_served_count": being_served_count,
    }

async def create_ticket(db: AsyncSession, broadcast_fn=None) -> Ticket:
    ticket_number = f"T-{uuid.uuid4().hex[:6].upper()}"
    ticket = Ticket(ticket_number=ticket_number, status=TicketStatus.waiting)
    db.add(ticket)
    await db.commit()
    await db.refresh(ticket)
    # Broadcast the updated queue to all connected clients
    if broadcast_fn:
        snapshot = await get_queue_snapshot(db)
        await broadcast_fn(snapshot)
    return ticket

async def call_next_ticket(db: AsyncSession, broadcast_fn=None) -> Ticket:
    result = await db.execute(
        select(Ticket)
        .where(Ticket.status == TicketStatus.waiting)
        .order_by(Ticket.created_at)
        .limit(1)
    )
    ticket = result.scalar_one_or_none()
    if not ticket:
        raise HTTPException(status_code=404, detail="No waiting tickets")
    validate_transition(ticket.status, TicketStatus.being_served)
    ticket.status = TicketStatus.being_served
    ticket.called_at = datetime.utcnow()
    await db.commit()
    await db.refresh(ticket)
    # Broadcast the updated queue to all connected clients
    if broadcast_fn:
        snapshot = await get_queue_snapshot(db)
        await broadcast_fn(snapshot)
    return ticket

async def complete_ticket(ticket_id: int, db: AsyncSession, broadcast_fn=None) -> Ticket:
    result = await db.execute(select(Ticket).where(Ticket.id == ticket_id))
    ticket = result.scalar_one_or_none()
    if not ticket:
        raise HTTPException(status_code=404, detail="Ticket not found")
    validate_transition(ticket.status, TicketStatus.completed)
    ticket.status = TicketStatus.completed
    ticket.completed_at = datetime.utcnow()
    await db.commit()
    await db.refresh(ticket)
    # Broadcast the updated queue to all connected clients
    if broadcast_fn:
        snapshot = await get_queue_snapshot(db)
        await broadcast_fn(snapshot)
    return ticket

What changed in the service layer?

  • get_queue_snapshot() queries all waiting tickets (ordered by arrival) and counts tickets being served. It returns a dictionary matching the QueueSnapshot schema shape.
  • Each service function now accepts an optional broadcast_fn parameter. After a successful commit, it calls broadcast_fn(snapshot) to push the updated queue state to all WebSocket clients.
  • Making broadcast_fn optional (defaulting to None) means the service still works in tests or contexts where no WebSocket manager exists.
  • Save app/services/ticket_service.py.

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before continuing.

ⓧ I'd like to double check the full code

from fastapi import HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.models import Ticket, TicketStatus
from datetime import datetime
import uuid

VALID_TRANSITIONS = {
    TicketStatus.waiting: [TicketStatus.being_served, TicketStatus.no_show],
    TicketStatus.being_served: [TicketStatus.completed, TicketStatus.no_show],
    TicketStatus.completed: [],
    TicketStatus.no_show: [],
}

def validate_transition(current: TicketStatus, next_status: TicketStatus):
    if next_status not in VALID_TRANSITIONS[current]:
        raise HTTPException(
            status_code=400,
            detail=f"Invalid transition from {current} to {next_status}"
        )

async def get_queue_snapshot(db: AsyncSession) -> dict:
    result = await db.execute(
        select(Ticket)
        .where(Ticket.status == TicketStatus.waiting)
        .order_by(Ticket.created_at)
    )
    waiting_tickets = result.scalars().all()
    being_served_result = await db.execute(
        select(Ticket).where(Ticket.status == TicketStatus.being_served)
    )
    being_served_count = len(being_served_result.scalars().all())
    return {
        "waiting": [
            {"position": i + 1, "ticket_number": t.ticket_number, "ticket_id": t.id}
            for i, t in enumerate(waiting_tickets)
        ],
        "being_served_count": being_served_count,
    }

async def create_ticket(db: AsyncSession, broadcast_fn=None) -> Ticket:
    ticket_number = f"T-{uuid.uuid4().hex[:6].upper()}"
    ticket = Ticket(ticket_number=ticket_number, status=TicketStatus.waiting)
    db.add(ticket)
    await db.commit()
    await db.refresh(ticket)
    if broadcast_fn:
        snapshot = await get_queue_snapshot(db)
        await broadcast_fn(snapshot)
    return ticket

async def call_next_ticket(db: AsyncSession, broadcast_fn=None) -> Ticket:
    result = await db.execute(
        select(Ticket)
        .where(Ticket.status == TicketStatus.waiting)
        .order_by(Ticket.created_at)
        .limit(1)
    )
    ticket = result.scalar_one_or_none()
    if not ticket:
        raise HTTPException(status_code=404, detail="No waiting tickets")
    validate_transition(ticket.status, TicketStatus.being_served)
    ticket.status = TicketStatus.being_served
    ticket.called_at = datetime.utcnow()
    await db.commit()
    await db.refresh(ticket)
    if broadcast_fn:
        snapshot = await get_queue_snapshot(db)
        await broadcast_fn(snapshot)
    return ticket

async def complete_ticket(ticket_id: int, db: AsyncSession, broadcast_fn=None) -> Ticket:
    result = await db.execute(select(Ticket).where(Ticket.id == ticket_id))
    ticket = result.scalar_one_or_none()
    if not ticket:
        raise HTTPException(status_code=404, detail="Ticket not found")
    validate_transition(ticket.status, TicketStatus.completed)
    ticket.status = TicketStatus.completed
    ticket.completed_at = datetime.utcnow()
    await db.commit()
    await db.refresh(ticket)
    if broadcast_fn:
        snapshot = await get_queue_snapshot(db)
        await broadcast_fn(snapshot)
    return ticket

Finally, update the router to pass the ConnectionManager's broadcast method into each service function.

  • Replace the entire contents of app/routers/tickets.py with the following code:
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.schemas import TicketResponse
from app.services import ticket_service
from app.websocket import manager

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

@router.post("/", response_model=TicketResponse)
async def create_ticket(db: AsyncSession = Depends(get_db)):
    return await ticket_service.create_ticket(db, broadcast_fn=manager.broadcast)

@router.post("/call-next", response_model=TicketResponse)
async def call_next(db: AsyncSession = Depends(get_db)):
    return await ticket_service.call_next_ticket(db, broadcast_fn=manager.broadcast)

@router.post("/{ticket_id}/complete", response_model=TicketResponse)
async def complete_ticket(ticket_id: int, db: AsyncSession = Depends(get_db)):
    return await ticket_service.complete_ticket(ticket_id, db, broadcast_fn=manager.broadcast)

What changed in the router?

The router now imports manager from app.websocket and passes manager.broadcast as the broadcast_fn argument to every service call.

This keeps the service layer decoupled from WebSocket details. The service only knows it has a callable to send messages. The router decides which callable to inject.

  • Save app/routers/tickets.py.
  • Restart the dev server in your terminal by pressing Ctrl+C to stop it, then running:
fastapi dev app/main.py

The server should start without errors. Now test the full broadcasting flow.

  • Open two browser tabs and navigate to http://127.0.0.1:8000/docs in the first tab.
  • In the second tab, open the browser developer console by pressing Cmd+Option+J (macOS) or Ctrl+Shift+J (Windows).
  • Paste the following JavaScript into the console and press Enter:
const ws = new WebSocket('ws://127.0.0.1:8000/ws/queue');
ws.onmessage = (event) => console.log('Queue update:', JSON.parse(event.data));
  • Switch to the first tab (/docs) and use the POST /tickets/ endpoint to create a ticket.
  • Switch back to the second tab. You should see a Queue update: log in the console showing the new ticket's position in the waiting list.

Not seeing console output in the second tab?

  • Make sure the WebSocket connection opened successfully. Look for any error messages in the console immediately after pasting the connection code.
  • Confirm your dev server is still running. If it crashed after the file changes, check the terminal for import or syntax errors.
  • Verify you used ws:// (not http://) in the WebSocket URL.

Why is my WebSocket not receiving broadcast messages from the ticket service?

Every connected client now receives a live queue snapshot the instant any ticket changes state. Next up, you will add a background worker that automatically detects abandoned tickets and marks them as no-shows.

Build the Background No-Show Worker

Your WebSocket broadcasting is working. Clients connected to /ws/queue receive live updates whenever a ticket moves through the state machine. But what happens when someone takes a number and never shows up?

Real queue systems need a way to automatically clear abandoned tickets. In this step, you'll build a background worker that runs on a loop inside the FastAPI process, detects stale tickets, marks them as no-shows using the state machine, and broadcasts the change to all connected clients.

In this step, get ready to:
  • Create an async worker function that detects and marks stale tickets as no-shows.
  • Register the worker in the FastAPI lifespan with proper shutdown handling.
  • Test automatic no-show detection with live WebSocket broadcasting.
Create the no-show worker function

The worker is a long-running async function that sleeps for a configurable interval, wakes up, queries the database for tickets that have been waiting too long, and marks each one as a no-show. It manages its own database session independently from the request lifecycle.

  • Create a new file app/services/worker.py and add the following code:
import asyncio
from datetime import datetime, timedelta
from sqlalchemy import select
from app.database import AsyncSessionLocal, settings
from app.models import Ticket, TicketStatus
from app.services.ticket_service import validate_transition, get_queue_snapshot

async def no_show_worker(manager):
    # Run indefinitely until cancelled
    while True:
        # Sleep first, then check for stale tickets
        await asyncio.sleep(settings.worker_interval_seconds)
        # Open a fresh database session for each cycle
        async with AsyncSessionLocal() as db:
            # Calculate the cutoff time for stale tickets
            threshold = datetime.utcnow() - timedelta(minutes=settings.no_show_threshold_minutes)
            # Find all waiting tickets older than the threshold
            result = await db.execute(
                select(Ticket)
                .where(
                    Ticket.status == TicketStatus.waiting,
                    Ticket.created_at < threshold
                )
            )
            stale_tickets = result.scalars().all()
            # Mark each stale ticket as no_show using the state machine
            for ticket in stale_tickets:
                try:
                    validate_transition(ticket.status, TicketStatus.no_show)
                    ticket.status = TicketStatus.no_show
                except Exception:
                    pass
            # Commit and broadcast only if tickets were marked
            if stale_tickets:
                await db.commit()
                snapshot = await get_queue_snapshot(db)
                await manager.broadcast(snapshot)

What does this code do?

The no_show_worker function accepts a manager parameter (your ConnectionManager instance) so it can broadcast updates to WebSocket clients.

  • It sleeps first with asyncio.sleep() so the server finishes starting before the first scan runs.
  • Each cycle opens its own database session via AsyncSessionLocal() instead of relying on the request-scoped get_db() dependency. Background tasks must manage their own resources.
  • The try/except around validate_transition handles edge cases where a ticket's status changed between the query and the loop iteration.
  • After committing, it broadcasts the updated queue snapshot so every connected WebSocket client sees the change instantly.

Why not use Celery or APScheduler?

The no-show detection is lightweight. It runs one database query and a few updates per cycle. Because it lives in the same process as the FastAPI app, it can directly access the ConnectionManager and broadcast to WebSocket clients without needing a message broker.

Adding Celery would require Redis or RabbitMQ and a separate worker process. That is overkill for a single periodic scan. The asyncio.create_task pattern inside the lifespan is the idiomatic approach for lightweight in-process background work in FastAPI.

  • Save app/services/worker.py.

Seeing an import error?

  • Make sure you have both app/services/__init__.py (or .gitkeep) and app/__init__.py so Python recognizes the directories as packages.
  • Check that your app/database.py exports both AsyncSessionLocal and settings. These were created in Step 1.

Still stuck? Why is my worker.py showing an import error?

Register the worker in the FastAPI lifespan

The worker function exists, but nothing starts it yet. You need to launch it as an asyncio task when the app boots up, and cancel it cleanly when the app shuts down.

  • In app/main.py, find the existing import line:
from app.websocket import router as ws_router
  • Replace the entire contents of app/main.py with the updated code below:
from fastapi import FastAPI
from app.routers import tickets
from app.database import engine, Base
from app.websocket import router as ws_router, manager
from app.services.worker import no_show_worker
from contextlib import asynccontextmanager
import asyncio

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Create database tables on startup
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    # Start the background no-show worker
    task = asyncio.create_task(no_show_worker(manager))
    yield
    # Cancel the worker on shutdown
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass

app = FastAPI(title="Real-Time Service Queue API", lifespan=lifespan)

app.include_router(tickets.router)
app.include_router(ws_router)

What changed in the lifespan?

  • The import of manager from app.websocket gives the lifespan access to the ConnectionManager instance. The worker needs this to broadcast updates.
  • asyncio.create_task(no_show_worker(manager)) schedules the worker as a concurrent task that runs alongside the FastAPI event loop. It does not block the server from handling requests.
  • After yield, the shutdown code calls task.cancel() and catches asyncio.CancelledError. This ensures the worker stops cleanly when you press Ctrl+C or the server shuts down.
  • Save app/main.py.

Server won't start after editing main.py?

  • Check that the import line reads from app.websocket import router as ws_router, manager with both router and manager in the same import.
  • Verify you have import asyncio at the top of the file. Without it, asyncio.create_task is undefined.

Still stuck? Help me fix my main.py lifespan after adding the worker

Test automatic no-show detection

The default settings wait 5 minutes before marking a ticket as a no-show, with the worker running every 30 seconds. For testing, you can override these with environment variables so you see results in seconds instead of minutes.

  • Stop the dev server if it is running (press Ctrl+C).
  • Set the test environment variables and restart the server by running:
NO_SHOW_THRESHOLD_MINUTES=0 WORKER_INTERVAL_SECONDS=10 fastapi dev app/main.py

What do these variables do?

NO_SHOW_THRESHOLD_MINUTES=0 tells the worker that any waiting ticket older than 0 minutes (any age at all) should be marked as a no-show. WORKER_INTERVAL_SECONDS=10 makes the worker run every 10 seconds so you see results quickly.

Because your Settings class in app/database.py uses pydantic-settings, it automatically reads matching environment variable names and overrides the defaults.

  • Open your browser and navigate to http://127.0.0.1:8000/docs.
  • Expand POST /tickets/ and click Try it out, then Execute to create a ticket.
  • Wait approximately 10 seconds and watch your terminal.

You should see SQL queries appear in the terminal as the worker wakes up, finds the stale ticket, and updates its status to no_show.

  • To verify the WebSocket broadcast, open a second browser tab and navigate to http://127.0.0.1:8000/docs.
  • Use the WebSocket endpoint section (or a WebSocket client connected to ws://127.0.0.1:8000/ws/queue) to observe that the queue snapshot broadcasts automatically after the worker marks tickets as no-shows.

Worker not marking tickets?

  • Make sure you included NO_SHOW_THRESHOLD_MINUTES=0 before the fastapi dev command on the same line. If set on a separate line, use export first.
  • Check your terminal for error tracebacks. A common issue is a missing __init__.py in the app/services/ directory.
  • Confirm that the ticket you created still has waiting status. If you already called POST /tickets/call-next, it moved to being_served and the threshold logic does not apply.

Still stuck? My background worker is not detecting stale tickets

✔️ Awesome, I've got everything!

Great. Double check you've saved both files before moving on.

ⓧ I'd like to double check the full code

Here is the complete code for the two files you created or modified in this step.

import asyncio
from datetime import datetime, timedelta
from sqlalchemy import select
from app.database import AsyncSessionLocal, settings
from app.models import Ticket, TicketStatus
from app.services.ticket_service import validate_transition, get_queue_snapshot

async def no_show_worker(manager):
    while True:
        await asyncio.sleep(settings.worker_interval_seconds)
        async with AsyncSessionLocal() as db:
            threshold = datetime.utcnow() - timedelta(minutes=settings.no_show_threshold_minutes)
            result = await db.execute(
                select(Ticket)
                .where(
                    Ticket.status == TicketStatus.waiting,
                    Ticket.created_at < threshold
                )
            )
            stale_tickets = result.scalars().all()
            for ticket in stale_tickets:
                try:
                    validate_transition(ticket.status, TicketStatus.no_show)
                    ticket.status = TicketStatus.no_show
                except Exception:
                    pass
            if stale_tickets:
                await db.commit()
                snapshot = await get_queue_snapshot(db)
                await manager.broadcast(snapshot)
from fastapi import FastAPI
from app.routers import tickets
from app.database import engine, Base
from app.websocket import router as ws_router, manager
from app.services.worker import no_show_worker
from contextlib import asynccontextmanager
import asyncio

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    task = asyncio.create_task(no_show_worker(manager))
    yield
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass

app = FastAPI(title="Real-Time Service Queue API", lifespan=lifespan)

app.include_router(tickets.router)
app.include_router(ws_router)

Your queue system now manages itself. Abandoned tickets get cleared automatically, and every connected client sees the update in real time. This is the complete production pattern: state machine enforcement, real-time broadcasting, and periodic background maintenance all running in a single async process.

Secret mission

Add a Dashboard Caching Layer

Your queue API works in real time, but what about a dashboard that summarizes today's stats? Computing aggregate queries on every request is wasteful. In this secret mission, you'll add a cached statistics endpoint with TTL-based invalidation and cache-awareness headers.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally (Python files and a local PostgreSQL database), so there are no ongoing cloud costs.

Resources you used:

  • Project folder on your Desktop containing the app/ package and Python virtual environment.
  • PostgreSQL servicequeue database running locally.
  • FastAPI development server (running process in your terminal).

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with the API.

  • Your FastAPI dev server keeps running in the terminal until you close it.
  • Your servicequeue database stays in PostgreSQL and takes up minimal disk space.
  • Your project files remain in the folder on your Desktop.

Pause - I'll come back to this later

Shut down the running process to free up your terminal. Your files and database stay intact for next time.

  • In the terminal where FastAPI is running, press Ctrl+C to stop the development server.
  • Your project folder and virtual environment remain on your Desktop.
  • Your PostgreSQL servicequeue database stays available for when you return.

Delete - I don't want to use this again

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

  • In the terminal where FastAPI is running, press Ctrl+C to stop the development server.
  • Drop the PostgreSQL database by running:
dropdb servicequeue
  • Remove the project folder by running (macOS/Linux):
cd ~/Desktop
rm -rf service-queue
  • On Windows, run:
cd ~\Desktop
rmdir /s /q service-queue

This removes your virtual environment, all project source code, and the local database.

Nice Work!

Nice Work!

You just built a production-patterned real-time service queue API from scratch. This is exactly the kind of system that powers bank queues, hospital waiting rooms, and support desks.

You've learned how to:

  • Build a REST API with ticket lifecycle management and a state machine that enforces valid transitions and rejects invalid ones.
  • Implement real-time WebSocket broadcasting so all connected clients see queue updates instantly.
  • Create a periodic background worker using FastAPI lifespan and asyncio that detects abandoned tickets automatically.
  • Secret Mission: Added an in-memory caching layer with TTL and cache invalidation for a dashboard statistics endpoint.

Ready to quiz yourself?