Add FastAPI todobackend.com app example.

Author: dom96

fastapi-todo/README.md

@@ -0,0 +1,37 @@
+# FastAPI Todo Backend
+
+A Python FastAPI implementation of the [Todo-Backend](https://todobackend.com) spec, running on Cloudflare Workers with D1 for storage.
+
+## Development
+
+Initialize the local D1 database and start the dev server:
+
+```sh
+uv run pywrangler d1 execute todos --local --file db_init.sql
+uv run pywrangler dev
+```
+
+## Testing with the Todo-Backend spec runner
+
+Start the dev server, then open the spec runner pointing at your local instance:
+
+```
+https://todobackend.com/specs/index.html?http://localhost:8787/todos
+```
+
+You can also use the Todo-Backend client app:
+
+```
+https://todobackend.com/client/index.html?http://localhost:8787/todos
+```
+
+## API
+
+| Method   | Path             | Description        |
+| -------- | ---------------- | ------------------ |
+| `GET`    | `/todos`         | List all todos     |
+| `POST`   | `/todos`         | Create a todo      |
+| `DELETE`  | `/todos`         | Delete all todos   |
+| `GET`    | `/todos/{id}`    | Get a single todo  |
+| `PATCH`  | `/todos/{id}`    | Update a todo      |
+| `DELETE`  | `/todos/{id}`    | Delete a todo      |

fastapi-todo/db_init.sql

@@ -0,0 +1,6 @@
+CREATE TABLE IF NOT EXISTS todos (
+    id TEXT PRIMARY KEY,
+    title TEXT NOT NULL DEFAULT '',
+    completed INTEGER NOT NULL DEFAULT 0,
+    "order" INTEGER
+);

fastapi-todo/package.json

@@ -0,0 +1,13 @@
+{
+	"name": "fastapi-todo",
+	"version": "0.0.0",
+	"private": true,
+	"scripts": {
+		"deploy": "uv run pywrangler deploy",
+		"dev": "uv run pywrangler dev",
+		"start": "uv run pywrangler dev"
+	},
+	"devDependencies": {
+		"wrangler": "^4.114.0"
+	}
+}

fastapi-todo/pyproject.toml

@@ -0,0 +1,15 @@
+[project]
+name = "fastapi-todo"
+version = "0.1.0"
+description = "FastAPI todo backend conforming to the todobackend.com spec"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "fastapi",
+]
+
+[dependency-groups]
+dev = [
+    "workers-py",
+    "workers-runtime-sdk"
+]

fastapi-todo/src/worker.py

@@ -0,0 +1,132 @@
+import uuid
+
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
+from workers import WorkerEntrypoint
+
+app = FastAPI()
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+    expose_headers=["*"],
+)
+
+
+def _base_url(request: Request) -> str:
+    """Return the root URL for the todos collection."""
+    return str(request.base_url).rstrip("/") + "/todos"
+
+
+def _row_to_todo(row, request: Request) -> dict:
+    """Convert a D1 row into a todo dict with the absolute ``url`` field."""
+    return {
+        "id": row.id,
+        "title": row.title,
+        "completed": bool(row.completed),
+        "order": row.order,
+        "url": f"{_base_url(request)}/{row.id}",
+    }
+
+
+def _db(request: Request):
+    """Get the D1 database binding from the ASGI scope."""
+    return request.scope["env"].DB
+
+
+@app.get("/todos")
+async def list_todos(request: Request):
+    results = await _db(request).prepare("SELECT * FROM todos").all()
+    return [_row_to_todo(r, request) for r in results.results]
+
+
+@app.post("/todos")
+async def create_todo(request: Request):
+    body = await request.json()
+    todo_id = str(uuid.uuid4())
+    title = body.get("title", "")
+    completed = 1 if body.get("completed", False) else 0
+    order = body.get("order")
+
+    await (
+        _db(request)
+        .prepare(
+            'INSERT INTO todos (id, title, completed, "order") VALUES (?, ?, ?, ?)'
+        )
+        .bind(todo_id, title, completed, order)
+        .run()
+    )
+
+    row = (
+        await _db(request)
+        .prepare("SELECT * FROM todos WHERE id = ?")
+        .bind(todo_id)
+        .first()
+    )
+
+    return _row_to_todo(row, request)
+
+
+@app.delete("/todos")
+async def delete_all_todos(request: Request):
+    await _db(request).prepare("DELETE FROM todos").run()
+    return []
+
+
+@app.get("/todos/{todo_id}")
+async def get_todo(todo_id: str, request: Request):
+    row = (
+        await _db(request)
+        .prepare("SELECT * FROM todos WHERE id = ?")
+        .bind(todo_id)
+        .first()
+    )
+    if row is None:
+        return {"error": "not found"}
+    return _row_to_todo(row, request)
+
+
+@app.patch("/todos/{todo_id}")
+async def update_todo(todo_id: str, request: Request):
+    body = await request.json()
+    sets = []
+    values = []
+    if "title" in body:
+        sets.append("title = ?")
+        values.append(body["title"])
+    if "completed" in body:
+        sets.append("completed = ?")
+        values.append(1 if body["completed"] else 0)
+    if "order" in body:
+        sets.append('"order" = ?')
+        values.append(body["order"])
+
+    if sets:
+        values.append(todo_id)
+        await (
+            _db(request)
+            .prepare(f"UPDATE todos SET {', '.join(sets)} WHERE id = ?")
+            .bind(*values)
+            .run()
+        )
+
+    row = (
+        await _db(request)
+        .prepare("SELECT * FROM todos WHERE id = ?")
+        .bind(todo_id)
+        .first()
+    )
+    if row is None:
+        return {"error": "not found"}
+    return _row_to_todo(row, request)
+
+
+@app.delete("/todos/{todo_id}")
+async def delete_todo(todo_id: str, request: Request):
+    await _db(request).prepare("DELETE FROM todos WHERE id = ?").bind(todo_id).run()
+    return []
+
+import asgi
+Default = asgi.entrypoint(app)

