How a click reaches a database and comes back

"""Create the vending machine database and fill it with a few snacks.

Run this file once before you start the server.
Running it again resets the machine to its starting state.
"""

import sqlite3
from contextlib import closing
from pathlib import Path

# The database is one ordinary file that sits beside this script.
# Path(__file__).parent is the folder that holds this script, so the file
# ends up in the same place no matter where you run the script from.
DATABASE_FILE = Path(__file__).parent / "vending.db"

# The schema describes the shape of our data: one table with four columns.
# The comments after the double dashes are SQL comments.
SCHEMA = """
DROP TABLE IF EXISTS inventory;

CREATE TABLE inventory (
    slot_id     TEXT    PRIMARY KEY,                           -- identifies one slot, for example A1
    item_name   TEXT    NOT NULL,                              -- every slot must have a name
    price_cents INTEGER NOT NULL CHECK (price_cents >= 0),     -- whole cents, never decimals
    stock       INTEGER NOT NULL CHECK (stock >= 0)            -- the database refuses negative stock
);
"""

# Each tuple is one row: (slot_id, item_name, price_cents, stock).
# The Oat Biscuits start with no stock so that we can test a sold-out slot.
STARTING_STOCK = [
    ("A1", "Sparkling Water", 175, 8),
    ("A2", "Salted Peanuts", 125, 5),
    ("B1", "Chocolate Bar", 200, 3),
    ("B2", "Oat Biscuits", 150, 0),
]

# closing() makes sure that the connection is closed when the block ends.
with closing(sqlite3.connect(DATABASE_FILE)) as connection:
    # "with connection" groups the work into one transaction. If every
    # statement succeeds, the changes are committed together. If one fails,
    # the changes since the last commit are rolled back.
    with connection:
        # executescript runs several SQL statements in one call. It commits
        # anything that came before it, so creating the table is its own step.
        connection.executescript(SCHEMA)

        # The question marks are placeholders. The values travel to the
        # database separately from the SQL text. The four inserts that follow
        # belong to one transaction, so either all of them are stored or none.
        connection.executemany(
            "INSERT INTO inventory (slot_id, item_name, price_cents, stock) VALUES (?, ?, ?, ?)",
            STARTING_STOCK,
        )

    # Read the rows back so that we can see what was stored.
    for row in connection.execute(
        "SELECT slot_id, item_name, price_cents, stock FROM inventory ORDER BY slot_id"
    ):
        print(row)

"""A small web server that sells snacks from a SQLite database.

The file has four layers, from the bottom of the program to the top:
  1. The data layer talks to the database. It knows nothing about HTTP.
  2. The route layer connects URLs to data functions and checks the input.
  3. The HTTP layer reads requests and writes responses.
  4. The last lines start the server.
"""

import json
import re
import sqlite3
from contextlib import closing
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlsplit

# --- Settings ----------------------------------------------------------------

# 127.0.0.1 means "this computer", so only programs on your own machine can connect.
HOST = "127.0.0.1"

# The port decides which program on this computer receives the request.
PORT = 8000

DATABASE_FILE = Path(__file__).parent / "vending.db"
PUBLIC_FOLDER = Path(__file__).parent / "public"

# The only files that the server will send. Each URL path maps to a file name
# and a Content-Type. A fixed list stops visitors from asking for other files,
# such as server.py or vending.db, by typing a path like /../server.py.
STATIC_FILES = {
    "/": ("index.html", "text/html; charset=utf-8"),
    "/style.css": ("style.css", "text/css; charset=utf-8"),
    "/app.js": ("app.js", "text/javascript; charset=utf-8"),
}

# A valid slot id is one capital letter followed by one digit, such as A1 or B2.
SLOT_ID_PATTERN = re.compile(r"[A-Z][0-9]")

# --- Errors ------------------------------------------------------------------
# The data layer reports problems by raising these exceptions. Each name
# describes a problem in the language of the vending machine, and the HTTP
# layer decides which status code each problem deserves.


class BadRequest(Exception):
    """The client sent something we cannot make sense of."""


