diff --git a/.github/workflows/workshop-booking.yml b/.github/workflows/workshop-booking.yml new file mode 100644 index 00000000..a4a001d9 --- /dev/null +++ b/.github/workflows/workshop-booking.yml @@ -0,0 +1,38 @@ +name: Workshop booking checks + +on: + pull_request: + paths: + - 'applications/workshop-booking/**' + - '.github/workflows/workshop-booking.yml' + push: + branches: [main] + paths: + - 'applications/workshop-booking/**' + - '.github/workflows/workshop-booking.yml' + workflow_dispatch: + +permissions: + contents: read + +jobs: + check: + runs-on: ubuntu-latest + defaults: + run: + working-directory: applications/workshop-booking + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.12' + cache: pip + cache-dependency-path: applications/workshop-booking/requirements*.txt + - run: python -m pip install -r requirements-dev.txt + - run: python -m pip check + - run: ruff check app migrations tests + - run: ruff format --check app migrations tests + - run: pytest tests/unit + +# The explicitly configured managed-service tests are documented in tests/README.md. +# Pull requests do not receive Cloud or database credentials. diff --git a/README.md b/README.md index 2d9d660f..eed51bdf 100644 --- a/README.md +++ b/README.md @@ -25,7 +25,9 @@ Examples and recipes for building with ClickHouse, from your first SQL query to [ClickHouse](https://clickhouse.com/clickhouse) is an open-source, column-oriented SQL database for fast analytics on large datasets. It powers real-time dashboards, reporting, observability, and applications that need to query data as it arrives. -You can use the engine as a managed Cloud service, run your own server, embed it in Python with chDB, or query files from the command line with `clickhouse-local`. The wider ClickHouse ecosystem brings together Postgres for transactions, ClickStack for observability, and LibreChat for working with data through AI agents. +[ClickHouse Cloud](https://clickhouse.com/cloud) is the overall managed data platform. Within it, ClickHouse is the analytical database service, and [ClickHouse Managed Postgres](https://clickhouse.com/cloud/postgres) is the PostgreSQL service for transactional applications. + +You can use ClickHouse as a managed service within that platform, run your own server, embed it in Python with chDB, or query files from the command line with `clickhouse-local`. The wider ClickHouse ecosystem brings together Postgres for transactions, ClickStack for observability, and LibreChat for working with data through AI agents. ## How to get started @@ -42,18 +44,18 @@ Open the example's README for its prerequisites, setup, and commands. Each examp ## Explore products and examples -### ClickHouse and ClickHouse Cloud: real-time analytics +### ClickHouse: the analytical database service -Use [ClickHouse](https://clickhouse.com/clickhouse) to ingest, transform, and query analytical data with SQL. [ClickHouse Cloud](https://clickhouse.com/cloud) runs it as a fully managed service on AWS, Google Cloud, and Azure, handling infrastructure, scaling, and upgrades so you can focus on your application. +Use [ClickHouse](https://clickhouse.com/clickhouse) to ingest, transform, and query analytical data with SQL. The ClickHouse analytical database service runs within the ClickHouse Cloud platform on AWS, Google Cloud, and Azure, with managed infrastructure, scaling, and upgrades so you can focus on your application. - [Real-time stock data with Massive](./blog-examples/stock-data-demo/README.md): Stream trades and quotes into ClickHouse and explore them in a React dashboard. -- [Report results and run history](./applications/report-history/README.md): Build a TypeScript application that provisions ClickHouse Cloud, stores report results, and queries across completed runs. +- [Report results and run history](./applications/report-history/README.md): Build a TypeScript application that provisions a ClickHouse service within ClickHouse Cloud, stores report results, and queries across completed runs. - [Docker Compose recipes](./docker-compose-recipes/README.md): Run ClickHouse locally alongside Grafana, Dagster, Redpanda, SeaweedFS, and RustFS, or explore replicated clusters. - [Learn ClickHouse with Mark](./LearnClickHouseWithMark/README.md): Work through SQL techniques, JSON, aggregations, geospatial queries, and more, with companion videos. ### Language clients: the same tour in eight official clients -ClickHouse ships official client libraries for [C# / .NET](https://clickhouse.com/docs/integrations/csharp), [Java](https://clickhouse.com/docs/integrations/language-clients/java) (Client V2 and JDBC), [Rust](https://clickhouse.com/docs/integrations/rust), [Go](https://clickhouse.com/docs/integrations/go), [C++](https://clickhouse.com/docs/integrations/language-clients/cpp), [Python](https://clickhouse.com/docs/integrations/python), and [Node.js](https://clickhouse.com/docs/integrations/javascript). The language clients example writes one small program in every one of them against a ClickHouse Cloud service provisioned with `clickhousectl`: connect over TLS, create a table, batch insert typed rows, bind query parameters, stream results, map aggregates into typed records, and handle a server error. +ClickHouse ships official client libraries for [C# / .NET](https://clickhouse.com/docs/integrations/csharp), [Java](https://clickhouse.com/docs/integrations/language-clients/java) (Client V2 and JDBC), [Rust](https://clickhouse.com/docs/integrations/rust), [Go](https://clickhouse.com/docs/integrations/go), [C++](https://clickhouse.com/docs/integrations/language-clients/cpp), [Python](https://clickhouse.com/docs/integrations/python), and [Node.js](https://clickhouse.com/docs/integrations/javascript). The language clients example writes one small program in every one of them against a ClickHouse analytical database service provisioned within ClickHouse Cloud using `clickhousectl`: connect over TLS, create a table, batch insert typed rows, bind query parameters, stream results, map aggregates into typed records, and handle a server error. - [Language client tour](./language-clients/README.md): Pick your language, run it, and compare it side by side with the others. Every implementation prints the same output. @@ -61,16 +63,17 @@ ClickHouse ships official client libraries for [C# / .NET](https://clickhouse.co [`clickhousectl`](https://clickhouse.com/docs/products/cloud/features/cli) is the CLI for managing local ClickHouse installations and ClickHouse Cloud. Install and switch local versions, start development servers, provision and scale Cloud services, run queries, and manage Postgres and ClickPipes. JSON output and installable agent skills make it useful for scripts and AI coding agents. See the [CLI repository](https://github.com/ClickHouse/clickhousectl) for source code and installation options. -- [Provision a database for a TypeScript application](./applications/report-history/README.md): Use `clickhousectl` to create the Cloud service for a report-history application. +- [Provision a database for a TypeScript application](./applications/report-history/README.md): Use `clickhousectl` to create a ClickHouse analytical database service for a report-history application. - [Investigate and resolve a latency SLA breach](./ai/clickhousectl/agentic-sla-scaling/README.md): Give an agent access to `clickhousectl` to inspect a live service and apply a scaling change. -### ClickHouse Managed Postgres: transactions alongside analytics +### ClickHouse Managed Postgres: the PostgreSQL service -[ClickHouse Managed Postgres](https://clickhouse.com/cloud/postgres) is a managed PostgreSQL service in ClickHouse Cloud for transactional applications, with native integration into ClickHouse for analytics. Use Postgres for application records and transactions, then replicate changes to ClickHouse for reporting and aggregation. +[ClickHouse Managed Postgres](https://clickhouse.com/cloud/postgres) is the PostgreSQL service within the ClickHouse Cloud platform. It handles transactional applications and integrates with the ClickHouse analytical database service. Use Postgres for application records and transactions, then replicate changes to ClickHouse for reporting and aggregation when needed. -[ClickPipes](https://clickhouse.com/cloud/clickpipes) provides managed ingestion into ClickHouse Cloud, including Postgres change data capture (CDC), streaming sources, and object storage. +[ClickPipes](https://clickhouse.com/cloud/clickpipes) provides managed ingestion into the ClickHouse analytical database service within ClickHouse Cloud, including Postgres change data capture (CDC), streaming sources, and object storage. - [Shortwave link shortener](./applications/shortwave/README.md): Build and deploy a link shortener using ClickHouse Managed Postgres for application data, ClickPipes for metadata sync, and ClickHouse for click analytics. +- [Workshop Booking API](./applications/workshop-booking/README.md): Build a FastAPI and SQLAlchemy application on ClickHouse Managed Postgres with transactional seat allocation, cancellation, and safe retries. - [Postgres-to-ClickHouse data modeling](./postgresql-clickhouse-data-modeling/README.md): Replicate a tiny fixture from PostgreSQL to ClickHouse with PeerDB, then verify inserts, updates, and deletes. Follow the separate ClickHouse Managed Postgres and ClickPipes walkthrough for ClickHouse Cloud; a larger Stack Overflow import is optional. ### ClickStack: logs, metrics, traces, and session replay @@ -109,13 +112,13 @@ Browse [all local analytics examples](./local-analytics/README.md) for more file | Directory | What you'll find | | --- | --- | -| [applications](./applications/) | Application examples: a link shortener, report results, and run history with ClickHouse Cloud. | +| [applications](./applications/) | Application examples: workshop bookings, a link shortener, report results, and run history with ClickHouse Cloud. | | [ai](./ai/README.md) | AI agents, MCP integrations, and workflows using `clickhousectl`. | | [blog-examples](./blog-examples/) | Code and resources accompanying the [ClickHouse Blog](https://clickhouse.com/blog). | | [clickstack](./clickstack/) | Observability examples for LLM applications and MCP servers. | | [docker-compose-recipes](./docker-compose-recipes/README.md) | Local deployments, integrations, and cluster configurations. | | [ethereum](./ethereum/README.md) | Blockchain schemas, batch and streaming ingestion, and queries. | -| [language-clients](./language-clients/README.md) | The same client tour in C#, Java, Rust, Go, C++, Python, and Node.js against ClickHouse Cloud. | +| [language-clients](./language-clients/README.md) | The same client tour in C#, Java, Rust, Go, C++, Python, and Node.js against a ClickHouse service within ClickHouse Cloud. | | [LearnClickHouseWithMark](./LearnClickHouseWithMark/README.md) | Code accompanying Mark Needham's ClickHouse video tutorials. | | [local-analytics](./local-analytics/README.md) | File queries and conversions with `clickhouse-local` and chDB. | | [postgresql-clickhouse-data-modeling](./postgresql-clickhouse-data-modeling/README.md) | PostgreSQL replication and data modeling with PeerDB and ClickHouse. | diff --git a/applications/workshop-booking/.env.example b/applications/workshop-booking/.env.example new file mode 100644 index 00000000..43db4963 --- /dev/null +++ b/applications/workshop-booking/.env.example @@ -0,0 +1,12 @@ +# Copy to .env and fill in the connection details from ClickHouse Cloud. +PGHOST=your-managed-postgres-host +PGPORT=5432 +PGDATABASE=postgres +PGUSER=workshop_booking_app +PGPASSWORD=replace-with-the-runtime-role-password +# Download this CA from the service's connection details. TLS is always verify-full. +PGSSLROOTCERT=./certs/postgres-ca.pem + +# Map seeded attendee UUIDs to different, random server-side credentials. +# Generate each token with: python -c 'import secrets; print(secrets.token_urlsafe(32))' +API_BEARER_TOKENS='{"11111111-1111-4111-8111-111111111111":"GENERATE_ALICE_TOKEN","22222222-2222-4222-8222-222222222222":"GENERATE_BOB_TOKEN"}' diff --git a/applications/workshop-booking/.gitignore b/applications/workshop-booking/.gitignore new file mode 100644 index 00000000..829cc602 --- /dev/null +++ b/applications/workshop-booking/.gitignore @@ -0,0 +1,16 @@ +.env +.env.* +!.env.example +.venv/ +__pycache__/ +*.py[cod] +.pytest_cache/ +.ruff_cache/ +.coverage +htmlcov/ +certs/ +.deployment/ +*.pem +*.egg-info/ +dist/ +build/ diff --git a/applications/workshop-booking/README.md b/applications/workshop-booking/README.md new file mode 100644 index 00000000..f23c516b --- /dev/null +++ b/applications/workshop-booking/README.md @@ -0,0 +1,404 @@ +# Workshop Booking API + +Book a workshop seat, check availability, and cancel a booking with **FastAPI**, +**SQLAlchemy**, **psycopg**, and **Alembic** on **ClickHouse Managed Postgres**. +The example demonstrates a small but important promise: two concurrent requests +cannot take the same last seat. FastAPI's generated OpenAPI interface lets you +explore the application without building a separate frontend. + +**[Sign up for ClickHouse Cloud with $300 in free trial credits](https://clickhouse.com/cloud).** + +[ClickHouse Cloud](https://clickhouse.com/cloud) is the overall managed data +platform. Within it, [ClickHouse](https://clickhouse.com/clickhouse) is the +analytical database service, and +[ClickHouse Managed Postgres](https://clickhouse.com/cloud/postgres) is the +PostgreSQL service for transactional applications. + +ClickHouse Managed Postgres runs on local NVMe storage, with managed backups and +options for high availability. It integrates with the ClickHouse analytical +database through ClickPipes CDC and the `pg_clickhouse` extension. Together, the +two database services provide a unified transactional and analytical data stack +within ClickHouse Cloud. + +This application uses only the ClickHouse Managed Postgres service: bookings +need transactions, constraints, and current availability. [Shortwave](../shortwave) +provides the complete example and explicit setup commands for using both database +services with ClickPipes. + +## How it works + +```mermaid +flowchart LR + Client[curl or OpenAPI interface] --> API[FastAPI] + API --> ORM[SQLAlchemy + psycopg] + ORM -->|verified TLS| PG[ClickHouse Managed Postgres] + Alembic[Alembic + reviewed SQL files] --> PG +``` + +| Route | Purpose | Success | +| --- | --- | --- | +| `GET /workshops` | List upcoming workshops and available seats | `200` | +| `POST /bookings` | Book one seat using `{ "id": "UUID", "workshop_id": "UUID" }` | `201`, or `200` on retry | +| `GET /bookings/{id}` | Read your booking | `200` | +| `DELETE /bookings/{id}` | Cancel your booking | `200` | + +The booking routes require a bearer token. The server maps tokens to seeded +attendees; callers cannot choose a different owner in the request. Another +attendee's booking returns `404` on read or cancellation. The workshop list is +public and contains no attendee information. + +Three tables live in the `workshop_booking` schema: `workshops`, `attendees`, and +`bookings`. To book a seat, the API locks the workshop row, checks the current +active booking count, and inserts the booking in the same transaction. +Cancellation acquires the same workshop lock. A partial unique index permits at +most one active booking per attendee and workshop. All writes to bookings must +follow this locking protocol; the capacity rule is a transaction rule, not a +cross-table database constraint. + +## Set up the example + +Use a ClickHouse Cloud account with Managed Postgres access, Git, `curl`, `jq`, +OpenSSL, `psql` 15 or newer, and Python 3.12. Run application dependencies in a +Linux development environment. On macOS, maintainers can create an +[isolated OrbStack machine](https://docs.orbstack.dev/machines/isolated): + +```sh +orb create --isolated --isolate-network ubuntu:24.04 workshop-booking-dev +orb -m workshop-booking-dev +``` + +This disables host file sharing and integration, SSH-agent forwarding, and +access from the machine to the host and other machines. No host mounts are +needed. Keep Cloud API credentials on the machine you use for provisioning; +transfer only the database credentials needed for each step into Linux. +The commands below use Bash and run from the application directory unless noted. + +### 1. Install the application in Linux + +```sh +sudo apt-get update +sudo apt-get install -y python3 python3-venv ca-certificates git curl jq openssl postgresql-client +git clone https://github.com/ClickHouse/examples.git +cd examples/applications/workshop-booking +python3 --version +python3 -m venv .venv +.venv/bin/python -m pip install -r requirements.txt +``` + +The requirements file pins the dependency set. There is no Node.js installation +or JavaScript build step. + +### 2. Sign in to ClickHouse Cloud + +Run these provisioning commands where you manage Cloud resources. Install +[clickhousectl](https://clickhouse.com/docs/interfaces/cli), then sign in with +an Admin API key created using the [Cloud API instructions](https://clickhouse.com/docs/cloud/manage/openapi). +The interactive prompt keeps the secret out of shell history. OAuth is +read-only for Postgres provisioning. + +```sh +curl -fsSL https://clickhouse.com/cli | sh +export PATH="$HOME/.local/bin:$PATH" +clickhousectl --version +clickhousectl cloud auth login --interactive +clickhousectl cloud auth status +clickhousectl cloud org list +umask 077 +mkdir -p .deployment +``` + +In an editor, create `.deployment/resources.env` with your organization ID and +a region and size supported by your organization: + +```dotenv +CH_ORG_ID=YOUR_ORGANIZATION_UUID +DEPLOYMENT_NAME=workshop-booking-example +CLOUD_REGION=YOUR_AVAILABLE_AWS_REGION +PG_SIZE=YOUR_AVAILABLE_POSTGRES_SIZE +``` + +Review [Managed Postgres pricing](https://clickhouse.com/pricing) before creation. +This example provisions one Postgres 18 service without HA. Compute, storage, +backups, and network usage can incur charges. Stopping Python or the development +VM does not stop database charges; delete an unneeded dedicated service when +finished. + +### 3. Create Postgres and download its CA + +Run creation once. Preserve the response privately: it contains the initial +administrator password. + +```sh +source .deployment/resources.env +clickhousectl cloud postgres create \ + --org-id "$CH_ORG_ID" --name "$DEPLOYMENT_NAME" \ + --provider aws --region "$CLOUD_REGION" --size "$PG_SIZE" \ + --pg-version 18 --ha-type none \ + --tag project=workshop-booking --tag environment=development --json \ + > .deployment/postgres-create.json +PG_SERVICE_ID="$(jq -er '.id' .deployment/postgres-create.json)" +printf 'PG_SERVICE_ID=%s\n' "$PG_SERVICE_ID" >> .deployment/resources.env +``` + +Repeat **get** until `state` is `running`; do not repeat **create**: + +```sh +clickhousectl cloud postgres get "$PG_SERVICE_ID" \ + --org-id "$CH_ORG_ID" --json > .deployment/postgres-status.json +jq '{id, state, hostname, username, postgresVersion, size}' \ + .deployment/postgres-status.json +``` + +If creation was interrupted, first reconcile against +`clickhousectl cloud postgres list --org-id "$CH_ORG_ID" --json`. A missing local +response does not mean the service was not created. Retrieve the CA only after +the service is running: + +```sh +clickhousectl cloud postgres certs get "$PG_SERVICE_ID" \ + --org-id "$CH_ORG_ID" --output .deployment/postgres-ca.pem +``` + +Use `--output` to obtain a PEM certificate file. JSON-mode output redirected into +a `.pem` file is not a usable CA. Record the returned direct hostname and +administrator username in `.deployment/resources.env`: + +```dotenv +PGHOST=YOUR_POSTGRES_HOSTNAME +PGPORT=5432 +PGDATABASE=postgres +PGADMIN=YOUR_SERVICE_ADMIN_USERNAME +``` + +Use the service's direct Postgres endpoint for both migrations and runtime. +If using an isolated VM, copy the private provisioning receipt and CA only for +database setup, then remove administrator credentials from the runtime copy. +Do not copy the Cloud API credential store into the VM. + +### 4. Create roles, migrate, and seed + +The following commands run in Linux, where `psql`, Python, and the application +files are installed. Generate passwords **once** and preserve the file on retries: + +```sh +umask 077 +mkdir -p .deployment +cat > .deployment/passwords.env < .deployment/workshops.json +jq . .deployment/workshops.json +``` + +Choose a workshop ID from that response and generate one booking ID. Preserve +the booking ID when retrying a request after a timeout or lost response: + +```sh +WORKSHOP_ID=UUID_FROM_WORKSHOPS_RESPONSE +BOOKING_ID="$(.venv/bin/python -c 'import uuid; print(uuid.uuid4())')" +jq -n --arg id "$BOOKING_ID" --arg workshop "$WORKSHOP_ID" \ + '{id: $id, workshop_id: $workshop}' > .deployment/booking-request.json +curl --fail-with-body -sS -i "$BASE_URL/bookings" \ + -H "Authorization: Bearer $API_TOKEN" \ + -H 'Content-Type: application/json' \ + --data-binary @.deployment/booking-request.json +``` + +A new booking returns `201` with `status: "confirmed"`. Repeat the exact request: +it returns `200` with the same booking, without consuming another seat. +Reusing the ID for another existing workshop returns `409`. A new ID for an attendee +who already has an active booking in that workshop also returns `409`. + +```sh +curl --fail-with-body -sS "$BASE_URL/bookings/$BOOKING_ID" \ + -H "Authorization: Bearer $API_TOKEN" | jq +curl --fail-with-body -sS -X DELETE "$BASE_URL/bookings/$BOOKING_ID" \ + -H "Authorization: Bearer $API_TOKEN" | jq +``` + +Cancellation returns `200`, `status: "cancelled"`, and a non-null `cancelled_at`. +Availability increases by one. Repeating DELETE returns the cancelled booking +and leaves availability unchanged. Retrying the original POST after cancellation +returns the current cancelled booking; it never silently makes a new booking. +Use a new booking UUID to book again. + +The seed includes a workshop with one seat. Booking it as one attendee and +trying as another returns `409` for the second request. Sending the two requests +concurrently must produce exactly one confirmed booking. Missing/invalid tokens +return `401`; invalid UUIDs and unexpected body fields return `422`. +Booking request bodies are limited to 4 KiB, including streamed requests; larger +bodies return `413`. The workshop list accepts `limit` from 1 to 100 (default +20) and `offset` from 0 to 10,000, ordered by start time and ID. + +## Test changes + +Install test dependencies inside Linux and run the fast checks: + +```sh +.venv/bin/python -m pip install -r requirements-dev.txt +.venv/bin/pytest tests/unit +``` + +The integration tests use a real Postgres service to exercise competing requests, +retry semantics, cancellation, authorization, rollback, constraints, and TLS. +Run them against a dedicated example service with the fixture credentials +described in [tests/README.md](tests/README.md). They create and remove their own +fixtures. Keep privileged test credentials out of the runtime file. + +## Limits and operation + +- This is a small authenticated API with seeded attendees and workshops. It has + no signup flow, payment handling, calendar integration, waitlist, or admin UI. +- A booking is one seat. An attendee can have only one active booking in a + workshop. New bookings and first-time cancellations close when the workshop + starts. Retrying an existing booking or an already completed cancellation + still returns its current state. +- The engine explicitly uses `READ COMMITTED` isolation and a consistent workshop-first + lock order protect the capacity check. Do not remove the lock when adding + another writer or an administrative booking endpoint. List availability is a + snapshot; the booking transaction makes the final admission decision. +- The pool permits five persistent and five overflow connections per process. + Multiply that limit by the number + of worker processes or replicas before scaling. Requests waiting too long on + the database return a retryable error; retry creation with the same booking ID. +- Bookings, including cancelled records, are retained. Deleting them would remove + the durable retry history, so define a retry window before adding retention. +- Authorization is enforced in the API. The shared runtime database role is not + a separate database tenant boundary. Use only the migration login for schema + changes and never run the application with the administrator password. +- For hosted operation, run Uvicorn under a process supervisor behind HTTPS, + with request limits and appropriate network access. The development listener + is bound to localhost. This example does not provision an application host. + +## Troubleshooting and cleanup + +For TLS errors, verify the downloaded file is PEM, the CA is current, and +`PGHOST` matches this service. For authentication errors, distinguish the Cloud +API key, database administrator, migrator, application password, and HTTP token. +A failed startup can also mean a configured attendee was not seeded. + +If the administrator password is lost, reset it on the recorded service instead +of creating another service: + +```sh +clickhousectl cloud postgres reset-password "$PG_SERVICE_ID" \ + --org-id "$CH_ORG_ID" --generate --json \ + > .deployment/postgres-password-reset.json +``` + +Keep that response private. Resetting the administrator password does not rotate +the separate application login. + +Stop Uvicorn with Ctrl-C. To remove the example from a service you will keep, +review [sql/cleanup.sql](sql/cleanup.sql), reconnect with administrator credentials, +and run the file explicitly: + +```sh +export PGPASSWORD="$(jq -er '.password' .deployment/postgres-create.json)" +psql -X -v ON_ERROR_STOP=1 -U "$PGADMIN" -f sql/cleanup.sql +unset PGPASSWORD +``` + +Use the reset password if it has changed. Cleanup deletes this example's data +and roles. It does not delete the Managed Postgres service or stop its charges. For a +service created exclusively for this example, check the saved ID, then delete it: + +```sh +source .deployment/resources.env +clickhousectl cloud postgres get "$PG_SERVICE_ID" --org-id "$CH_ORG_ID" --json +clickhousectl cloud postgres delete "$PG_SERVICE_ID" --org-id "$CH_ORG_ID" +clickhousectl cloud postgres list --org-id "$CH_ORG_ID" --json +``` + +Confirm the recorded ID is absent from the service list. Preserve receipts until +the outcome is clear, then remove unneeded private credential files and the +dedicated VM. Never delete a shared service to clean up this example. diff --git a/applications/workshop-booking/alembic.ini b/applications/workshop-booking/alembic.ini new file mode 100644 index 00000000..62005ed2 --- /dev/null +++ b/applications/workshop-booking/alembic.ini @@ -0,0 +1,7 @@ +[alembic] +script_location = %(here)s/migrations +prepend_sys_path = . +path_separator = os + +# Connection credentials come exclusively from PG* environment variables. +# This file deliberately contains no sqlalchemy.url or passwords. diff --git a/applications/workshop-booking/app/__init__.py b/applications/workshop-booking/app/__init__.py new file mode 100644 index 00000000..e4116518 --- /dev/null +++ b/applications/workshop-booking/app/__init__.py @@ -0,0 +1 @@ +"""A small, transaction-safe workshop booking API.""" diff --git a/applications/workshop-booking/app/auth.py b/applications/workshop-booking/app/auth.py new file mode 100644 index 00000000..5436af4e --- /dev/null +++ b/applications/workshop-booking/app/auth.py @@ -0,0 +1,30 @@ +import secrets +from typing import Annotated +from uuid import UUID + +from fastapi import Depends, HTTPException, Request +from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer + +bearer = HTTPBearer( + auto_error=False, + description="An attendee token configured on the server in API_BEARER_TOKENS.", +) + + +def get_attendee_id( + request: Request, + credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)], +) -> UUID: + candidate = credentials.credentials if credentials else "" + matched = None + # Compare every configured token rather than exposing a match's position. + for attendee_id, token in request.app.state.settings.bearer_tokens.items(): + if secrets.compare_digest(candidate.encode(), token.encode()): + matched = attendee_id + if matched is None: + raise HTTPException( + status_code=401, + detail="A valid attendee bearer token is required.", + headers={"WWW-Authenticate": "Bearer"}, + ) + return matched diff --git a/applications/workshop-booking/app/config.py b/applications/workshop-booking/app/config.py new file mode 100644 index 00000000..bbe7c3c1 --- /dev/null +++ b/applications/workshop-booking/app/config.py @@ -0,0 +1,95 @@ +"""Explicit libpq connection settings and server-owned attendee credentials.""" + +import json +import os +import re +from dataclasses import dataclass, field +from pathlib import Path +from uuid import UUID + +from dotenv import load_dotenv + + +class ConfigurationError(ValueError): + """A safe-to-display configuration error that never includes secret values.""" + + +@dataclass(frozen=True) +class Settings: + host: str + port: int + database: str + user: str + password: str = field(repr=False) + sslrootcert: Path + bearer_tokens: dict[UUID, str] = field(repr=False) + + @classmethod + def from_env(cls, require_tokens: bool = True) -> "Settings": + # Only the example's .env is loaded; existing environment values win. + load_dotenv(Path.cwd() / ".env", override=False) + + def required(name: str) -> str: + value = os.environ.get(name, "") + if not value or "\x00" in value: + raise ConfigurationError(f"Set {name} to a nonempty value.") + return value + + host = required("PGHOST") + if not re.fullmatch(r"[A-Za-z0-9.-]+", host): + raise ConfigurationError("PGHOST must be a single DNS hostname.") + try: + port = int(os.environ.get("PGPORT", "5432")) + if not 1 <= port <= 65535: + raise ValueError + except ValueError: + raise ConfigurationError("PGPORT must be between 1 and 65535.") from None + + cert = Path(required("PGSSLROOTCERT")).expanduser().resolve() + if not cert.is_file(): + raise ConfigurationError("PGSSLROOTCERT must point to the downloaded CA file.") + + tokens = parse_tokens(required("API_BEARER_TOKENS")) if require_tokens else {} + return cls( + host=host, + port=port, + database=required("PGDATABASE"), + user=required("PGUSER"), + password=required("PGPASSWORD"), + sslrootcert=cert, + bearer_tokens=tokens, + ) + + +def parse_tokens(raw: str) -> dict[UUID, str]: + """Map stable, seeded attendee UUIDs to independently generated bearer tokens.""" + message = ( + "API_BEARER_TOKENS must be a JSON object of unique attendee UUIDs to unique " + "32–256 character tokens using letters, digits, underscores or hyphens." + ) + + def unique_pairs(pairs: list[tuple[str, object]]) -> dict[str, object]: + result = {} + for key, value in pairs: + if key in result: + raise ValueError + result[key] = value + return result + + try: + parsed = json.loads(raw, object_pairs_hook=unique_pairs) + if not isinstance(parsed, dict) or not 1 <= len(parsed) <= 100: + raise ValueError + tokens = {} + for attendee, token in parsed.items(): + attendee_id = UUID(attendee) + if attendee_id in tokens or not isinstance(token, str): + raise ValueError + if not re.fullmatch(r"[A-Za-z0-9_-]{32,256}", token): + raise ValueError + tokens[attendee_id] = token + if len(set(tokens.values())) != len(tokens): + raise ValueError + return tokens + except (ValueError, TypeError, AttributeError): + raise ConfigurationError(message) from None diff --git a/applications/workshop-booking/app/database.py b/applications/workshop-booking/app/database.py new file mode 100644 index 00000000..187942e3 --- /dev/null +++ b/applications/workshop-booking/app/database.py @@ -0,0 +1,47 @@ +"""One bounded SQLAlchemy connection pool per application process.""" + +from collections.abc import Iterator + +from fastapi import Request +from sqlalchemy import URL, Engine, create_engine +from sqlalchemy.orm import Session + +from app.config import Settings + + +def build_engine(settings: Settings) -> Engine: + # URL.create keeps special characters in passwords out of URI parsing. + url = URL.create( + "postgresql+psycopg", + username=settings.user, + password=settings.password, + host=settings.host, + port=settings.port, + database=settings.database, + ) + return create_engine( + url, + connect_args={ + "sslmode": "verify-full", + "sslrootcert": str(settings.sslrootcert), + "connect_timeout": 10, + "application_name": "workshop-booking", + "options": ( + "-c statement_timeout=5000 -c lock_timeout=3000 " + "-c idle_in_transaction_session_timeout=10000" + ), + }, + pool_size=5, + max_overflow=5, + pool_timeout=5, + pool_pre_ping=True, + pool_recycle=300, + isolation_level="READ COMMITTED", + hide_parameters=True, + ) + + +def get_session(request: Request) -> Iterator[Session]: + # Routes own the transaction so commit finishes before a success is returned. + with request.app.state.session_factory() as session: + yield session diff --git a/applications/workshop-booking/app/main.py b/applications/workshop-booking/app/main.py new file mode 100644 index 00000000..9702db3c --- /dev/null +++ b/applications/workshop-booking/app/main.py @@ -0,0 +1,148 @@ +import logging +from contextlib import asynccontextmanager +from typing import Annotated +from uuid import UUID + +from fastapi import Depends, FastAPI, HTTPException, Query, Request, Response +from fastapi.exceptions import RequestValidationError +from fastapi.responses import JSONResponse +from sqlalchemy import select +from sqlalchemy.exc import IntegrityError, SQLAlchemyError +from sqlalchemy.orm import Session, sessionmaker + +from app.auth import get_attendee_id +from app.config import ConfigurationError, Settings +from app.database import build_engine, get_session +from app.middleware import BookingBodyLimit +from app.models import Attendee +from app.schemas import BookingCreate, BookingResponse, WorkshopResponse +from app.services import cancel_booking, create_booking, get_booking, list_workshops + +logger = logging.getLogger("workshop_booking") +DatabaseSession = Annotated[Session, Depends(get_session)] +CurrentAttendee = Annotated[UUID, Depends(get_attendee_id)] + + +def create_app( + settings: Settings | None = None, session_factory: sessionmaker | None = None +) -> FastAPI: + @asynccontextmanager + async def lifespan(application: FastAPI): + configured = settings or Settings.from_env() + engine = None + factory = session_factory + if factory is None: + engine = build_engine(configured) + factory = sessionmaker(engine, expire_on_commit=False) + try: + # A typo in identity configuration must fail at startup, not impersonate + # an arbitrary client-selected attendee during a request. + with factory() as session: + found = set( + session.scalars( + select(Attendee.id).where(Attendee.id.in_(configured.bearer_tokens)) + ) + ) + if not configured.bearer_tokens or found != set(configured.bearer_tokens): + raise ConfigurationError("Every configured attendee must exist in the seed data.") + application.state.settings = configured + application.state.session_factory = factory + yield + except SQLAlchemyError: + raise RuntimeError( + "Database startup check failed. Check connection and migrations." + ) from None + finally: + if engine is not None: + engine.dispose() + + application = FastAPI( + title="Workshop Booking API", + version="1.0.0", + description=( + "Reserve one seat in a workshop using a client-generated booking UUID. " + "Reuse the same UUID to safely retry a request. Configure attendee tokens " + "on the server, then use the Authorize button for booking routes." + ), + lifespan=lifespan, + ) + application.add_middleware(BookingBodyLimit) + + @application.exception_handler(SQLAlchemyError) + async def database_error(_request: Request, error: SQLAlchemyError) -> JSONResponse: + # Never log the exception string: it can contain SQL, parameters or credentials. + logger.warning("Database operation failed (%s)", type(error).__name__) + return JSONResponse( + status_code=503, + content={ + "detail": "Database temporarily unavailable. Retry using the same booking ID." + }, + headers={"Retry-After": "1"}, + ) + + @application.exception_handler(RequestValidationError) + async def validation_error(_request: Request, error: RequestValidationError) -> JSONResponse: + # Keep useful field errors without reflecting submitted bodies or secret values. + return JSONResponse( + status_code=422, + content={ + "detail": [ + {"loc": list(item["loc"]), "msg": item["msg"], "type": item["type"]} + for item in error.errors() + ] + }, + ) + + @application.get("/workshops", response_model=list[WorkshopResponse], tags=["workshops"]) + def workshops( + session: DatabaseSession, + limit: Annotated[int, Query(ge=1, le=100)] = 20, + offset: Annotated[int, Query(ge=0, le=10000)] = 0, + ): + """List upcoming workshops with current seat availability.""" + with session.begin(): + return list_workshops(session, limit, offset) + + @application.post( + "/bookings", + response_model=BookingResponse, + status_code=201, + responses={200: {"description": "Existing booking returned for an idempotent retry."}}, + tags=["bookings"], + ) + def book( + body: BookingCreate, + response: Response, + attendee_id: CurrentAttendee, + session: DatabaseSession, + ): + """Reserve one seat. A cancelled UUID stays cancelled; rebook with a new UUID.""" + try: + with session.begin(): + booking, created = create_booking(session, attendee_id, body) + except IntegrityError as error: + # Includes the rare case of one UUID raced across two different workshops. + if getattr(error.orig, "sqlstate", None) == "23505": + raise HTTPException( + 409, "Booking ID or active booking has already been used." + ) from None + raise + response.status_code = 201 if created else 200 + return booking + + @application.get("/bookings/{booking_id}", response_model=BookingResponse, tags=["bookings"]) + def booking(booking_id: UUID, attendee_id: CurrentAttendee, session: DatabaseSession): + """Read your own booking; other attendees' bookings return 404.""" + with session.begin(): + return get_booking(session, attendee_id, booking_id) + + @application.delete("/bookings/{booking_id}", response_model=BookingResponse, tags=["bookings"]) + def cancel(booking_id: UUID, attendee_id: CurrentAttendee, session: DatabaseSession): + """Cancel before the workshop starts. Repeated cancellation is safe.""" + with session.begin(): + return cancel_booking(session, attendee_id, booking_id) + + return application + + +app = create_app() diff --git a/applications/workshop-booking/app/middleware.py b/applications/workshop-booking/app/middleware.py new file mode 100644 index 00000000..3f646d58 --- /dev/null +++ b/applications/workshop-booking/app/middleware.py @@ -0,0 +1,40 @@ +"""Bound JSON request bodies before FastAPI parses them, including streamed bodies.""" + +from starlette.responses import JSONResponse +from starlette.types import ASGIApp, Receive, Scope, Send + + +class BookingBodyLimit: + def __init__(self, app: ASGIApp, max_bytes: int = 4096): + self.app = app + self.max_bytes = max_bytes + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] != "http" or scope["method"] != "POST" or scope["path"] != "/bookings": + await self.app(scope, receive, send) + return + + body = bytearray() + while True: + message = await receive() + if message["type"] == "http.disconnect": + return + chunk = message.get("body", b"") + if len(body) + len(chunk) > self.max_bytes: + response = JSONResponse({"detail": "Request body exceeds 4096 bytes."}, 413) + await response(scope, receive, send) + return + body.extend(chunk) + if not message.get("more_body", False): + break + + sent = False + + async def buffered_receive(): + nonlocal sent + if not sent: + sent = True + return {"type": "http.request", "body": bytes(body), "more_body": False} + return await receive() + + await self.app(scope, buffered_receive, send) diff --git a/applications/workshop-booking/app/models.py b/applications/workshop-booking/app/models.py new file mode 100644 index 00000000..4d6ef6be --- /dev/null +++ b/applications/workshop-booking/app/models.py @@ -0,0 +1,52 @@ +"""ORM mapping. The checked-in SQL migrations own schema creation.""" + +from datetime import datetime +from uuid import UUID + +from sqlalchemy import DateTime, ForeignKey, Integer, MetaData, String, Text, func +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column + + +class Base(DeclarativeBase): + metadata = MetaData(schema="workshop_booking") + + +class Workshop(Base): + __tablename__ = "workshops" + + id: Mapped[UUID] = mapped_column(primary_key=True) + title: Mapped[str] = mapped_column(String(160)) + description: Mapped[str] = mapped_column(Text) + starts_at: Mapped[datetime] = mapped_column(DateTime(timezone=True)) + duration_minutes: Mapped[int] = mapped_column(Integer) + capacity: Mapped[int] = mapped_column(Integer) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.clock_timestamp() + ) + + +class Attendee(Base): + __tablename__ = "attendees" + + id: Mapped[UUID] = mapped_column(primary_key=True) + name: Mapped[str] = mapped_column(String(120)) + email: Mapped[str] = mapped_column(String(254), unique=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.clock_timestamp() + ) + + +class Booking(Base): + __tablename__ = "bookings" + + id: Mapped[UUID] = mapped_column(primary_key=True) + workshop_id: Mapped[UUID] = mapped_column(ForeignKey("workshop_booking.workshops.id")) + attendee_id: Mapped[UUID] = mapped_column(ForeignKey("workshop_booking.attendees.id")) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.clock_timestamp() + ) + cancelled_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + + @property + def status(self) -> str: + return "cancelled" if self.cancelled_at is not None else "confirmed" diff --git a/applications/workshop-booking/app/schemas.py b/applications/workshop-booking/app/schemas.py new file mode 100644 index 00000000..78753b62 --- /dev/null +++ b/applications/workshop-booking/app/schemas.py @@ -0,0 +1,35 @@ +from datetime import datetime +from typing import Literal +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field + + +class BookingCreate(BaseModel): + model_config = ConfigDict(extra="forbid") + + id: UUID = Field( + description="A new client-generated UUID; reuse it when retrying this booking." + ) + workshop_id: UUID + + +class BookingResponse(BaseModel): + model_config = ConfigDict(from_attributes=True) + + id: UUID + workshop_id: UUID + attendee_id: UUID + status: Literal["confirmed", "cancelled"] + created_at: datetime + cancelled_at: datetime | None + + +class WorkshopResponse(BaseModel): + id: UUID + title: str + description: str + starts_at: datetime + duration_minutes: int + capacity: int + available_seats: int diff --git a/applications/workshop-booking/app/services.py b/applications/workshop-booking/app/services.py new file mode 100644 index 00000000..3adac171 --- /dev/null +++ b/applications/workshop-booking/app/services.py @@ -0,0 +1,117 @@ +"""The seat invariant lives in these short, explicit database transactions.""" + +from uuid import UUID + +from fastapi import HTTPException +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from app.models import Booking, Workshop +from app.schemas import BookingCreate, BookingResponse, WorkshopResponse + + +def list_workshops(session: Session, limit: int, offset: int) -> list[WorkshopResponse]: + active_count = ( + select(func.count(Booking.id)) + .where(Booking.workshop_id == Workshop.id, Booking.cancelled_at.is_(None)) + .correlate(Workshop) + .scalar_subquery() + ) + rows = session.execute( + select(Workshop, active_count.label("booked")) + .where(Workshop.starts_at > func.statement_timestamp()) + .order_by(Workshop.starts_at, Workshop.id) + .limit(limit) + .offset(offset) + ) + return [ + WorkshopResponse( + id=workshop.id, + title=workshop.title, + description=workshop.description, + starts_at=workshop.starts_at, + duration_minutes=workshop.duration_minutes, + capacity=workshop.capacity, + available_seats=workshop.capacity - booked, + ) + for workshop, booked in rows + ] + + +def create_booking( + session: Session, attendee_id: UUID, request: BookingCreate +) -> tuple[BookingResponse, bool]: + # Every capacity-changing transaction locks the same parent row first. + workshop = session.scalar( + select(Workshop).where(Workshop.id == request.workshop_id).with_for_update() + ) + if workshop is None: + raise HTTPException(404, "Workshop not found.") + + existing = session.get(Booking, request.id) + if existing is not None: + if existing.attendee_id != attendee_id or existing.workshop_id != request.workshop_id: + raise HTTPException(409, "Booking ID has already been used.") + # Retrying a cancelled booking never creates a new reservation. + return BookingResponse.model_validate(existing), False + + # CURRENT_TIMESTAMP is fixed at transaction start; use actual time after waiting. + now = session.scalar(select(func.clock_timestamp())) + if workshop.starts_at <= now: + raise HTTPException(409, "This workshop has already started.") + duplicate = session.scalar( + select(Booking.id).where( + Booking.workshop_id == workshop.id, + Booking.attendee_id == attendee_id, + Booking.cancelled_at.is_(None), + ) + ) + if duplicate is not None: + raise HTTPException(409, "You already have an active booking for this workshop.") + booked = session.scalar( + select(func.count(Booking.id)).where( + Booking.workshop_id == workshop.id, Booking.cancelled_at.is_(None) + ) + ) + if booked >= workshop.capacity: + raise HTTPException(409, "This workshop is full.") + + booking = Booking(id=request.id, workshop_id=workshop.id, attendee_id=attendee_id) + session.add(booking) + session.flush() + return BookingResponse.model_validate(booking), True + + +def get_booking(session: Session, attendee_id: UUID, booking_id: UUID) -> BookingResponse: + booking = session.scalar( + select(Booking).where(Booking.id == booking_id, Booking.attendee_id == attendee_id) + ) + if booking is None: + raise HTTPException(404, "Booking not found.") + return BookingResponse.model_validate(booking) + + +def cancel_booking(session: Session, attendee_id: UUID, booking_id: UUID) -> BookingResponse: + # Read only the parent ID here: acquiring a booking lock first could deadlock + # with a create transaction. Workshop and owner IDs are never changed by the API. + workshop_id = session.scalar( + select(Booking.workshop_id).where( + Booking.id == booking_id, Booking.attendee_id == attendee_id + ) + ) + if workshop_id is None: + raise HTTPException(404, "Booking not found.") + workshop = session.scalar(select(Workshop).where(Workshop.id == workshop_id).with_for_update()) + booking = session.scalar( + select(Booking) + .where(Booking.id == booking_id, Booking.attendee_id == attendee_id) + .with_for_update() + ) + if booking.cancelled_at is not None: + return BookingResponse.model_validate(booking) + now = session.scalar(select(func.clock_timestamp())) + if workshop.starts_at <= now: + raise HTTPException(409, "This workshop has already started.") + booking.cancelled_at = now + session.flush() + return BookingResponse.model_validate(booking) diff --git a/applications/workshop-booking/migrations/env.py b/applications/workshop-booking/migrations/env.py new file mode 100644 index 00000000..b78bc18e --- /dev/null +++ b/applications/workshop-booking/migrations/env.py @@ -0,0 +1,42 @@ +"""Run SQL migrations with the separate migrator login and schema owner role.""" + +from alembic import context +from sqlalchemy import text + +from app.config import Settings +from app.database import build_engine +from app.models import Base + + +def run_migrations_online() -> None: + settings = Settings.from_env(require_tokens=False) + if settings.user != "workshop_booking_migrator": + raise RuntimeError("Run Alembic with PGUSER=workshop_booking_migrator.") + + engine = build_engine(settings) + try: + with engine.connect() as connection: + # The login cannot create objects until it deliberately assumes the + # non-login owner. Keep this and all DDL in a single transaction. + with connection.begin(): + connection.execute(text("SET LOCAL ROLE workshop_booking_owner")) + context.configure( + connection=connection, + target_metadata=Base.metadata, + version_table_schema="workshop_booking", + include_schemas=True, + transactional_ddl=True, + ) + with context.begin_transaction(): + context.run_migrations() + finally: + engine.dispose() + + +if context.is_offline_mode(): + raise RuntimeError( + "Offline migrations are disabled. Inspect sql/migrations/*.sql, then run " + "Alembic against the database with migrator credentials." + ) +else: + run_migrations_online() diff --git a/applications/workshop-booking/migrations/script.py.mako b/applications/workshop-booking/migrations/script.py.mako new file mode 100644 index 00000000..e326ad21 --- /dev/null +++ b/applications/workshop-booking/migrations/script.py.mako @@ -0,0 +1,18 @@ +"""${message}""" + +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + +revision = ${repr(up_revision)} +down_revision = ${repr(down_revision)} +branch_labels = ${repr(branch_labels)} +depends_on = ${repr(depends_on)} + + +def upgrade() -> None: + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + ${downgrades if downgrades else "pass"} diff --git a/applications/workshop-booking/migrations/versions/001_initial.py b/applications/workshop-booking/migrations/versions/001_initial.py new file mode 100644 index 00000000..97961514 --- /dev/null +++ b/applications/workshop-booking/migrations/versions/001_initial.py @@ -0,0 +1,22 @@ +"""Create the three booking tables and explicit runtime grants.""" + +from pathlib import Path + +from alembic import op + +revision = "001_initial" +down_revision = None +branch_labels = None +depends_on = None + +SQL_DIRECTORY = Path(__file__).resolve().parents[2] / "sql" + + +def upgrade() -> None: + # These files are intentionally readable without knowing Alembic or Python. + op.execute((SQL_DIRECTORY / "migrations" / "001_initial.sql").read_text()) + op.execute((SQL_DIRECTORY / "grants.sql").read_text()) + + +def downgrade() -> None: + op.execute((SQL_DIRECTORY / "migrations" / "001_initial_down.sql").read_text()) diff --git a/applications/workshop-booking/pyproject.toml b/applications/workshop-booking/pyproject.toml new file mode 100644 index 00000000..ec03bb4a --- /dev/null +++ b/applications/workshop-booking/pyproject.toml @@ -0,0 +1,18 @@ +[project] +name = "workshop-booking" +version = "1.0.0" +description = "A FastAPI workshop booking example for ClickHouse Managed Postgres" +requires-python = ">=3.12" + +[tool.pytest.ini_options] +testpaths = ["tests/unit"] +pythonpath = ["."] +addopts = "-ra" +markers = ["integration: requires the explicitly configured managed Postgres test database"] + +[tool.ruff] +target-version = "py312" +line-length = 100 + +[tool.ruff.lint] +select = ["E", "F", "I", "UP", "B"] diff --git a/applications/workshop-booking/requirements-dev.txt b/applications/workshop-booking/requirements-dev.txt new file mode 100644 index 00000000..d6267209 --- /dev/null +++ b/applications/workshop-booking/requirements-dev.txt @@ -0,0 +1,14 @@ +-r requirements.txt + +# Direct development dependencies. +httpx==0.28.1 +pytest==9.1.1 +ruff==0.16.8 + +# Transitive development dependencies. +certifi==2026.7.22 +httpcore==1.0.9 +iniconfig==2.3.0 +packaging==26.3 +pluggy==1.6.0 +Pygments==2.21.0 diff --git a/applications/workshop-booking/requirements.txt b/applications/workshop-booking/requirements.txt new file mode 100644 index 00000000..7fc57eac --- /dev/null +++ b/applications/workshop-booking/requirements.txt @@ -0,0 +1,24 @@ +# Direct dependencies. Resolved and tested on Python 3.12, 2026-09-17. +fastapi==0.141.1 +pydantic==2.13.5 +psycopg[binary]==3.3.5 +SQLAlchemy==2.0.54 +uvicorn==0.53.0 +python-dotenv==1.2.3 +alembic==1.20.0 + +# Transitive runtime dependencies, pinned for reproducible installs. +annotated-doc==0.0.5 +annotated-types==0.8.0 +anyio==4.15.1 +click==8.5.0 +greenlet==3.5.6 +h11==0.16.0 +idna==3.19 +Mako==1.4.1 +MarkupSafe==3.0.3 +psycopg-binary==3.3.5 +pydantic_core==2.46.5 +starlette==1.6.0 +typing-inspection==0.4.4 +typing_extensions==4.16.0 diff --git a/applications/workshop-booking/sql/bootstrap.sql b/applications/workshop-booking/sql/bootstrap.sql new file mode 100644 index 00000000..6a99aecd --- /dev/null +++ b/applications/workshop-booking/sql/bootstrap.sql @@ -0,0 +1,55 @@ +-- Run once as the service administrator, before Alembic. +-- psql -X --set=ON_ERROR_STOP=1 --file=sql/bootstrap.sql +-- Passwords come from the environment, never from command-line arguments. +\set ON_ERROR_STOP on +\getenv migrator_password WORKSHOP_MIGRATOR_PASSWORD +\getenv app_password WORKSHOP_APP_PASSWORD +\if :{?migrator_password} +\else + DO $$ BEGIN RAISE EXCEPTION 'Set WORKSHOP_MIGRATOR_PASSWORD before running bootstrap.sql.'; END $$; +\endif +\if :{?app_password} +\else + DO $$ BEGIN RAISE EXCEPTION 'Set WORKSHOP_APP_PASSWORD before running bootstrap.sql.'; END $$; +\endif +SELECT length(:'migrator_password') >= 24 AND length(:'app_password') >= 24 + AND :'migrator_password' <> :'app_password' AS passwords_valid \gset +\if :passwords_valid +\else + DO $$ BEGIN RAISE EXCEPTION 'Use distinct, randomly generated passwords of at least 24 characters.'; END $$; +\endif + +BEGIN; + +-- This is deliberately run-once: existing names abort the transaction instead +-- of changing ownership or rotating credentials for an unrelated application. +-- New roles have no superuser, database-create, role-create, replication or +-- bypass-RLS privileges by default; no superuser-only ALTER is necessary. +CREATE ROLE workshop_booking_owner NOLOGIN; +CREATE ROLE workshop_booking_migrator LOGIN NOINHERIT; +CREATE ROLE workshop_booking_app LOGIN NOINHERIT; + +SELECT format('ALTER ROLE workshop_booking_migrator PASSWORD %L', :'migrator_password') \gexec +SELECT format('ALTER ROLE workshop_booking_app PASSWORD %L', :'app_password') \gexec + +GRANT workshop_booking_owner TO workshop_booking_migrator; +-- The administrator needs membership to create a schema owned by this role. +GRANT workshop_booking_owner TO CURRENT_USER; +SELECT format('GRANT CONNECT ON DATABASE %I TO workshop_booking_migrator, workshop_booking_app', current_database()) \gexec + +CREATE SCHEMA workshop_booking AUTHORIZATION workshop_booking_owner; +REVOKE ALL ON SCHEMA workshop_booking FROM PUBLIC; + +ALTER ROLE workshop_booking_app SET search_path = pg_catalog, workshop_booking; +ALTER ROLE workshop_booking_app SET statement_timeout = '10s'; +ALTER ROLE workshop_booking_app SET lock_timeout = '5s'; +ALTER ROLE workshop_booking_app SET idle_in_transaction_session_timeout = '15s'; +ALTER ROLE workshop_booking_migrator SET search_path = pg_catalog, workshop_booking; + +COMMIT; + +-- A runtime role can inherit permissions granted to PUBLIC. Review these before +-- sharing the database with another application; this script does not revoke +-- privileges that other applications may rely on. +SELECT has_schema_privilege('workshop_booking_app', 'public', 'CREATE') + AS runtime_can_create_in_public; diff --git a/applications/workshop-booking/sql/cleanup.sql b/applications/workshop-booking/sql/cleanup.sql new file mode 100644 index 00000000..9ac87192 --- /dev/null +++ b/applications/workshop-booking/sql/cleanup.sql @@ -0,0 +1,11 @@ +-- DESTRUCTIVE: run as the service administrator only when finished with this +-- example. Stop the API first. This removes all example data, schema, and roles. +\set ON_ERROR_STOP on +BEGIN; +DROP SCHEMA IF EXISTS workshop_booking CASCADE; +-- Remove only this example's explicit CONNECT grants before dropping the roles. +SELECT format('REVOKE CONNECT ON DATABASE %I FROM workshop_booking_migrator, workshop_booking_app', current_database()) \gexec +DROP ROLE workshop_booking_app; +DROP ROLE workshop_booking_migrator; +DROP ROLE workshop_booking_owner; +COMMIT; diff --git a/applications/workshop-booking/sql/grants.sql b/applications/workshop-booking/sql/grants.sql new file mode 100644 index 00000000..97db0d0d --- /dev/null +++ b/applications/workshop-booking/sql/grants.sql @@ -0,0 +1,14 @@ +-- Run as workshop_booking_owner (Alembic applies this after the initial DDL). +-- No blanket default privileges: review runtime access in each new migration. +REVOKE ALL ON ALL TABLES IN SCHEMA workshop_booking FROM PUBLIC; +REVOKE ALL ON ALL TABLES IN SCHEMA workshop_booking FROM workshop_booking_app; +GRANT USAGE ON SCHEMA workshop_booking TO workshop_booking_app; +GRANT SELECT ON workshop_booking.workshops, workshop_booking.attendees, + workshop_booking.bookings TO workshop_booking_app; +GRANT INSERT ON workshop_booking.bookings TO workshop_booking_app; +GRANT UPDATE (cancelled_at) ON workshop_booking.bookings TO workshop_booking_app; + +-- SELECT ... FOR UPDATE needs UPDATE privilege on at least one column. +-- Grant only id; the application never changes it. Capacity, workshop details, +-- attendee details, booking identity, DELETE, and schema DDL remain restricted. +GRANT UPDATE (id) ON workshop_booking.workshops TO workshop_booking_app; diff --git a/applications/workshop-booking/sql/migrations/001_initial.sql b/applications/workshop-booking/sql/migrations/001_initial.sql new file mode 100644 index 00000000..a55f7032 --- /dev/null +++ b/applications/workshop-booking/sql/migrations/001_initial.sql @@ -0,0 +1,45 @@ +-- Alembic executes this file in its migration transaction after SET ROLE +-- workshop_booking_owner. Schema creation belongs to sql/bootstrap.sql. +CREATE TABLE workshop_booking.workshops ( + id uuid PRIMARY KEY, + title varchar(160) NOT NULL, + description text NOT NULL DEFAULT '', + starts_at timestamptz NOT NULL, + duration_minutes integer NOT NULL, + capacity integer NOT NULL, + created_at timestamptz NOT NULL DEFAULT clock_timestamp(), + CONSTRAINT workshops_title_not_blank CHECK (length(btrim(title)) > 0), + CONSTRAINT workshops_capacity_positive CHECK (capacity BETWEEN 1 AND 10000), + CONSTRAINT workshops_duration_positive CHECK (duration_minutes BETWEEN 1 AND 1440) +); + +CREATE INDEX workshops_starts_at_id ON workshop_booking.workshops (starts_at, id); + +CREATE TABLE workshop_booking.attendees ( + id uuid PRIMARY KEY, + name varchar(120) NOT NULL, + email varchar(254) NOT NULL UNIQUE, + created_at timestamptz NOT NULL DEFAULT clock_timestamp(), + CONSTRAINT attendees_name_not_blank CHECK (length(btrim(name)) > 0), + CONSTRAINT attendees_email_not_blank CHECK (length(btrim(email)) > 0) +); + +CREATE TABLE workshop_booking.bookings ( + -- The caller generates this UUID once and reuses it after network failures. + id uuid PRIMARY KEY, + workshop_id uuid NOT NULL REFERENCES workshop_booking.workshops(id), + attendee_id uuid NOT NULL REFERENCES workshop_booking.attendees(id), + created_at timestamptz NOT NULL DEFAULT clock_timestamp(), + cancelled_at timestamptz, + CONSTRAINT bookings_cancellation_after_creation CHECK ( + cancelled_at IS NULL OR cancelled_at >= created_at + ) +); + +-- A cancelled row remains as a receipt. A new UUID may book again, but there +-- can never be two active bookings for the same attendee and workshop. +CREATE UNIQUE INDEX bookings_one_active_per_attendee_workshop + ON workshop_booking.bookings (workshop_id, attendee_id) + WHERE cancelled_at IS NULL; + +CREATE INDEX bookings_attendee_id ON workshop_booking.bookings (attendee_id); diff --git a/applications/workshop-booking/sql/migrations/001_initial_down.sql b/applications/workshop-booking/sql/migrations/001_initial_down.sql new file mode 100644 index 00000000..3ee85559 --- /dev/null +++ b/applications/workshop-booking/sql/migrations/001_initial_down.sql @@ -0,0 +1,5 @@ +-- Destructive: this removes every booking, attendee, and workshop. +-- Alembic wraps this downgrade in a transaction and retains its version table. +DROP TABLE workshop_booking.bookings; +DROP TABLE workshop_booking.attendees; +DROP TABLE workshop_booking.workshops; diff --git a/applications/workshop-booking/sql/seed.sql b/applications/workshop-booking/sql/seed.sql new file mode 100644 index 00000000..65700472 --- /dev/null +++ b/applications/workshop-booking/sql/seed.sql @@ -0,0 +1,26 @@ +-- Run after migrations using the migrator credentials. +-- Re-running keeps existing workshop dates and bookings intact. +\set ON_ERROR_STOP on +BEGIN; +SET LOCAL ROLE workshop_booking_owner; + +INSERT INTO workshop_booking.attendees (id, name, email) VALUES + ('11111111-1111-4111-8111-111111111111', 'Alex Morgan', 'alex@example.test'), + ('22222222-2222-4222-8222-222222222222', 'Sam Taylor', 'sam@example.test') +ON CONFLICT (id) DO NOTHING; + +INSERT INTO workshop_booking.workshops + (id, title, description, starts_at, duration_minutes, capacity) +VALUES + ('aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa', 'Print your first linocut', + 'Design, carve, and print a two-color postcard. Tools and materials included.', + date_trunc('day', clock_timestamp() AT TIME ZONE 'UTC') AT TIME ZONE 'UTC' + interval '7 days 10 hours', 120, 8), + ('bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb', 'Wheel throwing for beginners', + 'A small group introduction to centering clay and throwing a bowl.', + date_trunc('day', clock_timestamp() AT TIME ZONE 'UTC') AT TIME ZONE 'UTC' + interval '8 days 14 hours', 90, 6), + ('cccccccc-cccc-4ccc-8ccc-cccccccccccc', 'Build a terrarium', + 'Create a miniature garden and learn how to keep it thriving.', + date_trunc('day', clock_timestamp() AT TIME ZONE 'UTC') AT TIME ZONE 'UTC' + interval '10 days 11 hours', 75, 1) +ON CONFLICT (id) DO NOTHING; + +COMMIT; diff --git a/applications/workshop-booking/tests/README.md b/applications/workshop-booking/tests/README.md new file mode 100644 index 00000000..d28cb7c4 --- /dev/null +++ b/applications/workshop-booking/tests/README.md @@ -0,0 +1,63 @@ +# Tests + +Run these commands inside your Linux development environment from the example +directory, after installing `requirements-dev.txt` into `.venv`. + +```bash +.venv/bin/pytest tests/unit -q +.venv/bin/ruff check app migrations tests +.venv/bin/ruff format --check app migrations tests +``` + +## Live Postgres tests + +Use a dedicated ClickHouse Managed Postgres service, complete the README's +bootstrap and Alembic steps, and load the seed data. The tests call the FastAPI +application through its HTTP test client while using real, separately committed +Postgres transactions. SQLite and mocked database connections are not used. + +Set `PGHOST`, `PGPORT`, `PGDATABASE`, `PGUSER=workshop_booking_app`, `PGPASSWORD`, +and `PGSSLROOTCERT` as described in the main README. Also set +`WORKSHOP_MIGRATOR_PASSWORD` so fixtures can create and remove their own rows. All +connections use `sslmode=verify-full`. The tests create their own temporary +attendee token mapping; `API_BEARER_TOKENS` is not needed for these tests. + +If you keep these values in a private, shell-compatible file, for example +`.deployment/test.env`, restrict its permissions and explicitly export it: + +```bash +chmod 600 .deployment/test.env +set -a +source .deployment/test.env +set +a +.venv/bin/pytest tests/integration -q +``` + +The explicitly selected integration suite **fails** if credentials are missing. +It creates fresh UUIDs for all fixture attendees and workshops, and removes only +those rows when it finishes. It does not truncate tables or change sample rows. +An interrupted process may leave fixture rows behind; use the dedicated test +service for that reason. Do not run this suite against customer data. + +The suite checks last-seat contention, repeated and concurrent cancellation, +booking UUID retries, ownership, start times after waiting for locks, transaction +rollback, database constraints, runtime role restrictions, TLS use, invalid CA +rejection, and hostname verification. Concurrency assertions inspect committed +database state as well as HTTP responses. + +## Migration lifecycle + +Run this separately, **after bootstrap and before seeding**. It upgrades to the +latest revision, checks ownership, downgrades to an empty schema, upgrades again, +and checks that upgrading an already current database is harmless. + +```bash +WORKSHOP_ALLOW_SCHEMA_RESET=1 .venv/bin/pytest tests/migrations -q +``` + +This command uses the same exported connection settings and +`WORKSHOP_MIGRATOR_PASSWORD`. It assumes the migrator role itself. The opt-in is +required, and a second safety check refuses to reset a schema containing any +application rows or unexpected tables. Run `sql/seed.sql` after this test. + +Keep local logs and validation receipts outside the example directory. diff --git a/applications/workshop-booking/tests/integration/conftest.py b/applications/workshop-booking/tests/integration/conftest.py new file mode 100644 index 00000000..6aa2b47f --- /dev/null +++ b/applications/workshop-booking/tests/integration/conftest.py @@ -0,0 +1,142 @@ +"""Live Postgres fixtures. Missing credentials are a failure, never a skip.""" + +import os +import secrets +from dataclasses import dataclass, replace +from datetime import UTC, datetime, timedelta +from uuid import UUID, uuid4 + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import Engine, text +from sqlalchemy.orm import sessionmaker + +from app.config import Settings +from app.database import build_engine +from app.main import create_app + + +@dataclass +class LiveDatabase: + settings: Settings + engine: Engine + owner_engine: Engine + attendees: tuple[UUID, UUID, UUID] + tokens: dict[UUID, str] + workshops: list[UUID] + + def headers(self, attendee: int = 0) -> dict[str, str]: + return {"Authorization": f"Bearer {self.tokens[self.attendees[attendee]]}"} + + def active_count(self, workshop_id: UUID) -> int: + with self.engine.connect() as connection: + return connection.scalar( + text( + "SELECT count(*) FROM workshop_booking.bookings " + "WHERE workshop_id = :id AND cancelled_at IS NULL" + ), + {"id": workshop_id}, + ) + + +@pytest.fixture(scope="session") +def live_db() -> LiveDatabase: + required = ( + "PGHOST", + "PGDATABASE", + "PGUSER", + "PGPASSWORD", + "PGSSLROOTCERT", + "WORKSHOP_MIGRATOR_PASSWORD", + ) + missing = [name for name in required if not os.environ.get(name)] + if missing: + pytest.fail("Live integration tests require: " + ", ".join(missing)) + settings = Settings.from_env(require_tokens=False) + if settings.user != "workshop_booking_app": + pytest.fail("Run integration tests with PGUSER=workshop_booking_app.") + + attendees = (uuid4(), uuid4(), uuid4()) + tokens = {attendee_id: secrets.token_urlsafe(32) for attendee_id in attendees} + settings = replace(settings, bearer_tokens=tokens) + engine = build_engine(settings) + owner_engine = build_engine( + replace( + settings, + user="workshop_booking_migrator", + password=os.environ["WORKSHOP_MIGRATOR_PASSWORD"], + ) + ) + database = LiveDatabase(settings, engine, owner_engine, attendees, tokens, []) + try: + with owner_engine.begin() as connection: + connection.execute(text("SET LOCAL ROLE workshop_booking_owner")) + for index, attendee_id in enumerate(attendees): + connection.execute( + text( + "INSERT INTO workshop_booking.attendees (id, name, email) " + "VALUES (:id, :name, :email)" + ), + { + "id": attendee_id, + "name": f"Integration attendee {index}", + "email": f"integration-{attendee_id}@example.test", + }, + ) + yield database + finally: + # Test-created rows are identified by fresh UUIDs, never by broad names + # or a table truncate. A failed test does not touch the sample data. + with owner_engine.begin() as connection: + connection.execute(text("SET LOCAL ROLE workshop_booking_owner")) + for attendee_id in attendees: + connection.execute( + text("DELETE FROM workshop_booking.bookings WHERE attendee_id = :id"), + {"id": attendee_id}, + ) + for workshop_id in database.workshops: + connection.execute( + text("DELETE FROM workshop_booking.workshops WHERE id = :id"), + {"id": workshop_id}, + ) + for attendee_id in attendees: + connection.execute( + text("DELETE FROM workshop_booking.attendees WHERE id = :id"), + {"id": attendee_id}, + ) + engine.dispose() + owner_engine.dispose() + + +@pytest.fixture(scope="session") +def client(live_db: LiveDatabase): + factory = sessionmaker(live_db.engine, expire_on_commit=False) + application = create_app(settings=live_db.settings, session_factory=factory) + with TestClient(application) as test_client: + yield test_client + + +@pytest.fixture +def workshop(live_db: LiveDatabase): + def create(capacity: int = 2, *, started: bool = False) -> UUID: + workshop_id = uuid4() + starts_at = datetime.now(UTC) + (timedelta(days=-1) if started else timedelta(days=30)) + with live_db.owner_engine.begin() as connection: + connection.execute(text("SET LOCAL ROLE workshop_booking_owner")) + connection.execute( + text( + "INSERT INTO workshop_booking.workshops " + "(id, title, description, starts_at, duration_minutes, capacity) " + "VALUES (:id, :title, '', :starts_at, 60, :capacity)" + ), + { + "id": workshop_id, + "title": f"Integration workshop {workshop_id}", + "starts_at": starts_at, + "capacity": capacity, + }, + ) + live_db.workshops.append(workshop_id) + return workshop_id + + return create diff --git a/applications/workshop-booking/tests/integration/test_bookings.py b/applications/workshop-booking/tests/integration/test_bookings.py new file mode 100644 index 00000000..37f5ea39 --- /dev/null +++ b/applications/workshop-booking/tests/integration/test_bookings.py @@ -0,0 +1,237 @@ +"""Exercise real HTTP request handling and committed concurrent transactions.""" + +from concurrent.futures import ThreadPoolExecutor +from threading import Barrier, Event +from uuid import uuid4 + +import pytest +from sqlalchemy import event, text +from sqlalchemy.exc import SQLAlchemyError + +import app.main + + +def book(client, database, workshop_id, *, attendee=0, booking_id=None): + return client.post( + "/bookings", + headers=database.headers(attendee), + json={"id": str(booking_id or uuid4()), "workshop_id": str(workshop_id)}, + ) + + +def test_booking_replay_cancellation_and_new_booking(client, live_db, workshop): + workshop_id, booking_id = workshop(1), uuid4() + first = book(client, live_db, workshop_id, booking_id=booking_id) + assert first.status_code == 201 + assert first.json()["status"] == "confirmed" + replay = book(client, live_db, workshop_id, booking_id=booking_id) + assert replay.status_code == 200 + assert replay.json() == first.json() + assert live_db.active_count(workshop_id) == 1 + assert book(client, live_db, workshop_id).status_code == 409 + + receipt = client.get(f"/bookings/{booking_id}", headers=live_db.headers()) + assert receipt.status_code == 200 + assert receipt.json() == first.json() + cancelled = client.delete(f"/bookings/{booking_id}", headers=live_db.headers()) + assert cancelled.status_code == 200 + assert cancelled.json()["status"] == "cancelled" + assert ( + client.delete(f"/bookings/{booking_id}", headers=live_db.headers()).json() + == cancelled.json() + ) + assert live_db.active_count(workshop_id) == 0 + + # Replaying a cancelled UUID must never resurrect the booking. + old_replay = book(client, live_db, workshop_id, booking_id=booking_id) + assert old_replay.status_code == 200 + assert old_replay.json() == cancelled.json() + assert live_db.active_count(workshop_id) == 0 + assert book(client, live_db, workshop_id).status_code == 201 + assert book(client, live_db, workshop_id, booking_id=booking_id).json()["status"] == "cancelled" + assert live_db.active_count(workshop_id) == 1 + + +def test_authentication_ownership_and_uuid_conflicts(client, live_db, workshop): + workshop_id, other_workshop_id, booking_id = workshop(), workshop(), uuid4() + assert book(client, live_db, workshop_id, booking_id=booking_id).status_code == 201 + for method in (client.get, client.delete): + assert method(f"/bookings/{booking_id}").status_code == 401 + assert ( + method(f"/bookings/{booking_id}", headers={"Authorization": "Bearer wrong"}).status_code + == 401 + ) + assert method(f"/bookings/{booking_id}", headers=live_db.headers(1)).status_code == 404 + assert method(f"/bookings/{uuid4()}", headers=live_db.headers()).status_code == 404 + assert book(client, live_db, workshop_id, attendee=1, booking_id=booking_id).status_code == 409 + assert book(client, live_db, other_workshop_id, booking_id=booking_id).status_code == 409 + assert live_db.active_count(workshop_id) == 1 + assert live_db.active_count(other_workshop_id) == 0 + + +@pytest.mark.parametrize("attempt", range(3)) +def test_two_attendees_compete_for_the_last_seat(client, live_db, workshop, attempt): + workshop_id = workshop(1) + start = Barrier(2) + + def compete(attendee): + start.wait(timeout=10) + return book(client, live_db, workshop_id, attendee=attendee) + + with ThreadPoolExecutor(max_workers=2) as pool: + responses = list(pool.map(compete, (0, 1))) + assert sorted(response.status_code for response in responses) == [201, 409] + assert live_db.active_count(workshop_id) == 1 + + +def test_concurrent_duplicate_uuid_returns_one_receipt(client, live_db, workshop): + workshop_id, booking_id = workshop(1), uuid4() + start = Barrier(2) + + def submit(_): + start.wait(timeout=10) + return book(client, live_db, workshop_id, booking_id=booking_id) + + with ThreadPoolExecutor(max_workers=2) as pool: + responses = list(pool.map(submit, (0, 1))) + assert sorted(response.status_code for response in responses) == [200, 201] + assert responses[0].json() == responses[1].json() + assert live_db.active_count(workshop_id) == 1 + + +def test_concurrent_same_uuid_for_different_workshops_conflicts(client, live_db, workshop): + workshops, booking_id = (workshop(), workshop()), uuid4() + start = Barrier(2) + + def submit(workshop_id): + start.wait(timeout=10) + return book(client, live_db, workshop_id, booking_id=booking_id) + + with ThreadPoolExecutor(max_workers=2) as pool: + responses = list(pool.map(submit, workshops)) + assert sorted(response.status_code for response in responses) == [201, 409] + assert sum(live_db.active_count(workshop_id) for workshop_id in workshops) == 1 + receipt = client.get(f"/bookings/{booking_id}", headers=live_db.headers()) + winner = next(response for response in responses if response.status_code == 201) + assert receipt.json() == winner.json() + + +def test_concurrent_cancellations_restore_exactly_one_seat(client, live_db, workshop): + workshop_id, booking_id = workshop(1), uuid4() + assert book(client, live_db, workshop_id, booking_id=booking_id).status_code == 201 + start = Barrier(3) + + def compete(operation): + start.wait(timeout=10) + if operation == "book": + return book(client, live_db, workshop_id, attendee=1) + return client.delete(f"/bookings/{booking_id}", headers=live_db.headers()) + + with ThreadPoolExecutor(max_workers=3) as pool: + responses = list(pool.map(compete, ("cancel", "cancel", "book"))) + assert [response.status_code for response in responses[:2]] == [200, 200] + assert responses[0].json() == responses[1].json() + assert responses[2].status_code in (201, 409) + if responses[2].status_code == 409: + assert book(client, live_db, workshop_id, attendee=1).status_code == 201 + assert live_db.active_count(workshop_id) == 1 + assert book(client, live_db, workshop_id, attendee=2).status_code == 409 + + +def test_started_workshops_reject_new_bookings_and_active_cancellations(client, live_db, workshop): + expired = workshop(started=True) + assert book(client, live_db, expired).status_code == 409 + workshop_id, booking_id = workshop(), uuid4() + assert book(client, live_db, workshop_id, booking_id=booking_id).status_code == 201 + with live_db.owner_engine.begin() as connection: + connection.execute(text("SET LOCAL ROLE workshop_booking_owner")) + connection.execute( + text( + "UPDATE workshop_booking.workshops " + "SET starts_at = clock_timestamp() - interval '1 hour' " + "WHERE id = :id" + ), + {"id": workshop_id}, + ) + assert client.delete(f"/bookings/{booking_id}", headers=live_db.headers()).status_code == 409 + # A network retry is still allowed after the start time. + replay = book(client, live_db, workshop_id, booking_id=booking_id) + assert replay.status_code == 200 + assert replay.json()["status"] == "confirmed" + assert live_db.active_count(workshop_id) == 1 + + +@pytest.mark.parametrize("operation", ("book", "cancel")) +def test_start_time_is_checked_after_waiting_for_the_lock(client, live_db, workshop, operation): + workshop_id, booking_id = workshop(2), uuid4() + if operation == "cancel": + assert book(client, live_db, workshop_id, booking_id=booking_id).status_code == 201 + attempting_lock = Event() + + def observe_lock(_connection, _cursor, statement, _parameters, _context, _many): + if "workshop_booking.workshops" in statement and "FOR UPDATE" in statement: + attempting_lock.set() + + def request(): + if operation == "book": + return book(client, live_db, workshop_id, booking_id=booking_id) + return client.delete(f"/bookings/{booking_id}", headers=live_db.headers()) + + event.listen(live_db.engine, "before_cursor_execute", observe_lock) + try: + with ThreadPoolExecutor(max_workers=1) as pool: + with live_db.owner_engine.begin() as connection: + connection.execute(text("SET LOCAL ROLE workshop_booking_owner")) + connection.execute( + text( + "UPDATE workshop_booking.workshops " + "SET starts_at = clock_timestamp() + interval '1 second' WHERE id = :id" + ), + {"id": workshop_id}, + ) + pending = pool.submit(request) + assert attempting_lock.wait(timeout=5) + # Keep the parent lock until after its new start time. The API + # must check actual database time after acquiring this lock. + connection.execute(text("SELECT pg_sleep(1.1)")) + assert not pending.done() + response = pending.result(timeout=10) + assert response.status_code == 409 + assert live_db.active_count(workshop_id) == (1 if operation == "cancel" else 0) + finally: + event.remove(live_db.engine, "before_cursor_execute", observe_lock) + + +def test_public_catalog_and_input_validation(client, live_db, workshop): + workshop_id = workshop() + expired_id = workshop(started=True) + response = client.get("/workshops?limit=100") + assert response.status_code == 200 + assert str(workshop_id) in response.text + assert str(expired_id) not in response.text + for query in ("limit=0", "limit=101", "offset=-1", "offset=10001"): + assert client.get(f"/workshops?{query}").status_code == 422 + assert ( + client.post("/bookings", headers=live_db.headers(), json={"id": "bad"}).status_code == 422 + ) + assert book(client, live_db, uuid4()).status_code == 404 + + +def test_application_rolls_back_a_flushed_booking_on_database_error( + client, live_db, workshop, monkeypatch +): + workshop_id = workshop(1) + original = app.main.create_booking + + def fail_after_insert(*args, **kwargs): + original(*args, **kwargs) # Includes the real INSERT and session.flush(). + raise SQLAlchemyError("Injected error after insert") + + monkeypatch.setattr(app.main, "create_booking", fail_after_insert) + response = book(client, live_db, workshop_id) + assert response.status_code == 503 + assert response.headers["Retry-After"] == "1" + assert "Injected" not in response.text + assert live_db.active_count(workshop_id) == 0 + monkeypatch.setattr(app.main, "create_booking", original) + assert book(client, live_db, workshop_id).status_code == 201 diff --git a/applications/workshop-booking/tests/integration/test_database.py b/applications/workshop-booking/tests/integration/test_database.py new file mode 100644 index 00000000..76d9ef0c --- /dev/null +++ b/applications/workshop-booking/tests/integration/test_database.py @@ -0,0 +1,125 @@ +"""Check independent database invariants, role boundaries, and verified TLS.""" + +import socket +from dataclasses import replace +from uuid import uuid4 + +import psycopg +import pytest +from sqlalchemy import text +from sqlalchemy.exc import DBAPIError, IntegrityError, OperationalError + +from app.database import build_engine + + +def test_database_rejects_duplicate_active_booking_and_rolls_back(live_db, workshop): + workshop_id, first_id = workshop(), uuid4() + statement = text( + "INSERT INTO workshop_booking.bookings (id, workshop_id, attendee_id) " + "VALUES (:id, :workshop, :attendee)" + ) + with pytest.raises(IntegrityError) as error: + with live_db.engine.begin() as connection: + parameters = {"id": first_id, "workshop": workshop_id, "attendee": live_db.attendees[0]} + connection.execute(statement, parameters) + connection.execute(statement, {**parameters, "id": uuid4()}) + assert error.value.orig.sqlstate == "23505" + assert error.value.orig.diag.constraint_name == "bookings_one_active_per_attendee_workshop" + assert live_db.active_count(workshop_id) == 0 + + +def test_database_rejects_missing_foreign_keys(live_db, workshop): + workshop_id = workshop() + with pytest.raises(IntegrityError) as error: + with live_db.engine.begin() as connection: + connection.execute( + text( + "INSERT INTO workshop_booking.bookings (id, workshop_id, attendee_id) " + "VALUES (:id, :workshop, :attendee)" + ), + {"id": uuid4(), "workshop": workshop_id, "attendee": uuid4()}, + ) + assert error.value.orig.sqlstate == "23503" + assert live_db.active_count(workshop_id) == 0 + + +def test_capacity_check_constraint(live_db, workshop): + workshop_id = workshop() + with pytest.raises(IntegrityError) as error: + with live_db.owner_engine.begin() as connection: + connection.execute(text("SET LOCAL ROLE workshop_booking_owner")) + connection.execute( + text("UPDATE workshop_booking.workshops SET capacity = 0 WHERE id = :id"), + {"id": workshop_id}, + ) + assert error.value.orig.sqlstate == "23514" + + +@pytest.mark.parametrize( + "statement", + [ + "CREATE TABLE workshop_booking.not_allowed (id integer)", + "ALTER TABLE workshop_booking.workshops ADD COLUMN not_allowed integer", + "UPDATE workshop_booking.workshops SET capacity = capacity + 1 WHERE id = :id", + "UPDATE workshop_booking.attendees SET name = 'not allowed' WHERE id = :id", + "UPDATE workshop_booking.bookings SET attendee_id = :id WHERE id = :id", + "DELETE FROM workshop_booking.bookings WHERE id = :id", + "SELECT * FROM workshop_booking.alembic_version", + "SET ROLE workshop_booking_owner", + ], +) +def test_runtime_role_cannot_mutate_schema_or_identity(live_db, workshop, statement): + workshop_id = workshop() + with pytest.raises(DBAPIError) as error: + with live_db.engine.begin() as connection: + connection.execute(text(statement), {"id": workshop_id}) + assert error.value.orig.sqlstate == "42501" + + +def test_tls_is_in_use_and_runtime_role_is_unprivileged(live_db): + with live_db.engine.connect() as connection: + assert ( + connection.scalar(text("SELECT ssl FROM pg_stat_ssl WHERE pid = pg_backend_pid()")) + is True + ) + row = connection.execute( + text( + "SELECT rolsuper, rolcreatedb, rolcreaterole, rolreplication, rolbypassrls " + "FROM pg_roles WHERE rolname = current_user" + ) + ).one() + assert not any(row) + assert not connection.scalar( + text("SELECT has_schema_privilege(current_user, 'workshop_booking', 'CREATE')") + ) + + +def test_invalid_ca_is_rejected(live_db, tmp_path): + invalid_ca = tmp_path / "invalid-ca.pem" + invalid_ca.write_text("This is deliberately not a certificate.\n") + engine = build_engine(replace(live_db.settings, sslrootcert=invalid_ca)) + try: + with pytest.raises(OperationalError): + with engine.connect(): + pytest.fail("A connection must not accept an invalid CA file.") + finally: + engine.dispose() + + +def test_certificate_hostname_is_verified(live_db): + settings = live_db.settings + address = socket.getaddrinfo(settings.host, settings.port, type=socket.SOCK_STREAM)[0][4][0] + # Connect to the real endpoint's address while asking libpq to verify a + # different hostname. No password can be sent before TLS verification. + with pytest.raises(psycopg.OperationalError, match="certificate"): + psycopg.connect( + host="deliberately-wrong-hostname.invalid", + hostaddr=address, + port=settings.port, + dbname=settings.database, + user=settings.user, + password=settings.password, + sslmode="verify-full", + sslrootcert=str(settings.sslrootcert), + connect_timeout=5, + ) diff --git a/applications/workshop-booking/tests/migrations/test_lifecycle.py b/applications/workshop-booking/tests/migrations/test_lifecycle.py new file mode 100644 index 00000000..d6929734 --- /dev/null +++ b/applications/workshop-booking/tests/migrations/test_lifecycle.py @@ -0,0 +1,75 @@ +"""Destructive migration test: run explicitly on an empty example schema.""" + +import os +from dataclasses import replace +from pathlib import Path + +import pytest +from alembic import command +from alembic.config import Config +from sqlalchemy import text + +from app.config import Settings +from app.database import build_engine + + +def test_upgrade_downgrade_reapply(monkeypatch): + if os.environ.get("WORKSHOP_ALLOW_SCHEMA_RESET") != "1": + pytest.fail( + "This test drops the example tables. Use an empty, dedicated test service " + "and explicitly set WORKSHOP_ALLOW_SCHEMA_RESET=1." + ) + password = os.environ.get("WORKSHOP_MIGRATOR_PASSWORD") + if not password: + pytest.fail("Set WORKSHOP_MIGRATOR_PASSWORD to run migration tests.") + settings = replace( + Settings.from_env(require_tokens=False), + user="workshop_booking_migrator", + password=password, + ) + monkeypatch.setenv("PGUSER", settings.user) + monkeypatch.setenv("PGPASSWORD", settings.password) + engine = build_engine(settings) + config = Config(str(Path(__file__).resolve().parents[2] / "alembic.ini")) + + def table_names(): + with engine.begin() as connection: + connection.execute(text("SET LOCAL ROLE workshop_booking_owner")) + return set( + connection.scalars( + text("SELECT tablename FROM pg_tables WHERE schemaname = 'workshop_booking'") + ) + ) + + # A deliberate second guard: even the opt-in cannot destroy populated data. + for table in table_names() - {"alembic_version"}: + if table not in {"workshops", "attendees", "bookings"}: + pytest.fail("The schema contains an unexpected table; refusing to reset it.") + with engine.begin() as connection: + connection.execute(text("SET LOCAL ROLE workshop_booking_owner")) + if connection.scalar(text(f"SELECT count(*) FROM workshop_booking.{table}")): + pytest.fail("The example schema contains data; refusing to reset it.") + + try: + command.upgrade(config, "head") + assert table_names() == {"workshops", "attendees", "bookings", "alembic_version"} + with engine.begin() as connection: + connection.execute(text("SET LOCAL ROLE workshop_booking_owner")) + owners = set( + connection.scalars( + text("SELECT tableowner FROM pg_tables WHERE schemaname = 'workshop_booking'") + ) + ) + assert owners == {"workshop_booking_owner"} + assert ( + connection.scalar(text("SELECT version_num FROM workshop_booking.alembic_version")) + == "001_initial" + ) + + command.downgrade(config, "base") + assert table_names() == {"alembic_version"} + command.upgrade(config, "head") + assert table_names() == {"workshops", "attendees", "bookings", "alembic_version"} + command.upgrade(config, "head") # Already at head is a harmless no-op. + finally: + engine.dispose() diff --git a/applications/workshop-booking/tests/unit/test_api.py b/applications/workshop-booking/tests/unit/test_api.py new file mode 100644 index 00000000..12bef39b --- /dev/null +++ b/applications/workshop-booking/tests/unit/test_api.py @@ -0,0 +1,117 @@ +from pathlib import Path +from typing import Annotated +from uuid import UUID + +import pytest +from fastapi import Depends, FastAPI +from fastapi.testclient import TestClient +from pydantic import ValidationError +from sqlalchemy.exc import OperationalError + +from app.auth import get_attendee_id +from app.config import Settings +from app.database import get_session +from app.main import create_app +from app.schemas import BookingCreate + +ALICE = UUID("11111111-1111-4111-8111-111111111111") +TOKEN = "a" * 43 +WORKSHOP = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" +BOOKING = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" + + +@pytest.fixture +def settings(): + return Settings("example.com", 5432, "db", "user", "secret", Path("ca.pem"), {ALICE: TOKEN}) + + +@pytest.fixture +def auth_client(settings): + app = FastAPI() + app.state.settings = settings + + @app.get("/identity") + def identity(attendee_id: Annotated[UUID, Depends(get_attendee_id)]): + return {"id": str(attendee_id)} + + return TestClient(app) + + +@pytest.mark.parametrize( + "headers", + [{}, {"Authorization": "Basic abc"}, {"Authorization": "Bearer wrong"}], +) +def test_auth_rejects_missing_or_wrong_bearer(auth_client, headers): + response = auth_client.get("/identity", headers=headers) + assert response.status_code == 401 + assert response.headers["www-authenticate"] == "Bearer" + + +def test_auth_resolves_the_server_owned_identity(auth_client): + response = auth_client.get("/identity", headers={"Authorization": f"Bearer {TOKEN}"}) + assert response.status_code == 200 + assert response.json() == {"id": str(ALICE)} + + +@pytest.mark.parametrize( + "body", + [ + {"id": BOOKING, "workshop_id": WORKSHOP, "attendee_id": str(ALICE)}, + {"id": BOOKING, "workshop_id": WORKSHOP, "seats": 2}, + {"id": BOOKING, "workshop_id": WORKSHOP, "status": "confirmed"}, + {"id": "invalid", "workshop_id": WORKSHOP}, + {"id": BOOKING, "workshop_id": 123}, + {"workshop_id": WORKSHOP}, + ], +) +def test_booking_payload_rejects_untrusted_owner_or_invalid_values(body): + with pytest.raises(ValidationError): + BookingCreate.model_validate(body) + + +def test_booking_payload_accepts_client_uuid(): + body = BookingCreate.model_validate({"id": BOOKING, "workshop_id": WORKSHOP}) + assert body.id == UUID(BOOKING) + + +@pytest.fixture +def api_client(settings): + app = create_app() + app.state.settings = settings + + class UnavailableSession: + def begin(self): + raise OperationalError("SELECT private-data", {}, Exception("password-secret")) + + app.dependency_overrides[get_session] = lambda: UnavailableSession() + return TestClient(app) + + +@pytest.mark.parametrize("query", ["limit=0", "limit=101", "offset=-1", "offset=10001"]) +def test_list_pagination_is_bounded(api_client, query): + assert api_client.get(f"/workshops?{query}").status_code == 422 + + +def test_database_error_is_redacted_and_retryable(api_client, caplog): + response = api_client.get("/workshops") + assert response.status_code == 503 + assert response.headers["retry-after"] == "1" + assert "private-data" not in response.text + caplog.text + assert "password-secret" not in response.text + caplog.text + + +def test_validation_does_not_echo_submitted_values(api_client): + response = api_client.post( + "/bookings", + headers={"Authorization": f"Bearer {TOKEN}"}, + json={"id": "secret-input", "workshop_id": WORKSHOP}, + ) + assert response.status_code == 422 + assert "secret-input" not in response.text + + +def test_openapi_exposes_bearer_auth_only_on_booking_routes(api_client): + schema = api_client.get("/openapi.json").json() + assert "security" not in schema["paths"]["/workshops"]["get"] + assert schema["paths"]["/bookings"]["post"]["security"] == [{"HTTPBearer": []}] + assert set(schema["paths"]) == {"/workshops", "/bookings", "/bookings/{booking_id}"} diff --git a/applications/workshop-booking/tests/unit/test_body_limit.py b/applications/workshop-booking/tests/unit/test_body_limit.py new file mode 100644 index 00000000..e420b045 --- /dev/null +++ b/applications/workshop-booking/tests/unit/test_body_limit.py @@ -0,0 +1,40 @@ +import asyncio + +from app.middleware import BookingBodyLimit + + +def run_request(chunks): + output = [] + delivered = [] + messages = iter( + {"type": "http.request", "body": body, "more_body": i < len(chunks) - 1} + for i, body in enumerate(chunks) + ) + + async def receive(): + return next(messages) + + async def send(message): + output.append(message) + + async def app(scope, receive, send): + delivered.append(await receive()) + + asyncio.run( + BookingBodyLimit(app)( + {"type": "http", "method": "POST", "path": "/bookings"}, receive, send + ) + ) + return output, delivered + + +def test_rejects_streamed_body_without_content_length(): + output, delivered = run_request([b"x" * 2048, b"x" * 2049]) + assert output[0]["status"] == 413 + assert delivered == [] + + +def test_boundary_body_is_delivered_to_parser_once(): + output, delivered = run_request([b"x" * 2048, b"x" * 2048]) + assert output == [] + assert delivered == [{"type": "http.request", "body": b"x" * 4096, "more_body": False}] diff --git a/applications/workshop-booking/tests/unit/test_config.py b/applications/workshop-booking/tests/unit/test_config.py new file mode 100644 index 00000000..998719c7 --- /dev/null +++ b/applications/workshop-booking/tests/unit/test_config.py @@ -0,0 +1,107 @@ +import json +from pathlib import Path +from uuid import UUID + +import pytest + +from app.config import ConfigurationError, Settings, parse_tokens +from app.database import build_engine + +ALICE = "11111111-1111-4111-8111-111111111111" +BOB = "22222222-2222-4222-8222-222222222222" +TOKEN = "a" * 43 + + +@pytest.fixture +def configured_env(monkeypatch, tmp_path): + monkeypatch.chdir(tmp_path) + cert = tmp_path / "ca.pem" + cert.write_text("test certificate") + values = { + "PGHOST": "test.postgres.example.com", + "PGPORT": "5432", + "PGDATABASE": "workshop_booking", + "PGUSER": "workshop_app", + "PGPASSWORD": "a password containing @:/?&#%", + "PGSSLROOTCERT": str(cert), + "API_BEARER_TOKENS": json.dumps({ALICE: TOKEN}), + } + for name, value in values.items(): + monkeypatch.setenv(name, value) + return values + + +def test_credentials_are_not_in_settings_repr(configured_env): + settings = Settings.from_env() + assert settings.bearer_tokens == {UUID(ALICE): TOKEN} + assert configured_env["PGPASSWORD"] not in repr(settings) + assert TOKEN not in repr(settings) + + +@pytest.mark.parametrize("port", ["0", "65536", "postgres", ""]) +def test_port_is_bounded(configured_env, monkeypatch, port): + monkeypatch.setenv("PGPORT", port) + with pytest.raises(ConfigurationError, match="PGPORT"): + Settings.from_env() + + +@pytest.mark.parametrize("host", ["/tmp", "one,two", "host user=admin", ""]) +def test_host_is_a_single_dns_name(configured_env, monkeypatch, host): + monkeypatch.setenv("PGHOST", host) + with pytest.raises(ConfigurationError, match="PGHOST"): + Settings.from_env() + + +def test_requires_a_ca_file(configured_env, monkeypatch): + monkeypatch.setenv("PGSSLROOTCERT", "/does/not/exist.pem") + with pytest.raises(ConfigurationError, match="PGSSLROOTCERT"): + Settings.from_env() + + +def test_migrations_do_not_require_api_credentials(configured_env, monkeypatch): + monkeypatch.delenv("API_BEARER_TOKENS") + assert Settings.from_env(require_tokens=False).bearer_tokens == {} + with pytest.raises(ConfigurationError, match="API_BEARER_TOKENS"): + Settings.from_env() + + +@pytest.mark.parametrize( + "raw", + [ + "not-json", + "[]", + "{}", + json.dumps({"invalid-uuid": TOKEN}), + json.dumps({ALICE: "too-short"}), + json.dumps({ALICE: TOKEN, BOB: TOKEN}), + json.dumps({ALICE: " " * 43}), + json.dumps({ALICE: "x" * 257}), + json.dumps({ALICE: 1234}), + '{"' + ALICE + '":"' + TOKEN + '","' + ALICE + '":"' + "b" * 43 + '"}', + json.dumps({ALICE: TOKEN, ALICE.replace("-", ""): "b" * 43}), + ], +) +def test_rejects_ambiguous_or_weak_token_configuration_without_echoing_it(raw): + with pytest.raises(ConfigurationError) as error: + parse_tokens(raw) + assert raw not in str(error.value) + assert TOKEN not in str(error.value) + + +def test_engine_forces_tls_and_read_committed(configured_env, monkeypatch): + captured = {} + + def capture_engine(url, **options): + captured.update(url=url, **options) + return "engine" + + monkeypatch.setattr("app.database.create_engine", capture_engine) + settings = Settings.from_env() + assert build_engine(settings) == "engine" + assert captured["url"].password == configured_env["PGPASSWORD"] + assert captured["connect_args"]["sslmode"] == "verify-full" + assert Path(captured["connect_args"]["sslrootcert"]).is_file() + assert captured["isolation_level"] == "READ COMMITTED" + assert captured["hide_parameters"] is True + assert captured["pool_size"] + captured["max_overflow"] == 10 + assert "lock_timeout=3000" in captured["connect_args"]["options"]