fastapi-todo/wrangler.jsonc

@@ -0,0 +1,19 @@
+{
+	"$schema": "node_modules/wrangler/config-schema.json",
+	"name": "fastapi-todo",
+	"main": "src/worker.py",
+	"compatibility_date": "2026-08-01",
+	"compatibility_flags": [
+		"python_workers",
+	],
+	"d1_databases": [
+		{
+			"binding": "DB",
+			"database_name": "todos",
+			"database_id": "00000000-0000-0000-0000-000000000000"
+		}
+	],
+	"observability": {
+		"enabled": true
+	}
+}

tests/test_examples.py

@@ -173,6 +173,87 @@ def test_10_workflows(dev_server):
     assert isinstance(status, dict)
 
 
+@pytest.fixture
+def init_fastapi_todo_db():
+    subprocess.run(
+        [
+            "uv",
+            "run",
+            "pywrangler",
+            "d1",
+            "execute",
+            "todos",
+            "--local",
+            "--file",
+            "db_init.sql",
+        ],
+        cwd=REPO_ROOT / "fastapi-todo",
+        check=True,
+    )
+
+
+def test_fastapi_todo(init_fastapi_todo_db, dev_server):
+    port = dev_server
+    base = f"http://localhost:{port}/todos"
+
+    # DELETE all todos
+    response = requests.delete(base)
+    assert response.status_code == 200
+
+    # GET should return empty list
+    response = requests.get(base)
+    assert response.status_code == 200
+    assert response.json() == []
+
+    # POST a new todo
+    response = requests.post(base, json={"title": "walk the dog"})
+    assert response.status_code == 200
+    todo = response.json()
+    assert todo["title"] == "walk the dog"
+    assert todo["completed"] is False
+    assert "url" in todo
+    todo_url = todo["url"]
+
+    # GET the individual todo by its url
+    response = requests.get(todo_url)
+    assert response.status_code == 200
+    assert response.json()["title"] == "walk the dog"
+
+    # PATCH the todo
+    response = requests.patch(
+        todo_url, json={"title": "bathe the cat", "completed": True}
+    )
+    assert response.status_code == 200
+    patched = response.json()
+    assert patched["title"] == "bathe the cat"
+    assert patched["completed"] is True
+
+    # POST a todo with an order field
+    response = requests.post(base, json={"title": "ordered todo", "order": 42})
+    assert response.status_code == 200
+    assert response.json()["order"] == 42
+
+    # GET all todos should return 2
+    response = requests.get(base)
+    assert response.status_code == 200
+    assert len(response.json()) == 2
+
+    # DELETE individual todo
+    response = requests.delete(todo_url)
+    assert response.status_code == 200
+
+    # GET all todos should return 1
+    response = requests.get(base)
+    assert response.status_code == 200
+    assert len(response.json()) == 1
+
+    # DELETE all
+    response = requests.delete(base)
+    assert response.status_code == 200
+    response = requests.get(base)
+    assert response.json() == []
+
+
 def test_18_django(dev_server):
     port = dev_server
     response = requests.get(f"http://localhost:{port}")