class NoSuchSlot(Exception):
    """The slot id is well formed, but no such slot exists."""


class SoldOut(Exception):
    """The slot exists, but it has no stock left."""

# --- Data layer --------------------------------------------------------------
# These functions are the only place where the program talks to the database.


def open_connection():
    """Open a new connection to the database file."""
    connection = sqlite3.connect(DATABASE_FILE)
    # With sqlite3.Row we can read a column by name, as in row["stock"],
    # and not only by position.
    connection.row_factory = sqlite3.Row
    return connection

def find_slot(connection, slot_id):
    """Return one slot as a dictionary, or None if there is no such slot."""
    # The ? is a placeholder. The value travels separately from the SQL text,
    # so a visitor can never change what the SQL means (SQL injection).
    # A tuple with one item needs the trailing comma.
    row = connection.execute(
        "SELECT slot_id, item_name, price_cents, stock FROM inventory WHERE slot_id = ?",
        (slot_id,),
    ).fetchone()
    return dict(row) if row else None


def get_all_slots():
    """Return every slot as a list of dictionaries."""
    with closing(open_connection()) as connection:
        # Without ORDER BY, SQL promises no particular order for the rows.
        rows = connection.execute(
            "SELECT slot_id, item_name, price_cents, stock FROM inventory ORDER BY slot_id"
        ).fetchall()
    return [dict(row) for row in rows]


def get_slot(slot_id):
    """Return one slot, or raise NoSuchSlot."""
    with closing(open_connection()) as connection:
        slot = find_slot(connection, slot_id)
    if slot is None:
        raise NoSuchSlot(f"No slot called {slot_id}.")
    return slot

def buy_slot(slot_id):
    """Take one item from a slot and return the updated slot."""
    with closing(open_connection()) as connection:
        # "with connection" is one transaction. The changes are committed if
        # the block ends normally and rolled back if an exception escapes.
        with connection:
            # One statement checks the stock and changes it. If two buyers
            # arrive at the same moment, the database handles them one after
            # the other, so they cannot both take the last item.
            update_result = connection.execute(
                "UPDATE inventory SET stock = stock - 1 WHERE slot_id = ? AND stock > 0",
                (slot_id,),
            )
            # Read the slot inside the same transaction, so that the numbers
            # we report agree with the change we just made.
            slot = find_slot(connection, slot_id)

    # rowcount is the number of rows that the UPDATE changed.
    if slot is None:
        raise NoSuchSlot(f"No slot called {slot_id}.")
    if update_result.rowcount == 0:
        raise SoldOut(f"{slot['item_name']} is sold out.")
    return slot

# --- Route layer -------------------------------------------------------------
# A handler connects one URL to one data function. It checks the input, calls
# the data layer and returns data. It contains no SQL and no HTTP details.


def validate_slot_id(raw_slot_id):
    """Tidy up a slot id from the URL and check that it has the right shape."""
    # upper() lets a visitor type a1 and mean A1.
    slot_id = raw_slot_id.upper()
    if not SLOT_ID_PATTERN.fullmatch(slot_id):
        raise BadRequest("Slot ids look like A1 or B2.")
    return slot_id

def handle_list_slots():
    return get_all_slots()


def handle_show_slot(raw_slot_id):
    return get_slot(validate_slot_id(raw_slot_id))


def handle_buy_slot(raw_slot_id):
    return buy_slot(validate_slot_id(raw_slot_id))


# The route table is the API written as data. Each row holds an HTTP method, a
# pattern for the URL path, and the handler for that route. In a pattern,
# ([^/]+) captures the text between two slashes, and the captured text is
# passed to the handler as an argument.
ROUTES = [
    ("GET", re.compile(r"/api/slots"), handle_list_slots),
    ("GET", re.compile(r"/api/slots/([^/]+)"), handle_show_slot),
    ("POST", re.compile(r"/api/slots/([^/]+)/buy"), handle_buy_slot),
]

# --- HTTP layer --------------------------------------------------------------


