Add flask todo parallel to fastapi-todo

README.md

@@ -32,9 +32,12 @@ Need to deploy your Worker to Cloudflare? Python Workers are in open beta and ha
 - [**`chatroom/`**](chatroom) - A real-time chatroom using WebSocket.
 - [**`sync-http-clients/`**](sync-http-clients) — demonstrates outbound HTTP with synchronous Python clients (`requests`, `urllib3`, and `httpx.Client`).
 - [**`dynamic-py-py/`**](dynamic-py-py) — shows how to load and run a Python Worker dynamically at runtime using a [Worker Loader](https://developers.cloudflare.com/workers/runtime-apis/bindings/worker-loader/) binding.
-- [**`django/`**](django) — runs a naive Django WSGI application directly on Python Workers.
-- [**`django-todo-d1/`**](django-todo-d1) — implements the Todo-Backend API with Django and D1.
 - [**`image-redraw/`**](image-redraw) — an example that combines [FastAPI](https://fastapi.tiangolo.com/), [R2](https://developers.cloudflare.com/r2/), [Queues](https://developers.cloudflare.com/queues/), [Workflows](https://developers.cloudflare.com/workflows/) and [Workers AI](https://developers.cloudflare.com/workers-ai/) to redraw uploaded images.
+- [**`django/`**](django) — runs a naive Django WSGI application directly on Python Workers.
+- [**`django-todo-d1/`**](django-todo-d1) — uses Django with D1 for a basic TODO application.
+- [**`fastapi-todo/`**](fastapi-todo) — implements the [Todo-Backend](https://todobackend.com) spec with FastAPI (ASGI) and D1.
+- [**`flask-todo/`**](flask-todo) — implements the same [Todo-Backend](https://todobackend.com) API with [Flask](https://flask.palletsprojects.com/) (WSGI) and D1.
+
 
 
 ## Open Beta and Limits

flask-todo/README.md

@@ -0,0 +1,37 @@
+# Flask Todo Backend
+
+A Python Flask 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      |

flask-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
+);

flask-todo/package.json

@@ -0,0 +1,13 @@
+{
+	"name": "flask-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"
+	}
+}

flask-todo/pyproject.toml

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

flask-todo/src/worker.py

@@ -0,0 +1,128 @@
+import uuid
+
+from flask import Flask, jsonify, request
+from flask_cors import CORS
+from pyodide.ffi import run_sync
+from werkzeug.exceptions import HTTPException
+from workers import wsgi
+
+app = Flask(__name__)
+# Preserve the field order used below rather than sorting keys alphabetically.
+app.json.sort_keys = False
+
+# The todobackend.com spec runner calls this Worker from another origin.
+CORS(app, origins="*", send_wildcard=True, allow_headers="*", expose_headers="*")
+
+
+@app.errorhandler(HTTPException)
+def _json_error(err: HTTPException):
+    """Render Werkzeug's HTTP errors as JSON instead of HTML."""
+    return jsonify({"detail": err.name}), err.code
+
+
+def _base_url() -> str:
+    """Return the root URL for the todos collection."""
+    return request.url_root.rstrip("/") + "/todos"
+
+
+def _row_to_todo(row) -> 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()}/{row.id}",
+    }
+
+
+def _db():
+    """Get the D1 database binding from the WSGI environ."""
+    return request.environ["workers.env"].DB
+
+
+@app.get("/todos")
+def list_todos():
+    results = run_sync(_db().prepare("SELECT * FROM todos").all())
+    return jsonify([_row_to_todo(r) for r in results.results])
+
+
+@app.post("/todos")
+def create_todo():
+    body = request.get_json(silent=True) or {}
+    todo_id = str(uuid.uuid4())
+    title = body.get("title", "")
+    completed = 1 if body.get("completed", False) else 0
+    order = body.get("order")
+
+    run_sync(
+        _db()
+        .prepare(
+            'INSERT INTO todos (id, title, completed, "order") VALUES (?, ?, ?, ?)'
+        )
+        .bind(todo_id, title, completed, order)
+        .run()
+    )
+
+    row = run_sync(
+        _db().prepare("SELECT * FROM todos WHERE id = ?").bind(todo_id).first()
+    )
+
+    return jsonify(_row_to_todo(row))
+
+
+@app.delete("/todos")
+def delete_all_todos():
+    run_sync(_db().prepare("DELETE FROM todos").run())
+    return jsonify([])
+
+
+@app.get("/todos/<todo_id>")
+def get_todo(todo_id: str):
+    row = run_sync(
+        _db().prepare("SELECT * FROM todos WHERE id = ?").bind(todo_id).first()
+    )
+    if row is None:
+        return jsonify({"error": "not found"})
+    return jsonify(_row_to_todo(row))
+
+
+@app.patch("/todos/<todo_id>")
+def update_todo(todo_id: str):
+    body = request.get_json(silent=True) or {}
+    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)
+        run_sync(
+            _db()
+            .prepare(f"UPDATE todos SET {', '.join(sets)} WHERE id = ?")
+            .bind(*values)
+            .run()
+        )
+
+    row = run_sync(
+        _db().prepare("SELECT * FROM todos WHERE id = ?").bind(todo_id).first()
+    )
+    if row is None:
+        return jsonify({"error": "not found"})
+    return jsonify(_row_to_todo(row))
+
+
+@app.delete("/todos/<todo_id>")
+def delete_todo(todo_id: str):
+    run_sync(_db().prepare("DELETE FROM todos WHERE id = ?").bind(todo_id).run())
+    return jsonify([])
+
+
+Default = wsgi.entrypoint(app)

flask-todo/wrangler.jsonc

@@ -0,0 +1,19 @@
+{
+	"$schema": "node_modules/wrangler/config-schema.json",
+	"name": "flask-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

@@ -256,6 +256,87 @@ def test_fastapi_todo(init_fastapi_todo_db, dev_server):
     assert_todo_backend(dev_server)
 
 
+@pytest.fixture
+def init_flask_todo_db():
+    subprocess.run(
+        [
+            "uv",
+            "run",
+            "pywrangler",
+            "d1",
+            "execute",
+            "todos",
+            "--local",
+            "--file",
+            "db_init.sql",
+        ],
+        cwd=REPO_ROOT / "flask-todo",
+        check=True,
+    )
+
+
+def test_flask_todo(init_flask_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_django(dev_server):
     port = dev_server
     response = requests.get(f"http://localhost:{port}")