class VendingRequestHandler(BaseHTTPRequestHandler):
    """Python creates one of these objects for every request that arrives."""

    # BaseHTTPRequestHandler calls do_GET for GET requests and do_POST for
    # POST requests. We send both to one method of our own.
    def do_GET(self):
        self.answer_request("GET")

    def do_POST(self):
        self.answer_request("POST")

    def answer_request(self, http_method):
        # self.path includes any query string, so we keep only the path.
        url_path = urlsplit(self.path).path

        try:
            # First, is the visitor asking for one of our page files?
            if http_method == "GET" and url_path in STATIC_FILES:
                self.send_static_file(url_path)
                return

            # Otherwise look for a matching route. The first match wins.
            for route_method, path_pattern, handler in ROUTES:
                path_match = path_pattern.fullmatch(url_path)
                if route_method == http_method and path_match:
                    # groups() holds the captured text, such as "A1".
                    self.send_json(200, handler(*path_match.groups()))
                    return

            # Nothing matched.
            self.send_json(404, {"error": "No such route."})

        # These lines turn the exceptions of the data layer into status codes.
        except BadRequest as error:
            self.send_json(400, {"error": str(error)})
        except NoSuchSlot as error:
            self.send_json(404, {"error": str(error)})
        except SoldOut as error:
            self.send_json(409, {"error": str(error)})
        except Exception as error:
            # Anything else is a bug in our program. We print the details for
            # the developer and tell the visitor very little.
            print("Unexpected error:", repr(error))
            self.send_json(500, {"error": "Something went wrong on the server."})

    def send_static_file(self, url_path):
        file_name, content_type = STATIC_FILES[url_path]
        body = (PUBLIC_FOLDER / file_name).read_bytes()
        self.send_body(200, content_type, body)

    def send_json(self, status_code, data):
        # json.dumps turns the data into text, and encode turns the text into
        # bytes, because a network carries bytes and not characters.
        body = json.dumps(data).encode("utf-8")
        self.send_body(status_code, "application/json; charset=utf-8", body)

    def send_body(self, status_code, content_type, body):
        # HTTP requires this order: status line, headers, a blank line, body.
        self.send_response(status_code)
        self.send_header("Content-Type", content_type)  # how to read the body
        self.send_header("Content-Length", str(len(body)))  # how many bytes follow
        self.send_header("Cache-Control", "no-store")  # always ask again while we develop
        self.end_headers()  # writes the blank line
        self.wfile.write(body)

# This block runs only when you start this file directly. It does not run when
# another file imports server.py to reuse the data functions.
if __name__ == "__main__":
    web_server = ThreadingHTTPServer((HOST, PORT), VendingRequestHandler)
    print(f"Vending machine running at http://{HOST}:{PORT}  (press Ctrl+C to stop)")
    try:
        # Wait for requests and answer each one in its own thread.
        web_server.serve_forever()
    except KeyboardInterrupt:
        # Ctrl+C in the terminal ends up here.
        print("\nStopping the server.")
        web_server.server_close()

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <!-- Makes the page fit a phone screen instead of shrinking a desktop layout. -->
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Vending Machine</title>
  <link rel="stylesheet" href="/style.css">
</head>

<body>
  <main class="machine">
    <h1>Vending Machine</h1>

    <!-- role="status" makes screen readers announce changes to this text. -->
    <p id="status" class="screen" role="status">Loading...</p>

    <!-- Empty at first. The script adds one card here for each slot. -->
    <section id="slots" class="slots"></section>
  </main>

  <!-- A template is HTML that the browser keeps but does not show. The script
       copies it once for each slot and fills in the copy. -->
  <template id="slot-template">
    <article class="slot">
      <h2 class="item-name"></h2>
      <p class="price"></p>
      <p class="stock"></p>
      <button type="button" class="buy">Buy</button>
    </article>
  </template>

  <!-- The script comes last, so every element it looks for already exists. -->
  <script type="module" src="/app.js"></script>
</body>
</html>

/* Named values that the rest of the file reuses. Change a colour here once. */
:root {
  --page: #eceff1;
  --body: #37474f;
  --screen-bg: #102a12;
  --screen-text: #39ff14;
  --card: #455a64;
  --accent: #ffb300;
  --text: #ffffff;
}

/* Centre the machine horizontally. */
body {
  font-family: system-ui, sans-serif;
  background: var(--page);
  display: flex;
  justify-content: center;
  padding: 24px 16px;
}

/* min() means "480 pixels wide, but never wider than the screen". */
.machine {
  background: var(--body);
  color: var(--text);
  padding: 24px;
  border-radius: 12px;
  width: min(100%, 480px);
}

h1 {
  margin: 0 0 16px;
  font-size: 1.4rem;
}

/* The status line looks like the green display of an old machine. */
.screen {
  background: var(--screen-bg);
  color: var(--screen-text);
  font-family: ui-monospace, Menlo, Consolas, monospace;
  padding: 12px 16px;
  border-radius: 6px;
  margin: 0 0 16px;
}

/* A grid with two equal columns. */
.slots {
  display: grid;
  grid-template-columns: 1fr 1fr;
  gap: 12px;
}

.slot {
  background: var(--card);
  padding: 12px;
  border-radius: 8px;
}

.slot h2 {
  font-size: 1rem;
  margin: 0 0 8px;
}

.slot p {
  margin: 0 0 4px;
}

/* The script only adds or removes the class "sold-out". This stylesheet
   decides what a sold-out card looks like. */
.slot.sold-out {
  opacity: 0.55;
}

button {
  margin-top: 8px;
  width: 100%;
  padding: 10px;
  font-weight: bold;
  background: var(--accent);
  border: none;
  border-radius: 6px;
  cursor: pointer;
}

/* The script sets the disabled property of a button when stock is zero. */
button:disabled {
  background: #90a4ae;
  cursor: not-allowed;
}

// This script runs in the browser. It asks the server for data and changes
// the page to match what the server says.

// Find the elements that we need, once, when the script loads.
const slotListElement = document.getElementById("slots");
const statusElement = document.getElementById("status");
const slotTemplate = document.getElementById("slot-template");

// A formatter from the browser that writes prices in euros, so that 1.75
// becomes €1.75.
const euroFormatter = new Intl.NumberFormat("en-IE", { style: "currency", currency: "EUR" });

// Send a request to our API and return the JSON from the reply. The function
// is async, so it can wait for the network without freezing the page.
async function requestJson(url, options) {
  let response;
  try {
    // fetch sends the request and returns a promise of a response.
    // "await" pauses this function until the response arrives.
    response = await fetch(url, options);
  } catch {
    // fetch fails like this only when no reply arrives at all, for example
    // when the server is not running.
    throw new Error("Cannot reach the server. Is it running?");
  }

  // The body of the reply is JSON text. json() parses it into an object.
  const data = await response.json();

  // A reply can arrive with a bad status such as 404 or 409.
  // response.ok is false for those, so we pass the server's message on.
  if (!response.ok) {
    throw new Error(data.error);
  }
  return data;
}

// Build one card for one slot. The card is not on the page until we add it.
function createSlotCard(slot) {
  // Copy the card layout from the template in index.html.
  const slotCard = slotTemplate.content.firstElementChild.cloneNode(true);
  const isSoldOut = slot.stock === 0;

  // data-slot remembers which slot this card belongs to, for the click handler.
  slotCard.dataset.slot = slot.slot_id;

  // The sold-out class lets the stylesheet decide how a sold-out card looks.
  slotCard.classList.toggle("sold-out", isSoldOut);

  // textContent treats a value as plain text, never as HTML, so data from the
  // server can never run as code in the page.
  slotCard.querySelector(".item-name").textContent = `${slot.slot_id}: ${slot.item_name}`;

  // Prices are stored in cents. We divide by 100 only when we show them.
  slotCard.querySelector(".price").textContent = euroFormatter.format(slot.price_cents / 100);

  slotCard.querySelector(".stock").textContent = isSoldOut ? "Sold out" : `${slot.stock} left`;
  slotCard.querySelector("button").disabled = isSoldOut;
  return slotCard;
}

// Ask the server for every slot and show them all on the page.
async function loadAllSlots() {
  try {
    const slots = await requestJson("/api/slots");

    // map turns a list of slot data into a list of cards.
    // replaceChildren puts the cards on the page. This changes the DOM, and
    // the browser then redraws the screen.
    slotListElement.replaceChildren(...slots.map(createSlotCard));
    statusElement.textContent = "Choose a snack.";
  } catch (error) {
    statusElement.textContent = error.message;
  }
}

// One listener on the whole list handles clicks on every Buy button. This
// technique is called event delegation, and it also covers buttons that the
// script adds later.
slotListElement.addEventListener("click", async (event) => {
  // closest() climbs from the clicked element up to the nearest Buy button.
  // It returns null when the click landed somewhere else.
  const buyButton = event.target.closest("button.buy");
  if (!buyButton) {
    return;
  }

  const slotCard = buyButton.closest("article");
  try {
    // POST asks the server to change something. The reply is the updated slot.
    const updatedSlot = await requestJson(`/api/slots/${slotCard.dataset.slot}/buy`, { method: "POST" });

    // Replace only this card, using what the server says and not what we guess.
    slotCard.replaceWith(createSlotCard(updatedSlot));
    statusElement.textContent = `Enjoy your ${updatedSlot.item_name}!`;
  } catch (error) {
    statusElement.textContent = error.message;
  }
});

// Show the slots as soon as the page opens.
loadAllSlots();

POST /api/slots/a1/buy HTTP/1.1

Host: 127.0.0.1:8000
User-Agent: curl/8.5.0
Accept: */*

HTTP/1.0 200 OK

Server: BaseHTTP/0.6 Python/3.13.16
Date: Sat, 17 Oct 2026 21:33:37 GMT
Content-Type: application/json; charset=utf-8
Content-Length: 81
Cache-Control: no-store

{"slot_id": "A1", "item_name": "Sparkling Water", "price_cents": 175, "stock": 7}

"""The same API, written with the FastAPI framework.

Install:  pip install fastapi uvicorn
Run:      uvicorn server_fastapi:app --port 8000
Then open http://127.0.0.1:8000 and also http://127.0.0.1:8000/docs
"""

from pathlib import Path
from typing import Annotated

from fastapi import FastAPI, Path as UrlPath, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles

# We reuse the data layer from server.py. Importing that file does not start
# the other server, because of the "if __name__ == '__main__'" guard there.
from server import NoSuchSlot, SoldOut, buy_slot, get_all_slots, get_slot

app = FastAPI(title="Vending machine")

# A slot id must match this pattern. The framework checks it for us, which
# replaces the validate_slot_id function in server.py.
SlotId = Annotated[str, UrlPath(pattern=r"^[A-Za-z][0-9]$")]

# Each decorator adds one row to the framework's route table. These are the same
# three routes as the ROUTES list in server.py.
@app.get("/api/slots")
def list_slots():
    return get_all_slots()


@app.get("/api/slots/{slot_id}")
def show_slot(slot_id: SlotId):
    return get_slot(slot_id.upper())


@app.post("/api/slots/{slot_id}/buy")
def buy(slot_id: SlotId):
    return buy_slot(slot_id.upper())

# These handlers turn exceptions into status codes. We keep the same
# {"error": ...} shape as before, so the page does not need to change.
@app.exception_handler(RequestValidationError)
def handle_bad_request(request: Request, error: RequestValidationError):
    return JSONResponse({"error": "Slot ids look like A1 or B2."}, status_code=400)


@app.exception_handler(NoSuchSlot)
def handle_no_such_slot(request: Request, error: NoSuchSlot):
    return JSONResponse({"error": str(error)}, status_code=404)


@app.exception_handler(SoldOut)
def handle_sold_out(request: Request, error: SoldOut):
    return JSONResponse({"error": str(error)}, status_code=409)

# Serve the page files. This line must come last, because the routes above
# are checked first.
app.mount("/", StaticFiles(directory=Path(__file__).parent / "public", html=True))