patx/wred
introducing wred, a stupidly simple asgi framework. yeah its not as good as micropie ;)
Commit d8fe826 · patx · 2026-10-13T01:37:49-04:00
Comments
No comments yet.
Diff
diff --git a/.coveragerc b/.coveragerc
new file mode 100644
index 0000000..da34f9d
--- /dev/null
+++ b/.coveragerc
@@ -0,0 +1,8 @@
+[run]
+branch = True
+source =
+ wred
+
+[report]
+fail_under = 100
+show_missing = True
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..536cf1b
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,25 @@
+# Python caches
+__pycache__/
+*.py[cod]
+
+# Virtual environments
+.venv/
+venv/
+
+# Build artifacts
+build/
+dist/
+*.egg-info/
+
+# Test and coverage output
+.pytest_cache/
+.coverage
+.coverage.*
+htmlcov/
+coverage.xml
+
+# Local secrets
+.env
+.env.*
+!.env.example
+.aws/
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..ba17aac
--- /dev/null
+++ b/README.md
@@ -0,0 +1,185 @@
+# Wred
+
+A small ASGI framework for JSON APIs. One file. JSON from `orjson`, server from `granian`.
+
+Your API needs to return some JSON. It probably doesn't need a goddamn cathedral.
+
+Oh, and it's pronounced "red", like the color.
+
+## What you get
+
+- Async handlers and a route decorator.
+- Request headers, cached body reads, and JSON parsing.
+- JSON responses (dicts and lists) & static file serving.
+- Response tuples: `(body, content_type, status[, headers])`.
+
+## What you don't get
+
+Auth, sessions, databases, validation schemas, templates, CORS, CSRF protection,
+rate limiting, dependency injection, generated docs, background jobs, WebSockets,
+streaming, or a reason to argue about folder structure.
+
+Those are your problem. You can solve them without making them everyone else's.
+
+## Use it
+
+Python 3.11+. Clone the repo, then install it:
+
+Procure your goods, yeah, use `uv` not `pip`, its 2026:
+
+```bash
+$ git clone https://gitman.io/git/patx/wred
+$ cd wred
+$ uv sync
+```
+
+Put this in your pipe (`app.py`):
+
+```python
+from wred import App
+
+app = App()
+
[email protected]('/')
+async def say_hello(request):
+ return {'hello': 'world'}
+```
+
+And smoke it:
+
+```bash
+uv run granian --interface asginl app:app
+```
+
+`asginl` disables lifespan handling. Wred doesn't implement it.
+
+## Routes
+
+Handlers are async. The first argument is the request. `methods` defaults to
+`['GET']`; path parameters arrive as strings, passed by keyword.
+
+```python
[email protected]('/users/{user_id}')
+async def get_user(request, user_id):
+ return {'id': user_id}
+
[email protected]('/users', methods=['POST'])
+async def create_user(request):
+ return await request.json(), 'application/json', 201
+
[email protected]('/users/{user_id}', methods=['PUT', 'PATCH'])
+async def update_user(request, user_id):
+ return {'id': user_id, 'method': request.method}
+```
+
+Different handlers can share a path with different methods. Fixed routes win
+over dynamic routes for the requested method; dynamic routes are checked in
+registration order. Registering the same path and method again replaces its handler.
+Unmatched paths or methods return JSON 404. No automatic HEAD, OPTIONS, or 405.
+
+## Requests
+
+| API | What you get |
+| --- | --- |
+| `request.path` | URL path, without the query string. |
+| `request.method` | HTTP method, such as `GET` or `POST`. |
+| `request.headers` | Dictionary of decoded headers; use lowercase names. |
+| `request.scope` | Raw ASGI scope: query string, client address, scheme, etc. |
+| `await request.body()` | Raw bytes, cached after the first read. |
+| `await request.json()` | Parsed JSON; an empty body becomes `{}`. |
+
+JSON parsing isn't validation. Arrays and scalars parse too. Check your own inputs.
+
+```python
+from urllib.parse import parse_qs
+from wred import WredRequestError
+
[email protected]('/echo', methods=['POST'])
+async def echo(request):
+ data = await request.json()
+ if not isinstance(data, dict):
+ raise WredRequestError('Send a JSON object')
+ return {'received': data}
+
[email protected]('/search')
+async def search(request):
+ query = parse_qs(request.scope.get('query_string', b'').decode('utf-8'))
+ return {'q': query.get('q', [''])[0]}
+```
+
+Read a bearer credential with `request.headers.get('authorization', '')`.
+Checking that credential is your job.
+
+## Responses
+
+Return a dict or list for JSON with HTTP 200. Return bytes for
+`application/octet-stream`. No `json()` helper. No ceremony.
+
+For a custom response, return `(body, content_type, status)` or
+`(body, content_type, status, headers)`. Headers are a list of string pairs,
+so repeated headers such as `Set-Cookie` work too.
+
+```python
[email protected]('/items')
+async def items(request):
+ return [{'id': 1}, {'id': 2}]
+
[email protected]('/items', methods=['POST'])
+async def create_item(request):
+ return {'id': 3}, 'application/json', 201, [('Location', '/items/3')]
+
[email protected]('/old')
+async def redirect(request):
+ return b'', 'text/plain', 303, [('Location', '/')]
+
[email protected]('/items/{item_id}', methods=['DELETE'])
+async def delete_item(request, item_id):
+ return b'', 'application/json', 204
+```
+
+Dicts and lists always get `application/json`, even inside a tuple.
+Other bodies must be bytes: encode text yourself, for example
+`return 'hello'.encode(), 'text/plain; charset=utf-8', 200`.
+Custom headers can carry cookies, cache policies, CORS rules, or `Retry-After`;
+Wred sends them, you decide what they mean.
+
+## Static files
+
+`await app.static(filename, root=directory)` returns `(bytes, content_type, status)`.
+The root is required and accepts a string or `Path`.
+
+```python
+from pathlib import Path
+
+static_root = Path(__file__).parent / 'static'
+
[email protected]('/')
+async def index(request):
+ return await app.static('index.html', root=static_root)
+
[email protected]('/static/{filename}')
+async def asset(request, filename):
+ return await app.static(filename, root=static_root)
+```
+
+The helper guesses the MIME type and rejects files outside the root, including
+symlink escapes. Missing or unreadable files return a plain-text 404. Files are
+read in a worker thread, entirely into memory. Small files. Don't feed it a movie.
+
+## Errors
+
+Malformed JSON and `WredRequestError` become HTTP 400 with `{'error': message}`.
+Unsupported handler response types become JSON 500 errors. Other exceptions
+aren't caught by Wred; fix your shit or handle them in your application.
+
+## Example app
+
+The [`example_app`](example_app/app.py) notebook demonstrates signup, login,
+refresh cookies, logout, protected endpoints, note CRUD, and rate limiting.
+Those features belong to the example, not the framework.
+
+```bash
+uv run granian --interface asginl --workers 1 example_app.app:app
+```
+
+Open <http://127.0.0.1:8000>. Demo data lives in memory and disappears on restart.
diff --git a/docs/index.html b/docs/index.html
new file mode 100644
index 0000000..bfa604a
--- /dev/null
+++ b/docs/index.html
@@ -0,0 +1,203 @@
+<!doctype html>
+<html lang="en">
+<head>
+ <meta charset="utf-8">
+ <meta name="viewport" content="width=device-width, initial-scale=1">
+ <meta name="description" content="Wred is a tiny Python/ASGI backend for HTML and JavaScript apps. Roll a route. Pass some JSON. Ship that shit.">
+ <title>Wred — a tiny Python/ASGI backend</title>
+ <style>
+ :root { color-scheme: light; font-family: Arial, Helvetica, sans-serif; color: #702a36; background: #fff8f8; }
+ * { box-sizing: border-box; }
+ body { margin: 0; }
+ main { width: 100%; }
+ .intro { text-align: center; padding: clamp(12px, 4dvh, 40px) 32px clamp(24px, 6dvh, 64px); }
+ h1 { margin: 0; }
+ h1 img { display: block; width: min(680px, 100%, 70dvh); height: auto; margin: auto; }
+ .intro p { margin: 16px auto 0; max-width: 800px; font-size: clamp(22px, 2.3vw, 30px); line-height: 1.5; }
+ .tagline { color: #bb2037; font-weight: 700; }
+ a { color: inherit; text-underline-offset: 4px; }
+ a:hover { color: #c92432; }
+ a:focus-visible { outline: 2px solid #c92432; outline-offset: 5px; }
+ footer { display: flex; justify-content: center; flex-wrap: wrap; gap: 16px 32px; padding: 40px 24px; font-size: 18px; }
+ .terminal { background: #fff0f0; border-top: 2px solid #dc8994; }
+ pre { margin: 0; padding: clamp(12px, 3vw, 32px); overflow: auto; white-space: pre-wrap; overflow-wrap: anywhere; color: #702a36; font: 20px/1.45 ui-monospace, SFMono-Regular, Consolas, monospace; tab-size: 2; }
+ code { font: inherit; }
+ .enhanced .stage { position: sticky; top: 0; height: 100dvh; }
+ .enhanced .intro { position: absolute; inset: 0; z-index: 1; display: flex; flex-direction: column; justify-content: center; background: #fff8f8; }
+ .enhanced .terminal-track { position: absolute; inset: 0; }
+ .enhanced .terminal { height: 100%; opacity: 0; }
+ .enhanced .terminal pre { height: 100%; overflow-y: hidden; }
+ .typing code::after { content: '▍'; color: #c92432; }
+ .token-command, .token-keyword { color: #c11e39; font-weight: 700; }
+ .token-string { color: #a3482d; }
+ .token-comment { color: #8e6670; font-style: italic; }
+ .token-tag { color: #a51e4c; font-weight: 700; }
+ .token-number { color: #a32f59; }
+ .sr-only { position: absolute; width: 1px; height: 1px; margin: -1px; padding: 0; overflow: hidden; clip-path: inset(50%); white-space: nowrap; }
+ @media (max-width: 700px) {
+ .intro { padding-left: 12px; padding-right: 12px; }
+ .intro p { font-size: clamp(18px, 5vw, 24px); }
+ }
+ </style>
+</head>
+<body>
+ <main>
+ <div class="stage">
+ <header class="intro">
+ <h1><a href="https://harrisonerd.com/wred"><img src="wred-logo.png" alt="Wred" width="1774" height="887"></a></h1>
+ <p>A tiny Python/ASGI backend for HTML and JavaScript apps.</p>
+ <p class="tagline">Roll a route. Pass some JSON. Ship that shit.</p>
+ </header>
+ <section class="terminal-track" aria-label="Scroll through installation and example files">
+ <div class="terminal">
+ <pre id="source" aria-label="Installation and example files"><code>$ git clone https://gitman.io/git/patx/wred
+$ cd wred
+$ uv sync
+
+$ cat > app.py <<'EOF'
+from wred import App
+
+app = App()
+
[email protected]('/')
+async def index(request):
+ return await app.static('index.html', root='static')
+
[email protected]('/data')
+async def populate(request):
+ return [{"name": "Jim", "email": "[email protected]"}]
+EOF
+
+$ mkdir -p static
+$ cat > static/index.html <<'EOF'
+<ul id="users"></ul>
+<script type="module">
+ const users = await fetch('/data').then(r => r.json());
+ for (const user of users) {
+ const li = document.createElement('li');
+ li.textContent = `${user.name} (${user.email})`;
+ document.querySelector('#users').append(li);
+ }
+</script>
+EOF
+
+$ uv run granian --interface asginl app:app
+</code></pre>
+ </div>
+ </section>
+ </div>
+ </main>
+ <footer>
+ <a href="https://gitman.io/patx/wred">Source on Gitman</a>
+ <a href="https://harrisonerd.com">© 2026 Harrison Erd</a>
+ </footer>
+ <script>
+ const source = document.querySelector('#source');
+ const text = source.textContent;
+ const main = document.querySelector('main');
+ const stage = document.querySelector('.stage');
+ const intro = document.querySelector('.intro');
+ const terminal = document.querySelector('.terminal');
+ const screen = source.cloneNode(true);
+ screen.removeAttribute('id');
+ screen.removeAttribute('aria-label');
+ screen.setAttribute('aria-hidden', 'true');
+ source.classList.add('sr-only');
+ source.after(screen);
+ const code = screen.querySelector('code');
+ code.replaceChildren();
+ const tokens = [];
+ let offset = 0;
+ let mode = 'shell';
+
+ // Highlight once, then reveal the same text nodes as the page scrolls.
+ const syntax = /("(?:\\.|[^"\\])*"|'(?:\\.|[^'\\])*'|`(?:\\.|[^`\\])*`)|(#.*|\/\/.*)|(<\/?[a-z][\w-]*)|\b(from|import|async|def|return|await|const|for|of|new|if|throw|try|catch)\b|\b\d+\b/g;
+ function add(value, type = '') {
+ const span = document.createElement('span');
+ if (type) span.className = `token-${type}`;
+ const node = document.createTextNode('');
+ span.append(node);
+ code.append(span);
+ tokens.push({ node, value, start: offset });
+ offset += value.length;
+ }
+ for (const line of text.split(/(?<=\n)/)) {
+ if (line.startsWith('$')) {
+ add(line, 'command');
+ if (line.includes('cat > app.py')) mode = 'python';
+ if (line.includes('cat > static/index.html')) mode = 'html';
+ } else if (line.trim() === 'EOF') {
+ add(line, 'command');
+ mode = 'shell';
+ } else if (mode === 'shell') {
+ add(line);
+ } else {
+ let end = 0;
+ for (const match of line.matchAll(syntax)) {
+ add(line.slice(end, match.index));
+ const type = match[1] ? 'string' : match[2] ? 'comment' : match[3] ? 'tag' : match[4] ? 'keyword' : 'number';
+ add(match[0], type);
+ end = match.index + match[0].length;
+ }
+ add(line.slice(end));
+ }
+ }
+
+ const reducedMotion = matchMedia('(prefers-reduced-motion: reduce)');
+ let scheduled = false;
+ let previousLength = -1;
+ const fadeDistance = 140;
+ const typingDistance = 900;
+ function reveal() {
+ scheduled = false;
+ const scroll = Math.max(0, -main.getBoundingClientRect().top);
+ const fade = reducedMotion.matches ? 0 : Math.min(1, scroll / fadeDistance);
+ intro.style.opacity = 1 - fade;
+ intro.inert = fade === 1;
+ terminal.style.opacity = reducedMotion.matches ? 1 : fade;
+ const progress = Math.min(1, scroll / typingDistance);
+ const length = reducedMotion.matches ? text.length : Math.floor(progress * text.length);
+ if (length === previousLength) return;
+ previousLength = length;
+ for (const token of tokens) {
+ const visible = token.value.slice(0, Math.max(0, length - token.start));
+ if (token.node.data !== visible) token.node.data = visible;
+ }
+ screen.classList.toggle('typing', length < text.length);
+ screen.scrollTop = screen.scrollHeight;
+ }
+ function schedule() {
+ if (!scheduled) { scheduled = true; requestAnimationFrame(reveal); }
+ }
+ function layout() {
+ document.documentElement.classList.toggle('enhanced', !reducedMotion.matches);
+ screen.style.fontSize = '';
+ if (!reducedMotion.matches) {
+ // Fit the complete example, including wrapped lines, to this viewport.
+ const measure = document.createElement('pre');
+ measure.setAttribute('aria-hidden', 'true');
+ measure.textContent = text;
+ measure.style.cssText = `position:fixed; visibility:hidden; pointer-events:none; width:${screen.clientWidth}px; height:auto; font-weight:700;`;
+ document.body.append(measure);
+ let small = 1;
+ let large = 20;
+ for (let step = 0; step < 10; step++) {
+ const size = (small + large) / 2;
+ measure.style.fontSize = `${size}px`;
+ if (measure.scrollHeight <= screen.clientHeight - 4) small = size;
+ else large = size;
+ }
+ screen.style.fontSize = `${small}px`;
+ measure.remove();
+ }
+ main.style.height = reducedMotion.matches ? '' : `${stage.offsetHeight + typingDistance + 100}px`;
+ previousLength = -1;
+ schedule();
+ }
+ addEventListener('scroll', schedule, { passive: true });
+ addEventListener('resize', layout);
+ reducedMotion.addEventListener('change', layout);
+ layout();
+ </script>
+</body>
+</html>
diff --git a/docs/wred-logo.png b/docs/wred-logo.png
new file mode 100644
index 0000000..bec2a6f
Binary files /dev/null and b/docs/wred-logo.png differ
diff --git a/example_app/app.py b/example_app/app.py
new file mode 100644
index 0000000..aeb1b94
--- /dev/null
+++ b/example_app/app.py
@@ -0,0 +1,57 @@
+"""Wred Notebook: a complete, commented application built on Wred.
+
+From the repository root:
+ uv sync
+ uv run granian --interface asginl --workers 1 examples.app:app
+
+Or, inside a virtual environment:
+ python3 -m pip install .
+ granian --interface asginl --workers 1 examples.app:app
+
+Open http://127.0.0.1:8000, create an account, and write a note.
+The browser keeps its bearer token in memory and restores login with an HttpOnly cookie.
+Users, login tokens, and notes reset when this single-process demo restarts.
+Production applications need shared storage and HTTPS; rate limits here are process-local.
+
+Example layout:
+ app.py Creates the app and registers each group of routes.
+ pages.py Browser UI, static assets, redirect, health, and rate-limit demo.
+ auth.py Passwords, bearer tokens, refresh cookies, and auth endpoints.
+ notes.py Private note CRUD, validation, filtering, and export.
+ rate_limit.py Reusable application-level throttling decorator.
+ storage.py Shared in-memory users, credentials, and notes.
+ static/ Browser HTML, CSS, and JavaScript.
+
+API walkthrough:
+ POST /api/signup {"username": "alice", "password": "a long passphrase"}
+ POST /api/login Same credentials; returns access_token.
+ POST /api/refresh Restore access using the browser's refresh cookie.
+ GET /api/me Send Authorization: Bearer <access_token>.
+ GET /api/notes?q=idea&limit=5&offset=0
+ POST /api/notes {"title": "An idea", "body": "Details", "pinned": false}
+ GET /api/notes/{note_id} Read an owned note.
+ PATCH /api/notes/{note_id} Change selected fields.
+ PUT /api/notes/{note_id} Replace title/body/pinned fields.
+ DELETE /api/notes/{note_id} Returns HTTP 204 with an empty body.
+ GET /api/notes/export Download this user's notes as a JSON array.
+ POST /api/logout Revoke the token used for this request.
+ GET /api/rate-limit-demo Three requests per client IP every ten seconds.
+
+All note endpoints require the bearer header. Login and refresh also require
+X-Notebook-Request: 1, an application-level CSRF defense. The UI and API share
+an origin, so CORS is unnecessary. Wred stays unchanged.
+"""
+
+from wred import App
+
+# Each module owns its handlers and registers them on the same application.
+from . import auth, notes, pages
+
+# This is the ASGI application Granian imports using examples.app:app.
+app = App()
+# Public routes serve the browser UI and demonstrate basic responses.
+pages.register_routes(app)
+# Authentication routes handle signup, login, refresh, profile, and logout.
+auth.register_routes(app)
+# Note routes demonstrate authenticated resource access and CRUD.
+notes.register_routes(app)
diff --git a/example_app/auth.py b/example_app/auth.py
new file mode 100644
index 0000000..367c3fb
--- /dev/null
+++ b/example_app/auth.py
@@ -0,0 +1,283 @@
+"""Signup, login, refresh cookies, and bearer authentication for the demo."""
+from asyncio import to_thread
+from functools import wraps
+from hashlib import pbkdf2_hmac, sha256
+from http.cookies import CookieError, SimpleCookie
+from secrets import compare_digest, token_bytes, token_urlsafe
+from time import monotonic
+
+from wred import WredRequestError
+from .rate_limit import rate_limit
+from .storage import users, tokens, refresh_tokens
+
+# These opaque tokens are server-side login state, even though Wred has no sessions.
+token_lifetime = 3600 # Each login token is valid for one hour.
+refresh_lifetime = 7 * 24 * 3600 # A browser can restore its login for up to seven days.
+refresh_cookie_name = 'notebook_refresh' # HttpOnly means JavaScript cannot read this cookie.
+
+
+def same_origin_request(request): # Protect endpoints that issue or use browser cookies.
+ # HTML forms cannot supply this custom header; cross-origin fetch requires a preflight.
+ if request.headers.get('x-notebook-request') != '1':
+ # This example provides no cross-origin preflight permission.
+ return False
+ # Fetch Metadata also distinguishes sibling sites from this exact origin.
+ if request.headers.get('sec-fetch-site', 'same-origin') != 'same-origin':
+ # Reject cross-site and same-site-but-cross-origin browser requests.
+ return False
+ # Browsers send Origin on POST; API clients may omit it but still need the custom header.
+ origin = request.headers.get('origin')
+ # Use ASGI's scheme and Host; do not trust arbitrary forwarded-origin headers.
+ expected = f"{request.scope.get('scheme', 'http')}://{request.headers.get('host', '')}"
+ # A supplied Origin must match the origin serving this application.
+ return origin is None or origin == expected
+
+
+def refresh_cookie(request, value, max_age): # Return one Set-Cookie header pair.
+ # SimpleCookie handles cookie formatting rather than hand-building a credential header.
+ cookie = SimpleCookie()
+ # The value is a random opaque token, not a username or password.
+ cookie[refresh_cookie_name] = value
+ # Select its attributes through the cookie morsel returned by SimpleCookie.
+ morsel = cookie[refresh_cookie_name]
+ # Scope browser transmission to API routes rather than static assets.
+ morsel['path'] = '/api'
+ # Prevent JavaScript from reading the credential.
+ morsel['httponly'] = True
+ # Exclude cross-site requests as an additional CSRF defense.
+ morsel['samesite'] = 'Strict'
+ # A positive age persists across reloads; zero removes the cookie on logout.
+ morsel['max-age'] = max_age
+ # HTTPS deployments get Secure; plain HTTP remains usable for the localhost demo.
+ if request.scope.get('scheme') == 'https':
+ # A Secure cookie is only transmitted over HTTPS.
+ morsel['secure'] = True
+ # Wred encodes this string pair when sending the custom response headers.
+ return 'Set-Cookie', morsel.OutputString()
+
+
+def read_refresh_cookie(request): # Only this server reads the HttpOnly credential.
+ # Start with an empty parser when a request has no Cookie header.
+ cookie = SimpleCookie()
+ # Malformed cookie syntax must not become an unhandled server exception.
+ try:
+ # Browser requests may include other cookies alongside this one.
+ cookie.load(request.headers.get('cookie', ''))
+ # Treat an invalid cookie like an absent credential.
+ except CookieError:
+ # Validation will return Unauthorized without exposing parser details.
+ return ''
+ # Select only the application's refresh cookie.
+ morsel = cookie.get(refresh_cookie_name)
+ # Return its opaque value, or an empty string when it is absent.
+ return morsel.value if morsel else ''
+
+
+def issue_access_token(username, login_id): # Both login and refresh issue bearer credentials.
+ # Access tokens are random and remain in browser memory only.
+ token = token_urlsafe(32)
+ # Store a digest so the server does not retain the raw bearer credential.
+ digest = sha256(token.encode('utf-8')).hexdigest()
+ # Tag tokens with their login ID so logout can revoke all tokens from that login.
+ tokens[digest] = {'user_id': username, 'login_id': login_id, 'expires_at': monotonic() + token_lifetime}
+ # Return identity too, so a tab cannot silently change accounts during a refresh.
+ return {'access_token': token, 'token_type': 'bearer', 'expires_in': token_lifetime, 'user': {'id': username}}
+
+
+def revoke_login(login_id): # Revoke this login while leaving other devices' logins active.
+ # Access and refresh credentials share a login ID but live in different stores.
+ for store in (tokens, refresh_tokens):
+ # Copy entries before deleting them from the dictionary.
+ for digest, record in list(store.items()):
+ # Delete only credentials belonging to the specified login.
+ if record['login_id'] == login_id:
+ # Previously issued bearer tokens and the refresh cookie now stop working.
+ del store[digest]
+
+
+async def read_credentials(request): # Shared validation for signup and login.
+ # Parse the request as JSON; malformed JSON already produces HTTP 400 in Wred.
+ data = await request.json()
+ # Credentials must be an object, rather than an array or another JSON value.
+ if not isinstance(data, dict):
+ # Wred turns this application-raised request error into a JSON 400 response.
+ raise WredRequestError('Send a JSON object with username and password')
+ # Read both fields; missing fields are None and fail the validation below.
+ username, password = data.get('username'), data.get('password')
+ # Require strings before using string methods or passing passwords to hashing.
+ if not isinstance(username, str) or not isinstance(password, str):
+ # Treat malformed credentials as a bad request, rather than a server error.
+ raise WredRequestError('Username and password must be strings')
+ # Normalize usernames so "Alice" and "alice" refer to the same account.
+ username = username.strip().casefold()
+ # Keep usernames nonempty; allow passphrases of 15–128 characters without truncation.
+ if not 1 <= len(username) <= 64 or not 15 <= len(password) <= 128:
+ # Report the accepted lengths without exposing stored user information.
+ raise WredRequestError('Use a username of 1–64 and a password of 15–128 characters')
+ # Pass the validated values back to whichever endpoint requested them.
+ return username, password
+
+
+async def hash_password(password, salt): # Salted PBKDF2 using only Python's standard library.
+ # Run the expensive 600,000-iteration hash in a thread to keep the event loop free.
+ return await to_thread(pbkdf2_hmac, 'sha256', password.encode('utf-8'), salt, 600_000)
+
+
+async def validate_access_token(token): # Return a user identity, or None if invalid.
+ # Hash the random token to find its record without storing the bearer credential.
+ digest = sha256(token.encode('utf-8')).hexdigest()
+ # An unknown token has no record, so it cannot authenticate anyone.
+ record = tokens.get(digest)
+ # Reject missing records before checking their expiry or user ID.
+ if record is None:
+ # The protected decorator will turn this into an Unauthorized response.
+ return None
+ # Compare against a monotonic clock so system clock changes cannot extend a login.
+ if record['expires_at'] <= monotonic():
+ # Remove an expired token when it is presented; it cannot be reused.
+ tokens.pop(digest, None)
+ # An expired token is no longer proof of an authenticated login.
+ return None
+ # Expose only the identity, never the password hash or stored token record.
+ return {'id': record['user_id']}
+
+
+def protected(handler): # Wrap a handler with authentication before Wred calls it.
+ # Keep the original handler's name and docstring on the wrapper.
+ @wraps(handler)
+ async def wrapper(request, *args, **kwargs): # Preserve dynamic route parameters too.
+ # Read the Authorization header; an absent header behaves like an empty value.
+ header = request.headers.get('authorization', '')
+ # Separate "Bearer" from the token in "Authorization: Bearer <token>".
+ scheme, _, token = header.partition(' ')
+ # Start unauthenticated; only an accepted bearer token can set the user.
+ user = None
+ # Accept case-insensitive Bearer and require a nonempty credential.
+ if scheme.lower() == 'bearer' and token:
+ # Check the application's token store and expiry before invoking the handler.
+ user = await validate_access_token(token)
+ # Missing, unknown, expired, and revoked tokens all deny access.
+ if user is None:
+ # Return a JSON error with HTTP 401 without running the protected handler.
+ # WWW-Authenticate identifies Bearer as the required authentication scheme.
+ return {'error': 'Unauthorized'}, 'application/json', 401, [('WWW-Authenticate', 'Bearer')]
+ # Attach the authenticated identity for the protected handler to use.
+ request.user = user
+ # Retain this request's token so logout can revoke precisely this login.
+ request.access_token = token
+ # Forward the request and path parameters to the original async handler.
+ response = await handler(request, *args, **kwargs)
+ # Preserve custom statuses/headers; ordinary dicts and lists still become JSON.
+ if isinstance(response, tuple):
+ # The fourth tuple value is optional, so three-value responses remain valid.
+ body, content_type, status = response[:3]
+ # Copy existing headers before adding our privacy policy.
+ headers = list(response[3]) if len(response) == 4 else []
+ else:
+ # Default JSON responses use HTTP 200 without extra handler boilerplate.
+ body, content_type, status, headers = response, 'application/json', 200, []
+ # Keep every protected response out of browser and intermediary caches.
+ return body, content_type, status, headers + [('Cache-Control', 'no-store')]
+ # Give the route decorator the wrapper, rather than the unprotected function.
+ return wrapper
+
+
+def register_routes(app):
+ """Register authentication endpoints on the supplied Wred app."""
+ # Register a public signup endpoint; send {"username": "alice", "password": "a long passphrase"}.
+ @app.route('/api/signup', methods=['POST'])
+ # Limit account creation per client IP before performing expensive password hashing.
+ @rate_limit(requests=10, seconds=60)
+ async def signup(request): # Signup creates an account; login issues its access token.
+ # Parse and validate the submitted credentials using the shared helper.
+ username, password = await read_credentials(request)
+ # Give each account a unique random salt, even when users choose identical passwords.
+ salt = token_bytes(16)
+ # Derive a password hash without blocking other requests on the event loop.
+ password_hash = await hash_password(password, salt)
+ # Check after the await so simultaneous signups cannot overwrite the same account.
+ if username in users:
+ # HTTP 409 means the requested username conflicts with an existing account.
+ return {'error': 'Username already exists'}, 'application/json', 409
+ # Store only the salt and derived hash in this demo's user dictionary.
+ users[username] = {'salt': salt, 'password_hash': password_hash}
+ # Return the public identity with HTTP 201 Created; do not return password material.
+ return {'id': username}, 'application/json', 201
+
+
+ # Register a public login endpoint; submit the same username and password as signup.
+ @app.route('/api/login', methods=['POST'])
+ # Count every login attempt, including invalid credentials, toward the client IP's budget.
+ @rate_limit(requests=5, seconds=60)
+ async def login(request): # A successful login returns a new opaque bearer token.
+ # Cookie-setting login requests need the same CSRF checks as cookie-based refreshes.
+ if not same_origin_request(request):
+ # A cross-origin page must not be able to log this browser into another account.
+ return {'error': 'A same-origin request with X-Notebook-Request: 1 is required'}, 'application/json', 403
+ # Validate the JSON credentials just as signup does.
+ username, password = await read_credentials(request)
+ # Use a dummy record for unknown users so failed logins still perform password hashing.
+ record = users.get(username, {'salt': bytes(16), 'password_hash': bytes(32)})
+ # Hash the submitted password using the stored salt and the same work factor.
+ password_hash = await hash_password(password, record['salt'])
+ # Compare hashes in constant time rather than comparing the original passwords.
+ if username not in users or not compare_digest(password_hash, record['password_hash']):
+ # Use the same message for unknown users and incorrect passwords.
+ return {'error': 'Invalid credentials'}, 'application/json', 401
+ # Create one identity for this login, separate from any other device's login.
+ login_id = token_urlsafe(16)
+ # Generate a separate credential for restoring access after a page reload.
+ refresh = token_urlsafe(32)
+ # Store only its digest and a fixed absolute expiry; refreshes do not extend this deadline.
+ refresh_tokens[sha256(refresh.encode()).hexdigest()] = {'user_id': username, 'login_id': login_id, 'expires_at': monotonic() + refresh_lifetime}
+ # Return the bearer token as JSON; Set-Cookie stores the refresh token outside JavaScript.
+ return issue_access_token(username, login_id), 'application/json', 200, [('Cache-Control', 'no-store'), refresh_cookie(request, refresh, refresh_lifetime)]
+
+
+ # The browser calls this on reload and when an access token expires.
+ @app.route('/api/refresh', methods=['POST'])
+ # Refresh is cheaper than password hashing but still gets a request budget.
+ @rate_limit(requests=30, seconds=60)
+ async def refresh(request): # This endpoint alone authenticates with the refresh cookie.
+ # Require a custom header and reject cross-origin requests before inspecting cookies.
+ if not same_origin_request(request):
+ # SameSite is defense in depth, not the sole CSRF control.
+ return {'error': 'A same-origin request with X-Notebook-Request: 1 is required'}, 'application/json', 403
+ # Hash the cookie value to look up the server-side record.
+ digest = sha256(read_refresh_cookie(request).encode()).hexdigest()
+ # Unknown, expired, or revoked cookies cannot create new access tokens.
+ record = refresh_tokens.get(digest)
+ # Expiry uses the original login's deadline, even after repeated reloads.
+ if record is None or record['expires_at'] <= monotonic():
+ # Revoke related access tokens too when a known login has expired.
+ if record is not None:
+ # Remove all credentials associated with this expired login.
+ revoke_login(record['login_id'])
+ # Clear stale browser cookies and report that a new login is required.
+ return {'error': 'Unauthorized'}, 'application/json', 401, [('Cache-Control', 'no-store'), refresh_cookie(request, '', 0)]
+ # Reuse the cookie while issuing a fresh memory-only access token; concurrent tabs work.
+ return issue_access_token(record['user_id'], record['login_id']), 'application/json', 200, [('Cache-Control', 'no-store')]
+
+
+ # Register GET /api/me; the route must register the protected wrapper, not the original handler.
+ @app.route('/api/me')
+ # Put @protected below @app.route because Python applies decorators from bottom to top.
+ @protected
+ async def me(request): # This handler runs only after the token has been validated.
+ # Return the identity supplied by the decorator, rather than trusting a client user ID.
+ return request.user
+
+
+ # Register POST /api/logout; send the same Authorization header used for /api/me.
+ @app.route('/api/logout', methods=['POST'])
+ # Require a valid login before attempting to revoke its token.
+ @protected
+ async def logout(request): # Logout revokes this token; other logins remain active.
+ # Hash the authenticated token to locate exactly the record created during login.
+ digest = sha256(request.access_token.encode('utf-8')).hexdigest()
+ # Find the login that issued this valid bearer token.
+ login_id = tokens[digest]['login_id']
+ # Revoke the refresh credential and every access token issued for the same login.
+ revoke_login(login_id)
+ # Confirm logout and delete the cookie; other devices' independent logins remain active.
+ return {'logged_out': True}, 'application/json', 200, [refresh_cookie(request, '', 0)]
diff --git a/example_app/notes.py b/example_app/notes.py
new file mode 100644
index 0000000..f21b2ad
--- /dev/null
+++ b/example_app/notes.py
@@ -0,0 +1,182 @@
+"""Private note CRUD, validation, filtering, pagination, and JSON export."""
+from datetime import UTC, datetime
+from secrets import token_urlsafe
+from urllib.parse import parse_qs
+
+from wred import WredRequestError
+from .auth import protected
+from .storage import notes
+
+
+async def read_note(request, partial=False): # Validate create/replace/partial update bodies.
+ # JSON arrays and scalars are valid JSON, but this endpoint needs an object.
+ data = await request.json()
+ # Reject empty updates and misspelled fields instead of silently ignoring them.
+ if not isinstance(data, dict) or not data or data.keys() - {'title', 'body', 'pinned'}:
+ # Wred turns validation exceptions into a JSON error with HTTP 400.
+ raise WredRequestError('Send an object containing title, body, or pinned')
+ # POST and PUT require a title; PATCH may update just one other field.
+ if not partial and 'title' not in data:
+ # Explain exactly which required field is missing.
+ raise WredRequestError('A title is required')
+ # A supplied title must be a nonempty string, with room for a short heading.
+ if 'title' in data and (not isinstance(data['title'], str) or not 1 <= len(data['title'].strip()) <= 100):
+ # Keep validation in the application rather than changing Wred's JSON parser.
+ raise WredRequestError('Title must contain 1–100 characters')
+ # Bodies are optional strings; cap their length for this small in-memory example.
+ if 'body' in data and (not isinstance(data['body'], str) or len(data['body']) > 10_000):
+ # A clear message lets both the browser and API users correct the request.
+ raise WredRequestError('Body must be a string of at most 10,000 characters')
+ # JSON true/false become Python bool values; strings like "true" are not accepted.
+ if 'pinned' in data and not isinstance(data['pinned'], bool):
+ # Prevent accidental truthiness from pinning notes unexpectedly.
+ raise WredRequestError('Pinned must be a boolean')
+ # Trim titles without altering whitespace in the note's main text.
+ if 'title' in data:
+ # This normalized value is what will be saved and returned to the browser.
+ data['title'] = data['title'].strip()
+ # PATCH changes only submitted fields; POST/PUT supply defaults for omitted fields.
+ return data if partial else {'body': '', 'pinned': False, **data}
+
+
+def owned_note(note_id, user): # Authentication and resource permissions are distinct.
+ # Look up the ID from the URL in this demo's note store.
+ note = notes.get(note_id)
+ # Missing resources should return HTTP 404, not an empty successful response.
+ if note is None:
+ # Return an error tuple that the calling handler can pass straight to Wred.
+ return {'error': 'Note not found'}, 'application/json', 404
+ # A valid login does not grant access to another person's notes.
+ if note['owner'] != user['id']:
+ # HTTP 403 means authenticated, but forbidden from accessing this resource.
+ return {'error': 'This note belongs to another user'}, 'application/json', 403
+ # Only the owner receives the actual note dictionary.
+ return note
+
+
+def register_routes(app):
+ """Register note endpoints; each handler requires a valid bearer token."""
+ # GET and POST share a path but have separate handlers; the router chooses the method.
+ @app.route('/api/notes', methods=['GET'])
+ # Register the protected wrapper so unauthenticated requests never reach this handler.
+ @protected
+ async def list_notes(request): # Query strings control filtering and pagination.
+ # ASGI exposes the query string as bytes; standard-library parsing handles escaping.
+ try:
+ # Limit query fields so malformed or excessively large queries fail cleanly.
+ query = parse_qs(request.scope.get('query_string', b'').decode('utf-8'), max_num_fields=20)
+ # Invalid encoding and too many query fields are client errors too.
+ except (UnicodeDecodeError, ValueError):
+ # Keep parser errors inside Wred's existing request-error handling.
+ raise WredRequestError('Invalid query string')
+ # Treat omitted pagination values as a request for the first five notes.
+ try:
+ # parse_qs returns lists because query keys may appear more than once.
+ limit = int(query.get('limit', ['5'])[0])
+ # Offset skips a number of filtered notes, not a number of unfiltered records.
+ offset = int(query.get('offset', ['0'])[0])
+ # int() raises ValueError for text such as ?limit=abc.
+ except ValueError:
+ # Convert conversion failures into a useful client-facing JSON error.
+ raise WredRequestError('Limit and offset must be integers')
+ # Bound page size and reject negative offsets.
+ if not 1 <= limit <= 50 or offset < 0:
+ # Defaults and bounds are application policy, not router behavior.
+ raise WredRequestError('Limit must be 1–50; offset must be zero or greater')
+ # Search ignores case and whitespace around the query.
+ search = query.get('q', [''])[0].strip().casefold()
+ # The optional pinned filter accepts only explicit boolean spellings.
+ pinned = query.get('pinned', [''])[0]
+ # Reject unsupported filter values instead of guessing their meaning.
+ if pinned not in ('', 'true', 'false'):
+ # The browser and API documentation use the same query vocabulary.
+ raise WredRequestError('Pinned filter must be true or false')
+ # Apply ownership before searching or counting so other users' data never leaks.
+ items = [note for note in notes.values() if note['owner'] == request.user['id']]
+ # Match either the heading or the note body.
+ items = [note for note in items if search in (note['title'] + '\n' + note['body']).casefold()]
+ # Apply a pinned/unpinned filter only when the query requested one.
+ if pinned:
+ # Compare real booleans rather than strings stored in the note.
+ items = [note for note in items if note['pinned'] == (pinned == 'true')]
+ # Show pinned notes first, then most recently edited notes within each group.
+ items.sort(key=lambda note: (note['pinned'], note['updated_at'], note['id']), reverse=True)
+ # Wrap the page in an object so clients can also read pagination metadata.
+ return {'items': items[offset:offset + limit], 'total': len(items), 'limit': limit, 'offset': offset}
+
+
+ # POST on the same URL creates a note without a GET/POST branch in the handler.
+ @app.route('/api/notes', methods=['POST'])
+ # All notes are owned by the authenticated user, not by a submitted owner field.
+ @protected
+ async def create_note(request): # Creation returns HTTP 201 and a Location header.
+ # Validate the note before touching the store.
+ data = await read_note(request)
+ # Use a random URL-safe ID, independent of usernames and note contents.
+ note_id = token_urlsafe(9)
+ # Record UTC timestamps in a format clients can parse and display.
+ now = datetime.now(UTC).isoformat(timespec='microseconds')
+ # The application supplies owner/ID/timestamps so clients cannot impersonate others.
+ note = {'id': note_id, 'owner': request.user['id'], **data, 'created_at': now, 'updated_at': now}
+ # Save the note in this single-process demo's in-memory store.
+ notes[note_id] = note
+ # Location identifies the newly created resource; the body contains the saved note.
+ return note, 'application/json', 201, [('Location', f'/api/notes/{note_id}')]
+
+
+ # A dynamic path captures the note ID as a string passed by keyword.
+ @app.route('/api/notes/{note_id}', methods=['GET'])
+ # Validate the token before looking up any private note.
+ @protected
+ async def get_note(request, note_id): # The ownership helper also supplies 403/404 responses.
+ # Return the owner's note, or the error tuple produced by the permission check.
+ return owned_note(note_id, request.user)
+
+
+ # Two related HTTP methods can share one handler when their logic substantially overlaps.
+ @app.route('/api/notes/{note_id}', methods=['PUT', 'PATCH'])
+ # Method dispatch still happens in Wred; this branch selects replacement/update semantics.
+ @protected
+ async def update_note(request, note_id): # PUT replaces fields; PATCH changes only supplied fields.
+ # Check permissions before reading changes or mutating the note.
+ note = owned_note(note_id, request.user)
+ # Error tuples must be returned without modifying the store.
+ if isinstance(note, tuple):
+ # Preserve the ownership helper's HTTP 403 or 404.
+ return note
+ # Choose partial validation only for PATCH; PUT resets omitted optional fields.
+ data = await read_note(request, partial=request.method == 'PATCH')
+ # Apply validated fields while preserving owner, ID, and creation time.
+ note.update(data)
+ # Refresh the modification timestamp used to order the notebook.
+ note['updated_at'] = datetime.now(UTC).isoformat(timespec='microseconds')
+ # A dictionary response sends the saved note as JSON with HTTP 200.
+ return note
+
+
+ # DELETE has its own handler on the same dynamic path.
+ @app.route('/api/notes/{note_id}', methods=['DELETE'])
+ # A valid token alone is insufficient: the handler still verifies ownership.
+ @protected
+ async def delete_note(request, note_id): # Successful deletion returns HTTP 204 No Content.
+ # Reuse the exact same permission rules as reading and editing.
+ note = owned_note(note_id, request.user)
+ # Pass missing/forbidden responses through unchanged.
+ if isinstance(note, tuple):
+ # Denied deletion must leave the note in the store.
+ return note
+ # Remove the owner's note only after the permission check succeeds.
+ del notes[note_id]
+ # A 204 response must have an empty body; clients should not attempt JSON parsing.
+ return b'', 'application/json', 204
+
+
+ # This fixed path wins over /api/notes/{note_id}, even though it is registered afterward.
+ @app.route('/api/notes/export', methods=['GET'])
+ # Exports must be private to the logged-in user too.
+ @protected
+ async def export_notes(request): # An array response and a download header in one tuple.
+ # Select only this user's records; export all of them rather than just the current page.
+ items = [note for note in notes.values() if note['owner'] == request.user['id']]
+ # Wred serializes the list as a JSON array; Content-Disposition gives the download a name.
+ return items, 'application/json', 200, [('Content-Disposition', 'attachment; filename="notes.json"')]
diff --git a/example_app/pages.py b/example_app/pages.py
new file mode 100644
index 0000000..5bdb70a
--- /dev/null
+++ b/example_app/pages.py
@@ -0,0 +1,50 @@
+"""Public page/assets, redirect, health check, and a rate-limit demonstration."""
+from pathlib import Path
+
+from .rate_limit import rate_limit
+
+# Resolve assets relative to this file, not the terminal's working directory.
+static_root = Path(__file__).parent / 'static'
+
+
+def register_routes(app):
+ """Register the browser UI and public demonstration endpoints."""
+ # GET / serves the browser application through the app's built-in file helper.
+ @app.route('/')
+ async def home(request): # Every handler receives Wred's Request object.
+ # Await the file read; the helper returns (body, content_type, status).
+ body, content_type, status = await app.static('index.html', root=static_root)
+ # A fourth tuple value adds response headers without introducing another class.
+ return body, content_type, status, [('Cache-Control', 'no-cache')]
+
+
+ # This route serves the app's stylesheet and JavaScript by filename.
+ @app.route('/static/{filename}')
+ async def assets(request, filename): # Dynamic parameters arrive as strings.
+ # Confinement and missing-file handling are provided by app.static().
+ body, content_type, status = await app.static(filename, root=static_root)
+ # Avoid stale assets while experimenting with the example's source.
+ return body, content_type, status, [('Cache-Control', 'no-cache')]
+
+
+ # Old links to /ui demonstrate a real redirect using the Location header.
+ @app.route('/ui')
+ async def old_ui(request): # Redirects need an empty bytes body and a custom status.
+ # HTTP 303 tells the client to follow Location with a GET request.
+ return b'', 'text/plain', 303, [('Location', '/')]
+
+
+ # A basic public health endpoint is useful for uptime checks.
+ @app.route('/api/health')
+ async def health(request): # No authentication is required for a simple health check.
+ # Plain dictionaries still become application/json responses with HTTP 200.
+ return {'status': 'ok'}
+
+
+ # This public endpoint makes throttling easy to try without locking users out of login.
+ @app.route('/api/rate-limit-demo')
+ # The fourth and fifth rapid requests get HTTP 429 with Retry-After.
+ @rate_limit(requests=3, seconds=10)
+ async def rate_limit_demo(request): # The frontend has a button that sends five requests.
+ # A successful request is an ordinary JSON dictionary response.
+ return {'message': 'Request accepted'}
diff --git a/example_app/rate_limit.py b/example_app/rate_limit.py
new file mode 100644
index 0000000..68b74ce
--- /dev/null
+++ b/example_app/rate_limit.py
@@ -0,0 +1,47 @@
+"""Application-owned sliding-window rate limiting, independent of Wred."""
+from collections import deque
+from functools import wraps
+from math import ceil
+from time import monotonic
+
+rate_buckets = {} # (handler, client IP, window) -> accepted request timestamps.
+
+
+def rate_limit(requests, seconds): # A sliding-window decorator, owned by this application.
+ # Construct a decorator that can be placed beneath @app.route.
+ def decorator(handler):
+ # Preserve the original function's identity and documentation.
+ @wraps(handler)
+ async def wrapper(request, *args, **kwargs):
+ # A monotonic clock makes windows independent of wall-clock adjustments.
+ now = monotonic()
+ # Drop inactive clients so their buckets do not remain in memory forever.
+ for key, timestamps in list(rate_buckets.items()):
+ # Each key records its own window length, which may differ by endpoint.
+ if timestamps[-1] <= now - key[2]:
+ # The whole bucket has aged out of its window.
+ del rate_buckets[key]
+ # Use ASGI's client address, not an untrusted X-Forwarded-For header.
+ client = (request.scope.get('client') or ('unknown', 0))[0]
+ # Keep different endpoints' request budgets independent.
+ key = (handler.__name__, client, seconds)
+ # A deque efficiently removes the oldest accepted requests.
+ timestamps = rate_buckets.setdefault(key, deque())
+ # Expire each timestamp that no longer belongs to this sliding window.
+ while timestamps and timestamps[0] <= now - seconds:
+ # Newer timestamps stay in order at the right end.
+ timestamps.popleft()
+ # Reject excess requests before password hashing or other handler work begins.
+ if len(timestamps) >= requests:
+ # Tell clients exactly how many whole seconds to wait before retrying.
+ retry_after = max(1, ceil(timestamps[0] + seconds - now))
+ # Our fourth response tuple value carries Retry-After and a cache policy.
+ return {'error': 'Too many requests', 'retry_after': retry_after}, 'application/json', 429, [('Retry-After', str(retry_after)), ('Cache-Control', 'no-store')]
+ # Reserve the slot before any await, so concurrent requests cannot bypass it.
+ timestamps.append(now)
+ # Allow the original handler to process the request and route parameters.
+ return await handler(request, *args, **kwargs)
+ # Give Wred the limited wrapper instead of the original function.
+ return wrapper
+ # Return the configured decorator for use on each endpoint.
+ return decorator
diff --git a/example_app/static/app.css b/example_app/static/app.css
new file mode 100644
index 0000000..787478b
--- /dev/null
+++ b/example_app/static/app.css
@@ -0,0 +1,126 @@
+:root {
+ color-scheme: light;
+ --ink: #172746;
+ --muted: #64748b;
+ --blue: #2457d6;
+ --paper: #fff;
+ --canvas: #f5f7fc;
+ --line: #d7e0ee;
+ font-family: "Segoe UI", "Helvetica Neue", Arial, sans-serif;
+ color: var(--ink);
+ background: var(--canvas);
+ font-synthesis: none;
+}
+* { box-sizing: border-box; }
+[hidden] { display: none !important; }
+body { margin: 0; }
+button, input, textarea { font: inherit; }
+button { cursor: pointer; }
+button:disabled { cursor: default; opacity: .45; }
+a { color: inherit; }
+button:focus-visible, a:focus-visible { outline: 3px solid #9bb9ff; outline-offset: 4px; }
+input:focus, textarea:focus { outline: 2px solid #9bb9ff; outline-offset: 2px; border-color: var(--blue); }
+button { border: 0; border-radius: 6px; padding: 11px 16px; font-size: 14px; font-weight: 600; }
+.primary { color: #fff; background: var(--blue); }
+.primary:hover:not(:disabled) { background: #1944b1; }
+.quiet { color: var(--ink); background: transparent; }
+.quiet:hover:not(:disabled) { background: #e9eef9; }
+.danger { color: #a1353d; background: transparent; }
+.danger:hover:not(:disabled) { background: #fcebee; }
+.small { padding: 9px 12px; white-space: nowrap; }
+.shell { max-width: 1260px; margin: auto; padding: 0 38px; }
+.app-header { display: flex; justify-content: space-between; align-items: center; min-height: 100px; gap: 24px; }
+.brand { display: inline-flex; gap: 12px; align-items: center; font-size: 19px; font-weight: 700; text-decoration: none; letter-spacing: -.5px; }
+.book-mark { width: 24px; height: 28px; border: 2px solid var(--blue); border-left: 7px solid var(--blue); border-radius: 2px 5px 5px 2px; transform: rotate(-5deg); }
+.account { display: flex; align-items: center; gap: 8px; font-size: 14px; }
+#username { margin-right: 12px; color: var(--muted); overflow-wrap: anywhere; }
+.notice { margin: 0 0 20px; padding: 14px 18px; color: #244f48; background: #e5f2ee; border-radius: 6px; line-height: 1.5; }
+.notice.error { color: #902c39; background: #fcebee; }
+.welcome { display: grid; grid-template-columns: 1.15fr 1fr; align-items: center; gap: 90px; min-height: 660px; padding: 40px 35px 65px; }
+.introduction h1 { font-size: clamp(44px, 5.1vw, 68px); line-height: 1.06; font-weight: 650; letter-spacing: -3px; margin: 0 0 25px; }
+.introduction > p { color: var(--muted); max-width: 360px; font-size: 18px; line-height: 1.65; margin-bottom: 40px; }
+.paper-preview { max-width: 385px; padding: 28px 32px 24px 40px; background: var(--paper); border: 1px solid var(--line); border-left: 4px solid #9bb9ff; transform: rotate(-3deg); box-shadow: 0 12px 30px #17274608; }
+.preview-title { display: block; margin-bottom: 17px; font-family: Georgia, serif; font-size: 23px; }
+.preview-line { display: block; min-height: 34px; line-height: 33px; border-bottom: 1px solid #e8edf5; font-family: Georgia, serif; font-size: 14px; color: var(--muted); }
+.auth-panel { padding: 32px; background: var(--paper); border: 1px solid var(--line); border-radius: 10px; }
+.auth-switch { display: flex; gap: 4px; background: #eef2f9; border-radius: 6px; padding: 4px; margin-bottom: 28px; }
+.auth-switch button { width: 50%; color: var(--muted); background: transparent; }
+.auth-switch button[aria-pressed="true"] { background: var(--paper); color: var(--ink); box-shadow: 0 1px 3px #17274612; }
+.auth-panel h2 { margin: 0 0 9px; font-size: 25px; letter-spacing: -.7px; }
+.auth-panel > p { margin: 0 0 26px; color: var(--muted); line-height: 1.5; }
+label { display: block; font-size: 13px; font-weight: 600; margin-bottom: 9px; }
+input:not([type="checkbox"]), textarea { width: 100%; color: var(--ink); background: #fff; border: 1px solid var(--line); border-radius: 6px; padding: 12px 14px; }
+.auth-panel input { margin-bottom: 21px; }
+.auth-panel input[type="password"], .auth-panel input[name="password"] { margin-bottom: 8px; }
+.field-help { font-size: 12px; color: var(--muted); margin: 0 0 25px; line-height: 1.5; }
+.auth-panel .primary { width: 100%; }
+.workspace { display: grid; grid-template-columns: 350px minmax(0, 1fr); min-height: 680px; background: var(--paper); border: 1px solid var(--line); border-radius: 10px; overflow: hidden; }
+.notes-pane { border-right: 1px solid var(--line); padding: 26px 23px; background: #fafbfe; }
+.pane-heading, .editor-heading { display: flex; justify-content: space-between; align-items: center; gap: 14px; }
+.pane-heading { margin-bottom: 24px; }
+.pane-heading h1 { font-size: 22px; margin: 0; letter-spacing: -.6px; }
+.notes-pane input[type="search"] { font-size: 14px; }
+.check { display: flex; align-items: center; gap: 9px; font-weight: 400; font-size: 13px; margin: 0; }
+.check input { accent-color: var(--blue); width: 15px; height: 15px; margin: 0; }
+.filter { margin: 16px 0 23px; }
+.list-summary { font-size: 12px; color: var(--muted); margin: 0 0 12px; }
+.note-list { display: grid; gap: 8px; }
+.note-item { display: block; width: 100%; padding: 16px; text-align: left; color: var(--ink); border: 1px solid transparent; background: transparent; }
+.note-item:hover { background: #eef2fa; }
+.note-item[aria-current="true"] { border-color: #b1c6f7; background: #eef3ff; }
+.note-item strong { display: block; font-size: 15px; font-weight: 650; overflow-wrap: anywhere; }
+.note-item .excerpt { display: block; font-family: Georgia, serif; color: var(--muted); font-size: 13px; font-weight: 400; line-height: 1.6; margin-top: 7px; overflow-wrap: anywhere; }
+.note-item .note-meta { display: block; color: var(--muted); font-size: 11px; font-weight: 400; margin-top: 12px; }
+.empty-list { margin-top: 45px; color: var(--muted); font-family: Georgia, serif; line-height: 1.7; font-size: 17px; }
+.pagination { display: flex; justify-content: space-between; align-items: center; gap: 3px; margin-top: 20px; }
+.pagination button { padding: 9px 6px; font-size: 12px; }
+.pagination span { color: var(--muted); font-size: 12px; }
+.editor { position: relative; padding: 30px 38px 25px 48px; }
+.editor::before { content: ""; position: absolute; top: 104px; bottom: 105px; left: 25px; border-left: 1px solid #c7d7f7; }
+.editor-heading { margin-bottom: 34px; }
+.editor-heading h2 { font-size: 16px; font-weight: 600; margin: 0; }
+.save-state { color: var(--muted); font-size: 12px; }
+.editor label { color: var(--muted); }
+#note-title { font-family: Georgia, serif; font-size: 28px; padding: 10px 0 14px; border: 0; border-bottom: 1px solid var(--line); border-radius: 0; margin-bottom: 25px; }
+#note-body { font-family: Georgia, serif; font-size: 17px; line-height: 1.75; border: 0; padding: 8px 0; resize: vertical; height: 300px; min-height: 295px; }
+#note-title::placeholder { color: #a0aabc; }
+#note-body::placeholder { color: #a0aabc; }
+.editor-details { display: flex; justify-content: space-between; align-items: center; color: var(--muted); font-size: 12px; gap: 14px; padding: 20px 0; border-top: 1px solid var(--line); }
+.editor-actions { display: flex; flex-wrap: wrap; align-items: center; gap: 7px; }
+.editor-actions .danger { margin-left: auto; }
+.note-date { margin: 24px 0 0; color: var(--muted); font-size: 11px; }
+.app-footer { padding: 25px 0; text-align: center; color: var(--muted); font-size: 12px; line-height: 1.6; }
+.demo-tools { margin-top: 20px; padding: 16px 20px; color: var(--muted); border: 1px solid var(--line); border-radius: 6px; font-size: 13px; }
+.demo-tools summary { cursor: pointer; color: var(--ink); font-weight: 600; }
+.demo-tools p { line-height: 1.6; }
+.demo-tools button { background: #e9eef9; }
+.sr-only { position: absolute; width: 1px; height: 1px; padding: 0; margin: -1px; overflow: hidden; clip: rect(0, 0, 0, 0); white-space: nowrap; border: 0; }
+@media (max-width: 1000px) {
+ .shell { padding: 0 24px; }
+ .welcome { gap: 40px; padding: 30px 0 60px; }
+ .workspace { grid-template-columns: 300px minmax(0, 1fr); }
+ .editor { padding-right: 26px; padding-left: 35px; }
+ .editor::before { left: 15px; }
+}
+@media (max-width: 760px) {
+ .shell { padding: 0 18px; }
+ .app-header { min-height: 85px; gap: 14px; flex-wrap: wrap; padding: 18px 0; }
+ .account { width: 100%; justify-content: flex-end; gap: 3px; }
+ #username { margin-right: auto; }
+ .welcome { grid-template-columns: 1fr; gap: 28px; min-height: auto; padding: 25px 0; }
+ .introduction h1 { font-size: 44px; letter-spacing: -1.8px; }
+ .introduction > p { max-width: 100%; margin-bottom: 0; font-size: 16px; }
+ .paper-preview { display: none; }
+ .auth-panel { padding: 25px; }
+ .workspace { grid-template-columns: 1fr; }
+ .notes-pane { border-right: 0; border-bottom: 1px solid var(--line); padding: 22px; }
+ .note-list { max-height: 270px; overflow-y: auto; }
+ .editor { padding: 26px 24px 26px 35px; }
+ .editor::before { left: 17px; }
+ .editor-heading { margin-bottom: 25px; }
+ #note-title { font-size: 25px; }
+ #note-body { min-height: 260px; }
+ .editor-actions .danger { margin-left: 0; }
+ .empty-list { margin-top: 20px; font-size: 15px; }
+ .app-footer { text-align: left; }
+}
diff --git a/example_app/static/app.js b/example_app/static/app.js
new file mode 100644
index 0000000..c7a4405
--- /dev/null
+++ b/example_app/static/app.js
@@ -0,0 +1,330 @@
+// Bearer tokens stay in memory. The server's HttpOnly refresh cookie restores login.
+const state = { token: null, user: null, selected: null, notes: [], total: 0, offset: 0, limit: 5, mode: 'login' };
+const $ = id => document.getElementById(id);
+let listSequence = 0;
+let selectionSequence = 0;
+let searchTimer;
+let sessionGeneration = 0;
+let refreshPromise = null;
+
+function notice(message, error = false) {
+ $('notice').textContent = message;
+ $('notice').classList.toggle('error', error);
+ $('notice').hidden = !message;
+}
+
+// This one client helper handles JSON, bearer auth, errors, and empty 204 responses.
+async function api(path, { method = 'GET', data, authenticated = true, retry = true } = {}) {
+ const token = state.token;
+ const generation = sessionGeneration;
+ // Cookie-using endpoints require this header; cross-origin HTML forms cannot set it.
+ const headers = { 'X-Notebook-Request': '1' };
+ if (data !== undefined) headers['Content-Type'] = 'application/json';
+ if (authenticated && token) headers.Authorization = `Bearer ${token}`;
+ let response;
+ try {
+ response = await fetch(path, { method, headers, credentials: 'same-origin', cache: 'no-store',
+ body: data === undefined ? undefined : JSON.stringify(data) });
+ } catch {
+ throw new Error('Could not reach the notebook. Check that the server is running and try again.');
+ }
+ const result = response.status === 204 ? null : await response.json();
+ // Ignore a response from a previous login rather than exposing it in another account.
+ if (authenticated && generation !== sessionGeneration) throw new DOMException('Login changed', 'AbortError');
+ // Share one refresh across concurrent requests, then retry the original request once.
+ if (response.status === 401 && authenticated && retry) {
+ try { await renewLogin(); }
+ catch (error) {
+ if (error.status === 401) {
+ leaveNotebook();
+ throw new Error('Your login has ended. Log in again to continue.');
+ }
+ throw error;
+ }
+ return api(path, { method, data, authenticated, retry: false });
+ }
+ if (response.status === 401 && authenticated) {
+ leaveNotebook();
+ throw new Error('Your login has ended. Log in again to continue.');
+ }
+ if (!response.ok) {
+ const wait = Number(response.headers.get('Retry-After')) || result.retry_after || 0;
+ const error = new Error((result.error || `Request failed (${response.status}).`) + (wait ? `. Try again in ${wait} seconds.` : ''));
+ error.status = response.status;
+ error.retryAfter = wait;
+ throw error;
+ }
+ return { data: result, headers: response.headers };
+}
+
+function renewLogin() {
+ if (!refreshPromise) {
+ const generation = sessionGeneration;
+ const previousUser = state.user?.id;
+ refreshPromise = api('/api/refresh', { method: 'POST', authenticated: false })
+ .then(({ data }) => {
+ if (generation !== sessionGeneration) throw new DOMException('Login changed', 'AbortError');
+ // Another tab may have logged into a different account using the shared cookie.
+ // Never retry an old account's pending write using that new account's identity.
+ if (previousUser && previousUser !== data.user.id) {
+ leaveNotebook();
+ throw new Error('This browser changed accounts. Reload to open the current notebook.');
+ }
+ state.token = data.access_token;
+ return data;
+ }).finally(() => { refreshPromise = null; });
+ }
+ return refreshPromise;
+}
+
+async function enterNotebook(login, message) {
+ sessionGeneration++;
+ state.token = login.access_token;
+ const { data: user } = await api('/api/me');
+ state.user = user;
+ $('auth-password').value = '';
+ $('username').textContent = user.id;
+ $('welcome').hidden = true;
+ $('workspace').hidden = false;
+ $('account').hidden = false;
+ showNote(null);
+ await loadNotes();
+ notice(message);
+}
+
+async function restoreNotebook() {
+ const buttons = [$('auth-submit'), $('login-mode'), $('signup-mode')];
+ buttons.forEach(button => { button.disabled = true; });
+ notice('Checking your login…');
+ try { await enterNotebook(await renewLogin(), 'Welcome back.'); }
+ catch (error) {
+ leaveNotebook();
+ // A new visitor simply has no refresh cookie; other failures deserve a visible message.
+ notice(error.status === 401 ? '' : error.message, error.status !== 401);
+ }
+ finally { buttons.forEach(button => { button.disabled = false; }); }
+}
+
+async function action(callback, button) {
+ if (button) { button.disabled = true; button.setAttribute('aria-busy', 'true'); }
+ try { await callback(); }
+ catch (error) { if (error.name !== 'AbortError') notice(error.message, true); }
+ finally { if (button) { button.disabled = false; button.removeAttribute('aria-busy'); } }
+}
+
+function authMode(mode) {
+ state.mode = mode;
+ const signup = mode === 'signup';
+ $('signup-mode').setAttribute('aria-pressed', String(signup));
+ $('login-mode').setAttribute('aria-pressed', String(!signup));
+ $('auth-title').textContent = signup ? 'Start your notebook' : 'Open your notebook';
+ $('auth-description').textContent = signup ? 'Choose a username and a long password. Your notes belong to you.' : 'Log in to pick up where you left off.';
+ $('auth-submit').textContent = signup ? 'Create account' : 'Log in';
+ $('auth-password').autocomplete = signup ? 'new-password' : 'current-password';
+ notice('');
+}
+
+function leaveNotebook() {
+ sessionGeneration++;
+ state.token = null;
+ state.user = null;
+ state.notes = [];
+ state.total = 0;
+ state.offset = 0;
+ listSequence++;
+ selectionSequence++;
+ $('note-list').replaceChildren();
+ $('search').value = '';
+ $('pinned-filter').checked = false;
+ showNote(null);
+ $('workspace').hidden = true;
+ $('account').hidden = true;
+ $('welcome').hidden = false;
+ $('username').textContent = '';
+ $('auth-password').value = '';
+}
+
+function dirty() {
+ const original = state.selected || { title: '', body: '', pinned: false };
+ return $('note-title').value !== original.title || $('note-body').value !== original.body || $('note-pinned').checked !== original.pinned;
+}
+
+function mayDiscard() {
+ return !dirty() || window.confirm('Discard your unsaved changes?');
+}
+
+function updateDraftState() {
+ $('character-count').textContent = `${$('note-body').value.length.toLocaleString()} / 10,000`;
+ $('save-state').textContent = dirty() ? 'Unsaved changes' : state.selected ? 'Saved' : 'Ready when you are';
+}
+
+function showNote(note) {
+ state.selected = note ? structuredClone(note) : null;
+ $('note-title').value = note?.title || '';
+ $('note-body').value = note?.body || '';
+ $('note-pinned').checked = note?.pinned || false;
+ $('editor-heading').textContent = note ? 'Edit note' : 'New note';
+ $('save-note').textContent = note ? 'Save changes' : 'Save note';
+ $('discard').textContent = note ? 'Discard changes' : 'Clear draft';
+ $('delete-note').hidden = !note;
+ $('note-date').textContent = note ? `Last saved ${new Date(note.updated_at).toLocaleString()}` : '';
+ updateDraftState();
+}
+
+function renderNotes() {
+ $('note-list').replaceChildren();
+ for (const note of state.notes) {
+ const button = document.createElement('button');
+ button.type = 'button';
+ button.className = 'note-item';
+ button.setAttribute('aria-current', String(note.id === state.selected?.id));
+ const title = document.createElement('strong');
+ title.textContent = note.title;
+ const excerpt = document.createElement('span');
+ excerpt.className = 'excerpt';
+ excerpt.textContent = note.body.replace(/\s+/g, ' ').slice(0, 85) || 'Empty note';
+ const meta = document.createElement('span');
+ meta.className = 'note-meta';
+ meta.textContent = `${note.pinned ? 'Pinned · ' : ''}${new Date(note.updated_at).toLocaleDateString(undefined, { month: 'short', day: 'numeric' })}`;
+ button.append(title, excerpt, meta);
+ button.addEventListener('click', () => action(() => selectNote(note.id)));
+ $('note-list').append(button);
+ }
+ $('empty-list').hidden = state.total !== 0;
+ $('empty-list').textContent = $('search').value || $('pinned-filter').checked
+ ? 'No notes match this filter. Try another search or turn off Pinned only.'
+ : 'Your first note starts in the editor. Give it a title and save it.';
+ $('list-summary').textContent = state.total ? `${state.offset + 1}–${Math.min(state.offset + state.limit, state.total)} of ${state.total} notes` : 'No notes yet';
+ $('pagination').hidden = state.total <= state.limit;
+ $('previous').disabled = state.offset === 0;
+ $('next').disabled = state.offset + state.limit >= state.total;
+ $('page-number').textContent = `Page ${Math.floor(state.offset / state.limit) + 1}`;
+}
+
+async function loadNotes() {
+ const sequence = ++listSequence;
+ const query = new URLSearchParams({ limit: state.limit, offset: state.offset });
+ const search = $('search').value.trim();
+ if (search) query.set('q', search);
+ if ($('pinned-filter').checked) query.set('pinned', 'true');
+ const { data } = await api(`/api/notes?${query}`);
+ if (sequence !== listSequence) return;
+ // If deletion emptied the last page, show the last remaining page instead.
+ if (data.total && state.offset >= data.total) {
+ state.offset = Math.floor((data.total - 1) / state.limit) * state.limit;
+ return loadNotes();
+ }
+ if (!data.total) state.offset = 0;
+ state.notes = data.items;
+ state.total = data.total;
+ renderNotes();
+}
+
+async function selectNote(id) {
+ if (id === state.selected?.id || !mayDiscard()) return;
+ const sequence = ++selectionSequence;
+ const { data } = await api(`/api/notes/${encodeURIComponent(id)}`);
+ if (sequence !== selectionSequence) return;
+ showNote(data);
+ renderNotes();
+ notice('');
+ if (window.matchMedia('(max-width: 760px)').matches) $('editor-heading').scrollIntoView({ block: 'start' });
+}
+
+$('login-mode').addEventListener('click', () => authMode('login'));
+$('signup-mode').addEventListener('click', () => authMode('signup'));
+$('auth-form').addEventListener('submit', event => {
+ event.preventDefault();
+ const credentials = { username: $('auth-username').value, password: $('auth-password').value };
+ const signup = state.mode === 'signup';
+ action(async () => {
+ if (signup) await api('/api/signup', { method: 'POST', data: credentials, authenticated: false });
+ const { data: login } = await api('/api/login', { method: 'POST', data: credentials, authenticated: false });
+ await enterNotebook(login, signup ? 'Your notebook is ready. Write your first note.' : 'Welcome back.');
+ $('note-title').focus();
+ }, $('auth-submit'));
+});
+
+$('note-form').addEventListener('input', updateDraftState);
+$('note-form').addEventListener('submit', event => {
+ event.preventDefault();
+ const id = state.selected?.id;
+ const values = { title: $('note-title').value, body: $('note-body').value, pinned: $('note-pinned').checked };
+ action(async () => {
+ const { data } = await api(id ? `/api/notes/${encodeURIComponent(id)}` : '/api/notes', { method: id ? 'PATCH' : 'POST', data: values });
+ selectionSequence++;
+ showNote(data);
+ state.offset = 0;
+ await loadNotes();
+ notice('Note saved.');
+ }, $('save-note'));
+});
+
+$('new-note').addEventListener('click', () => {
+ if (!mayDiscard()) return;
+ selectionSequence++;
+ showNote(null);
+ renderNotes();
+ notice('');
+ $('note-title').focus();
+});
+$('discard').addEventListener('click', () => {
+ if (!mayDiscard()) return;
+ showNote(state.selected);
+ notice('');
+});
+$('delete-note').addEventListener('click', () => {
+ const id = state.selected?.id;
+ if (!id || !window.confirm('Delete this note? This cannot be undone.')) return;
+ action(async () => {
+ await api(`/api/notes/${encodeURIComponent(id)}`, { method: 'DELETE' });
+ selectionSequence++;
+ showNote(null);
+ await loadNotes();
+ notice('Note deleted.');
+ }, $('delete-note'));
+});
+$('search').addEventListener('input', () => {
+ clearTimeout(searchTimer);
+ searchTimer = setTimeout(() => { state.offset = 0; action(loadNotes); }, 250);
+});
+$('pinned-filter').addEventListener('change', () => { state.offset = 0; action(loadNotes); });
+$('previous').addEventListener('click', () => { state.offset = Math.max(0, state.offset - state.limit); action(loadNotes); });
+$('next').addEventListener('click', () => { state.offset += state.limit; action(loadNotes); });
+$('export').addEventListener('click', () => action(async () => {
+ const { data } = await api('/api/notes/export');
+ const url = URL.createObjectURL(new Blob([JSON.stringify(data, null, 2)], { type: 'application/json' }));
+ const link = document.createElement('a');
+ link.href = url;
+ link.download = 'notes.json';
+ document.body.append(link);
+ link.click();
+ link.remove();
+ setTimeout(() => URL.revokeObjectURL(url), 1000);
+ notice('Your notes have been exported.');
+}, $('export')));
+$('logout').addEventListener('click', () => action(async () => {
+ if (!mayDiscard()) return;
+ await api('/api/logout', { method: 'POST' });
+ leaveNotebook();
+ notice('Logged out.');
+}, $('logout')));
+
+// A harmless endpoint makes the request limit visible without exhausting login attempts.
+$('rate-demo').addEventListener('click', () => action(async () => {
+ let accepted = 0;
+ let limited = 0;
+ let wait = 0;
+ for (let index = 0; index < 5; index++) {
+ try { await api('/api/rate-limit-demo', { authenticated: false }); accepted++; }
+ catch (error) {
+ if (error.status !== 429) throw error;
+ limited++;
+ wait = Math.max(wait, error.retryAfter);
+ }
+ }
+ $('rate-result').textContent = `${accepted} accepted; ${limited} limited (HTTP 429).${wait ? ` Try again in ${wait} seconds.` : ''}`;
+}, $('rate-demo')));
+
+// The browser sends the HttpOnly cookie automatically; JS never reads its value.
+restoreNotebook();
diff --git a/example_app/static/index.html b/example_app/static/index.html
new file mode 100644
index 0000000..63aebcc
--- /dev/null
+++ b/example_app/static/index.html
@@ -0,0 +1,106 @@
+<!doctype html>
+<html lang="en">
+<head>
+ <meta charset="utf-8">
+ <meta name="viewport" content="width=device-width, initial-scale=1">
+ <meta name="theme-color" content="#2457d6">
+ <title>Wred notebook</title>
+ <link rel="icon" href="data:,">
+ <link rel="stylesheet" href="/static/app.css">
+ <script src="/static/app.js" defer></script>
+</head>
+<body>
+ <div class="shell">
+ <header class="app-header">
+ <a class="brand" href="/" aria-label="Wred notebook home"><span class="book-mark" aria-hidden="true"></span>Wred notebook</a>
+ <div id="account" class="account" hidden>
+ <span id="username"></span>
+ <button id="export" class="quiet" type="button">Export notes</button>
+ <button id="logout" class="quiet" type="button">Log out</button>
+ </div>
+ </header>
+
+ <p id="notice" class="notice" role="status" aria-live="polite" hidden></p>
+
+ <main id="welcome" class="welcome">
+ <section class="introduction">
+ <h1>A notebook<br>of your own.</h1>
+ <p>Keep your ideas, lists, and little reminders in one place. Find them when you need them.</p>
+ <div class="paper-preview" aria-hidden="true">
+ <span class="preview-title">Things to remember</span>
+ <span class="preview-line">Leave a little room for the next idea.</span>
+ <span class="preview-line">Start with a blank page.</span>
+ <span class="preview-line"></span>
+ </div>
+ </section>
+ <section class="auth-panel" aria-labelledby="auth-title">
+ <div class="auth-switch" aria-label="Choose how to sign in">
+ <button id="login-mode" type="button" aria-pressed="true">Log in</button>
+ <button id="signup-mode" type="button" aria-pressed="false">Create account</button>
+ </div>
+ <h2 id="auth-title">Open your notebook</h2>
+ <p id="auth-description">Log in to pick up where you left off.</p>
+ <form id="auth-form">
+ <label for="auth-username">Username</label>
+ <input id="auth-username" name="username" autocomplete="username" maxlength="64" required>
+ <label for="auth-password">Password</label>
+ <input id="auth-password" name="password" type="password" autocomplete="current-password" minlength="15" maxlength="128" aria-describedby="password-help" required>
+ <p id="password-help" class="field-help">Use 15–128 characters. Spaces are welcome.</p>
+ <button id="auth-submit" class="primary" type="submit">Log in</button>
+ </form>
+ </section>
+ </main>
+
+ <main id="workspace" class="workspace" hidden>
+ <aside class="notes-pane" aria-labelledby="notes-heading">
+ <div class="pane-heading">
+ <h1 id="notes-heading">Your notes</h1>
+ <button id="new-note" class="primary small" type="button">+ New note</button>
+ </div>
+ <label class="sr-only" for="search">Search notes</label>
+ <input id="search" type="search" placeholder="Search your notes…" maxlength="200">
+ <label class="check filter"><input id="pinned-filter" type="checkbox"> Pinned only</label>
+ <p id="list-summary" class="list-summary" aria-live="polite"></p>
+ <div id="note-list" class="note-list" aria-label="Notes"></div>
+ <p id="empty-list" class="empty-list" hidden>Your first note starts on the right. Give it a title and save it.</p>
+ <nav id="pagination" class="pagination" aria-label="Notes pages" hidden>
+ <button id="previous" class="quiet" type="button">← Previous</button>
+ <span id="page-number"></span>
+ <button id="next" class="quiet" type="button">Next →</button>
+ </nav>
+ </aside>
+
+ <section class="editor" aria-labelledby="editor-heading">
+ <div class="editor-heading">
+ <h2 id="editor-heading">New note</h2>
+ <span id="save-state" class="save-state">Ready when you are</span>
+ </div>
+ <form id="note-form">
+ <label for="note-title">Title</label>
+ <input id="note-title" name="title" placeholder="Give this thought a title" maxlength="100" required>
+ <label for="note-body">Note</label>
+ <textarea id="note-body" name="body" placeholder="Start writing…" maxlength="10000" rows="12"></textarea>
+ <div class="editor-details">
+ <label class="check"><input id="note-pinned" name="pinned" type="checkbox"> Keep at top</label>
+ <span id="character-count">0 / 10,000</span>
+ </div>
+ <div class="editor-actions">
+ <button id="save-note" class="primary" type="submit">Save note</button>
+ <button id="discard" class="quiet" type="button">Clear draft</button>
+ <button id="delete-note" class="danger" type="button" hidden>Delete note</button>
+ </div>
+ </form>
+ <p id="note-date" class="note-date"></p>
+ </section>
+ </main>
+
+ <details class="demo-tools">
+ <summary>Try request limits</summary>
+ <p>This demo endpoint accepts three requests every ten seconds. Extra requests return HTTP 429 and tell you when to retry.</p>
+ <button id="rate-demo" class="quiet" type="button">Send 5 requests</button>
+ <p id="rate-result" role="status" aria-live="polite"></p>
+ </details>
+ <footer class="app-footer">This is a demo. Your login survives reloads for up to seven days; data and logins reset when the server restarts.</footer>
+ </div>
+</body>
+</html>
diff --git a/example_app/storage.py b/example_app/storage.py
new file mode 100644
index 0000000..6f9c8d9
--- /dev/null
+++ b/example_app/storage.py
@@ -0,0 +1,7 @@
+"""Shared in-memory demo data; all records disappear when the process restarts."""
+
+# Real deployments need persistent shared storage and shared rate-limit counters.
+users = {} # Username -> salt and password hash; never store plaintext passwords.
+tokens = {} # Token digest -> user ID and expiry; never store raw bearer tokens.
+refresh_tokens = {} # Cookie digest -> user ID, login ID, and absolute expiry.
+notes = {} # Note ID -> note, including the owning username.
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..c8f4fc2
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,14 @@
+[build-system]
+requires = ["flit_core>=3.11,<5"]
+build-backend = "flit_core.buildapi"
+
+[project]
+name = "wred"
+dynamic = ["version"]
+description = "An ultra-minimal ASGI framework for JSON APIs."
+readme = "README.md"
+requires-python = ">=3.11"
+license = "BSD-3-Clause"
+authors = [{ name = "Harrison Erd", email = "[email protected]" }]
+dependencies = ["orjson", "granian"]
+
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/test_notebook.py b/tests/test_notebook.py
new file mode 100644
index 0000000..f006f5c
--- /dev/null
+++ b/tests/test_notebook.py
@@ -0,0 +1,314 @@
+"""Exercise the example's API contracts through the actual Wred ASGI app."""
+import importlib
+import asyncio
+from http.cookies import SimpleCookie
+import unittest
+from urllib.parse import unquote, urlsplit
+from unittest.mock import patch
+
+import orjson
+
+from tests.test_wred import call_app
+from examples import auth, storage, rate_limit
+
+notebook = importlib.import_module('examples.app')
+
+
+class NotebookTests(unittest.IsolatedAsyncioTestCase):
+ def setUp(self):
+ storage.users.clear()
+ storage.tokens.clear()
+ storage.refresh_tokens.clear()
+ rate_limit.rate_buckets.clear()
+ storage.notes.clear()
+
+ async def request(self, path, method='GET', data=None, token=None, authorization=None,
+ cookie=None, client='127.0.0.1', csrf_header=True, extra_headers=(), scheme='http'):
+ url = urlsplit(path)
+
+ async def receive():
+ if method == 'GET':
+ raise AssertionError('GET handlers should not consume the request body')
+ return {'type': 'http.request', 'body': orjson.dumps(data)}
+
+ headers = [(b'content-type', b'application/json'), (b'host', b'notebook.test')]
+ if csrf_header:
+ headers.append((b'x-notebook-request', b'1'))
+ if cookie is not None:
+ headers.append((b'cookie', cookie.encode()))
+ headers.extend(extra_headers)
+ if token is not None:
+ headers.append((b'authorization', f'Bearer {token}'.encode()))
+ if authorization is not None:
+ headers.append((b'authorization', authorization))
+ messages = await call_app(notebook.app, {
+ 'type': 'http', 'path': unquote(url.path), 'method': method,
+ 'query_string': url.query.encode(), 'headers': headers,
+ 'scheme': scheme, 'client': (client, 12345),
+ }, receive)
+ self.assertEqual(len(messages), 2)
+ self.assertEqual(messages[0]['type'], 'http.response.start')
+ self.assertEqual(messages[1]['type'], 'http.response.body')
+ response_headers = dict(messages[0]['headers'])
+ body = messages[1]['body']
+ if body and response_headers[b'content-type'] == b'application/json':
+ body = orjson.loads(body)
+ return messages[0]['status'], body, response_headers
+
+ async def account(self, username='alice'):
+ credentials = {'username': username, 'password': 'a long example passphrase'}
+ status, user, _ = await self.request('/api/signup', 'POST', credentials)
+ self.assertEqual((status, user), (201, {'id': username.strip().casefold()}))
+ status, login, headers = await self.request('/api/login', 'POST', credentials)
+ self.assertEqual(status, 200)
+ self.assertEqual(headers[b'cache-control'], b'no-store')
+ self.assertEqual(login['token_type'], 'bearer')
+ self.assertEqual(login['expires_in'], 3600)
+ return login['access_token']
+
+ async def create(self, token, **data):
+ status, note, headers = await self.request('/api/notes', 'POST', data, token)
+ self.assertEqual(status, 201)
+ self.assertEqual(headers[b'location'], f"/api/notes/{note['id']}".encode())
+ self.assertEqual(headers[b'cache-control'], b'no-store')
+ return note
+
+ async def test_public_page_assets_health_and_redirect(self):
+ for path, content_type, marker in [
+ ('/', b'text/html', b'Wred notebook'),
+ ('/static/app.css', b'text/css', b'.workspace'),
+ ('/static/app.js', b'text/javascript', b'/api/notes'),
+ ]:
+ with self.subTest(path=path):
+ status, body, headers = await self.request(path)
+ self.assertEqual(status, 200)
+ self.assertEqual(headers[b'content-type'], content_type)
+ self.assertEqual(headers[b'cache-control'], b'no-cache')
+ self.assertIn(marker, body)
+ status, body, headers = await self.request('/ui')
+ self.assertEqual((status, body, headers[b'location']), (303, b'', b'/'))
+ self.assertEqual((await self.request('/api/health'))[:2], (200, {'status': 'ok'}))
+ self.assertEqual((await self.request('/static/missing.js'))[0], 404)
+
+ async def test_signup_login_and_independent_login_revocation(self):
+ token = await self.account(' Alice ')
+ status, profile, headers = await self.request('/api/me', token=token)
+ self.assertEqual((status, profile), (200, {'id': 'alice'}))
+ self.assertEqual(headers[b'cache-control'], b'no-store')
+ credentials = {'username': 'alice', 'password': 'a long example passphrase'}
+ self.assertEqual((await self.request('/api/signup', 'POST', credentials))[0], 409)
+ for data in [dict(credentials, username='unknown'), dict(credentials, password='an incorrect long passphrase')]:
+ self.assertEqual((await self.request('/api/login', 'POST', data))[:2],
+ (401, {'error': 'Invalid credentials'}))
+ _, login, _ = await self.request('/api/login', 'POST', credentials)
+ second_token = login['access_token']
+ self.assertNotEqual(token, second_token)
+ self.assertEqual((await self.request('/api/logout', 'POST', token=token))[:2],
+ (200, {'logged_out': True}))
+ self.assertEqual((await self.request('/api/me', token=token))[0], 401)
+ self.assertEqual((await self.request('/api/me', token=second_token))[:2],
+ (200, {'id': 'alice'}))
+ expired_time = auth.monotonic() + 3601
+ with patch.object(auth, 'monotonic', return_value=expired_time):
+ self.assertEqual((await self.request('/api/me', token=second_token))[0], 401)
+
+ async def test_invalid_credentials_and_bearer_headers(self):
+ for data in [[], {}, {'username': 1, 'password': 'a long passphrase'},
+ {'username': 'alice', 'password': False},
+ {'username': ' ', 'password': 'a long passphrase'},
+ {'username': 'alice', 'password': 'short'},
+ {'username': 'x' * 65, 'password': 'a long passphrase'},
+ {'username': 'alice', 'password': 'x' * 129}]:
+ with self.subTest(data=data):
+ self.assertEqual((await self.request('/api/signup', 'POST', data))[0], 400)
+ for header in [b'', b'Basic example', b'Bearer ', b'Bearer unknown']:
+ with self.subTest(header=header):
+ status, error, headers = await self.request('/api/notes', authorization=header)
+ self.assertEqual((status, error), (401, {'error': 'Unauthorized'}))
+ self.assertEqual(headers[b'www-authenticate'], b'Bearer')
+
+ async def test_create_read_patch_replace_delete_and_export(self):
+ token = await self.account()
+ note = await self.create(token, title=' First idea ', body='Some details', pinned=True)
+ self.assertEqual((note['title'], note['body'], note['pinned'], note['owner']),
+ ('First idea', 'Some details', True, 'alice'))
+ path = f"/api/notes/{note['id']}"
+ self.assertEqual((await self.request(path, token=token))[:2], (200, note))
+ status, edited, _ = await self.request(path, 'PATCH', {'body': 'Edited details'}, token)
+ self.assertEqual(status, 200)
+ self.assertEqual((edited['title'], edited['body'], edited['pinned']),
+ ('First idea', 'Edited details', True))
+ self.assertEqual(edited['created_at'], note['created_at'])
+ status, replaced, _ = await self.request(path, 'PUT', {'title': 'Replacement'}, token)
+ self.assertEqual(status, 200)
+ self.assertEqual((replaced['title'], replaced['body'], replaced['pinned']),
+ ('Replacement', '', False))
+ status, exported, headers = await self.request('/api/notes/export', token=token)
+ self.assertEqual((status, exported), (200, [replaced]))
+ self.assertEqual(headers[b'content-disposition'], b'attachment; filename="notes.json"')
+ self.assertEqual(headers[b'cache-control'], b'no-store')
+ status, body, _ = await self.request(path, 'DELETE', token=token)
+ self.assertEqual((status, body), (204, b''))
+ self.assertEqual((await self.request(path, token=token))[0], 404)
+ self.assertEqual((await self.request('/api/notes/export', token=token))[:2], (200, []))
+
+ async def test_accounts_cannot_read_modify_delete_or_export_each_others_notes(self):
+ alice = await self.account('alice')
+ bob = await self.account('bob')
+ note = await self.create(alice, title='Private', body='Only Alice can read this')
+ bob_note = await self.create(bob, title='Bob only')
+ path = f"/api/notes/{note['id']}"
+ for method, data in [('GET', None), ('PATCH', {'title': 'Stolen'}),
+ ('PUT', {'title': 'Stolen'}), ('DELETE', None)]:
+ with self.subTest(method=method):
+ self.assertEqual((await self.request(path, method, data, bob))[0], 403)
+ self.assertEqual((await self.request(path, token=alice))[:2], (200, note))
+ _, page, _ = await self.request('/api/notes', token=bob)
+ self.assertEqual((page['items'], page['total']), ([bob_note], 1))
+ self.assertEqual((await self.request('/api/notes/export', token=bob))[:2], (200, [bob_note]))
+ for method in ['GET', 'PATCH', 'DELETE']:
+ self.assertEqual((await self.request('/api/notes/missing', method, {'title': 'Missing'}, bob))[0], 404)
+
+ async def test_filtering_pagination_and_invalid_queries(self):
+ token = await self.account()
+ first = await self.create(token, title='Coffee idea', body='Learn C++', pinned=True)
+ second = await self.create(token, title='Second idea', body='Coffee later')
+ await self.create(token, title='A separate thought')
+ _, page, _ = await self.request('/api/notes?q=COFFEE&limit=1&offset=0', token=token)
+ self.assertEqual((page['items'], page['total'], page['limit'], page['offset']), ([first], 2, 1, 0))
+ _, page, _ = await self.request('/api/notes?q=coffee&limit=1&offset=1', token=token)
+ self.assertEqual(page['items'], [second])
+ _, page, _ = await self.request('/api/notes?q=C%2B%2B&pinned=true', token=token)
+ self.assertEqual(page['items'], [first])
+ _, page, _ = await self.request('/api/notes?pinned=false', token=token)
+ self.assertEqual(page['total'], 2)
+ for query in ['limit=abc', 'offset=abc', 'limit=0', 'limit=51', 'offset=-1',
+ 'pinned=maybe', '&'.join(f'x={i}' for i in range(21))]:
+ with self.subTest(query=query):
+ self.assertEqual((await self.request('/api/notes?' + query, token=token))[0], 400)
+
+ async def test_note_validation_prevents_invalid_or_unowned_field_updates(self):
+ token = await self.account()
+ for data in [[], {}, {'body': 'No title'}, {'title': 7}, {'title': ' '},
+ {'title': 'x' * 101}, {'title': 'Valid', 'body': 3},
+ {'title': 'Valid', 'body': 'x' * 10_001},
+ {'title': 'Valid', 'pinned': 'true'}, {'title': 'Valid', 'owner': 'bob'}]:
+ with self.subTest(data=data):
+ self.assertEqual((await self.request('/api/notes', 'POST', data, token))[0], 400)
+ note = await self.create(token, title='Keep me')
+ path = f"/api/notes/{note['id']}"
+ for data in [{}, {'owner': 'bob'}, {'title': ''}, {'pinned': 1}]:
+ self.assertEqual((await self.request(path, 'PATCH', data, token))[0], 400)
+ self.assertEqual((await self.request(path, token=token))[:2], (200, note))
+ self.assertEqual((await self.request(path, 'PUT', {'body': 'Missing title'}, token))[0], 400)
+ status, pinned, _ = await self.request(path, 'PATCH', {'pinned': True}, token)
+ self.assertEqual((status, pinned['pinned'], pinned['title']), (200, True, 'Keep me'))
+
+ async def test_cookie_restores_login_and_logout_revokes_the_whole_login(self):
+ credentials = {'username': 'alice', 'password': 'a long example passphrase'}
+ self.assertEqual((await self.request('/api/signup', 'POST', credentials))[0], 201)
+ status, login, headers = await self.request('/api/login', 'POST', credentials, scheme='https')
+ self.assertEqual(status, 200)
+ cookies = SimpleCookie(headers[b'set-cookie'].decode())
+ morsel = cookies[auth.refresh_cookie_name]
+ self.assertTrue(morsel['httponly'])
+ self.assertTrue(morsel['secure'])
+ self.assertEqual(morsel['samesite'], 'Strict')
+ self.assertEqual(morsel['path'], '/api')
+ self.assertEqual(int(morsel['max-age']), 7 * 24 * 3600)
+ self.assertEqual(morsel['domain'], '')
+ cookie = f'{auth.refresh_cookie_name}={morsel.value}'
+ status, restored, headers = await self.request('/api/refresh', 'POST', cookie=cookie)
+ self.assertEqual(status, 200)
+ self.assertNotEqual(restored['access_token'], login['access_token'])
+ self.assertEqual(restored['user'], {'id': 'alice'})
+ self.assertEqual(headers[b'cache-control'], b'no-store')
+ self.assertEqual((await self.request('/api/me', token=restored['access_token']))[:2],
+ (200, {'id': 'alice'}))
+ # Cookies alone must never authorize notes endpoints, keeping bearer auth separate.
+ self.assertEqual((await self.request('/api/notes', cookie=cookie))[0], 401)
+ _, _, headers = await self.request('/api/logout', 'POST', token=restored['access_token'])
+ cleared = SimpleCookie(headers[b'set-cookie'].decode())[auth.refresh_cookie_name]
+ self.assertEqual(cleared['max-age'], '0')
+ self.assertEqual(cleared['path'], '/api')
+ for access in [login['access_token'], restored['access_token']]:
+ self.assertEqual((await self.request('/api/me', token=access))[0], 401)
+ self.assertEqual((await self.request('/api/refresh', 'POST', cookie=cookie))[0], 401)
+
+ async def test_refresh_rejects_cross_origin_requests_without_revoking_valid_login(self):
+ credentials = {'username': 'alice', 'password': 'a long example passphrase'}
+ await self.request('/api/signup', 'POST', credentials)
+ _, _, headers = await self.request('/api/login', 'POST', credentials)
+ cookie = SimpleCookie(headers[b'set-cookie'].decode())[auth.refresh_cookie_name]
+ cookie = f'{auth.refresh_cookie_name}={cookie.value}'
+ for options in [
+ {'csrf_header': False},
+ {'extra_headers': [(b'origin', b'https://attacker.test')]},
+ {'extra_headers': [(b'origin', b'null')]},
+ {'extra_headers': [(b'sec-fetch-site', b'cross-site')]},
+ {'extra_headers': [(b'sec-fetch-site', b'same-site')]},
+ ]:
+ with self.subTest(options=options):
+ self.assertEqual((await self.request('/api/refresh', 'POST', cookie=cookie, **options))[0], 403)
+ self.assertEqual((await self.request('/api/refresh', 'POST', cookie=cookie,
+ extra_headers=[(b'origin', b'http://notebook.test')]))[0], 200)
+ self.assertEqual((await self.request('/api/login', 'POST', credentials, csrf_header=False))[0], 403)
+
+ async def test_expired_and_invalid_refresh_cookies_clear_browser_state(self):
+ credentials = {'username': 'alice', 'password': 'a long example passphrase'}
+ await self.request('/api/signup', 'POST', credentials)
+ _, login, headers = await self.request('/api/login', 'POST', credentials)
+ morsel = SimpleCookie(headers[b'set-cookie'].decode())[auth.refresh_cookie_name]
+ self.assertFalse(morsel['secure'], 'Local HTTP demo cookies should work without TLS')
+ cookie = f'{auth.refresh_cookie_name}={morsel.value}'
+ # Access expiry is recoverable while the independent refresh credential is valid.
+ now = auth.monotonic()
+ with patch.object(auth, 'monotonic', return_value=now + 3601):
+ self.assertEqual((await self.request('/api/me', token=login['access_token']))[0], 401)
+ self.assertEqual((await self.request('/api/refresh', 'POST', cookie=cookie))[0], 200)
+ with patch.object(auth, 'monotonic', return_value=now + 7 * 24 * 3600 + 1):
+ status, _, headers = await self.request('/api/refresh', 'POST', cookie=cookie)
+ self.assertEqual(status, 401)
+ self.assertEqual(SimpleCookie(headers[b'set-cookie'].decode())[auth.refresh_cookie_name]['max-age'], '0')
+ for invalid in [None, 'notebook_refresh=unknown', 'malformed; cookie']:
+ self.assertEqual((await self.request('/api/refresh', 'POST', cookie=invalid))[0], 401)
+
+ async def test_rate_limits_slide_and_are_isolated_by_client_and_endpoint(self):
+ with patch.object(rate_limit, 'monotonic', return_value=100):
+ self.assertEqual((await self.request('/api/rate-limit-demo'))[0], 200)
+ with patch.object(rate_limit, 'monotonic', return_value=105):
+ self.assertEqual((await self.request('/api/rate-limit-demo'))[0], 200)
+ with patch.object(rate_limit, 'monotonic', return_value=106):
+ self.assertEqual((await self.request('/api/rate-limit-demo'))[0], 200)
+ with patch.object(rate_limit, 'monotonic', return_value=109):
+ status, error, headers = await self.request('/api/rate-limit-demo')
+ self.assertEqual(status, 429)
+ self.assertEqual(error['retry_after'], 1)
+ self.assertEqual(headers[b'retry-after'], b'1')
+ self.assertEqual(headers[b'cache-control'], b'no-store')
+ self.assertEqual((await self.request('/api/rate-limit-demo', client='127.0.0.2'))[0], 200)
+ self.assertEqual((await self.request('/api/health'))[0], 200)
+ self.assertEqual((await self.request('/api/rate-limit-demo',
+ extra_headers=[(b'x-forwarded-for', b'127.0.0.3')]))[0], 429)
+ with patch.object(rate_limit, 'monotonic', return_value=110):
+ self.assertEqual((await self.request('/api/rate-limit-demo'))[0], 200)
+ with patch.object(rate_limit, 'monotonic', return_value=200):
+ self.assertEqual((await self.request('/api/rate-limit-demo'))[0], 200)
+
+ async def test_concurrent_login_attempts_cannot_bypass_rate_limit(self):
+ credentials = {'username': 'unknown', 'password': 'a long example passphrase'}
+ with patch.object(rate_limit, 'monotonic', return_value=100):
+ responses = await asyncio.gather(*[
+ self.request('/api/login', 'POST', credentials) for _ in range(6)
+ ])
+ self.assertEqual(sorted(status for status, _, _ in responses), [401] * 5 + [429])
+ limited = next(response for response in responses if response[0] == 429)
+ self.assertEqual(limited[2][b'retry-after'], b'60')
+ with patch.object(rate_limit, 'monotonic', return_value=110):
+ self.assertEqual((await self.request('/api/login', 'POST', credentials))[2][b'retry-after'], b'50')
+ with patch.object(rate_limit, 'monotonic', return_value=160):
+ self.assertEqual((await self.request('/api/login', 'POST', credentials))[0], 401)
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/tests/test_wred.py b/tests/test_wred.py
new file mode 100644
index 0000000..399ca14
--- /dev/null
+++ b/tests/test_wred.py
@@ -0,0 +1,285 @@
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+import orjson
+
+from wred import App
+
+
+async def call_app(app, scope, receive):
+ """Drive an ASGI request and capture its actual outgoing messages."""
+ messages = []
+
+ async def send(message):
+ messages.append(message)
+
+ await app(scope, receive, send)
+ return messages
+
+
+class WredTests(unittest.IsolatedAsyncioTestCase):
+ async def test_file_delivery_and_root_confinement(self):
+ # Exercise the public helper through the actual ASGI response boundary.
+ # This guards MIME/status delivery and accidental exposure outside root.
+ with tempfile.TemporaryDirectory() as directory:
+ base = Path(directory)
+ root = base / 'static'
+ root.mkdir()
+ (root / 'index.html').write_bytes(b'<p>Hello</p>')
+ (root / 'asset.unknownextension').write_bytes(b'\x00\xff')
+ (base / 'secret.txt').write_text('private')
+ (root / 'escape.txt').symlink_to(base / 'secret.txt')
+
+ app = App()
+
+ @app.route('/file', methods=['POST'])
+ async def file(request):
+ data = await request.json()
+ return await app.static(data['filename'], root=root)
+
+ cases = [
+ ('index.html', 200, b'text/html', b'<p>Hello</p>'),
+ ('asset.unknownextension', 200, b'application/octet-stream', b'\x00\xff'),
+ ('missing.txt', 404, b'text/plain; charset=utf-8', b'Not Found'),
+ ('.', 404, b'text/plain; charset=utf-8', b'Not Found'),
+ ('../secret.txt', 404, b'text/plain; charset=utf-8', b'Not Found'),
+ (str(base / 'secret.txt'), 404, b'text/plain; charset=utf-8', b'Not Found'),
+ ('escape.txt', 404, b'text/plain; charset=utf-8', b'Not Found'),
+ ('invalid\x00.txt', 404, b'text/plain; charset=utf-8', b'Not Found'),
+ ]
+ for filename, status, content_type, body in cases:
+ with self.subTest(filename=filename):
+ async def receive():
+ return {'type': 'http.request', 'body': orjson.dumps({'filename': filename})}
+
+ messages = await call_app(app, {'type': 'http', 'path': '/file', 'method': 'POST'}, receive)
+ self.assertEqual(messages, [
+ {'type': 'http.response.start', 'status': status,
+ 'headers': [(b'content-type', content_type)]},
+ {'type': 'http.response.body', 'body': body},
+ ])
+
+ async def test_dictionary_and_list_responses_are_json(self):
+ app = App()
+
+ @app.route('/')
+ async def home(request):
+ return value
+
+ async def receive():
+ raise AssertionError('GET handler should not read a body')
+
+ cases = [
+ ({'hello': 'world'}, b'{"hello":"world"}'),
+ ([{'id': 1}, {'id': 2}], b'[{"id":1},{"id":2}]'),
+ ([], b'[]'),
+ ]
+ for value, body in cases:
+ with self.subTest(value=value):
+ messages = await call_app(app, {'type': 'http', 'path': '/', 'method': 'GET'}, receive)
+ self.assertEqual(messages, [
+ {'type': 'http.response.start', 'status': 200,
+ 'headers': [(b'content-type', b'application/json')]},
+ {'type': 'http.response.body', 'body': body},
+ ])
+
+ async def test_custom_status_and_json_errors(self):
+ # Covers the error paths that previously sent ASGI messages separately,
+ # plus a response tuple returned through dynamic routing.
+ app = App()
+
+ @app.route('/{name}/created')
+ async def created(request, name):
+ return {'name': name}, 'application/json', 201
+
+ @app.route('/invalid')
+ async def invalid(request):
+ return 'not a supported response'
+
+ @app.route('/input', methods=['POST'])
+ async def input_data(request):
+ return await request.json()
+
+ cases = [
+ ('GET', '/alice/created', b'', 201, b'{"name":"alice"}'),
+ ('GET', '/missing', b'', 404, b'{"error":"Not Found"}'),
+ ('GET', '/invalid', b'', 500,
+ b'{"error":"Response must be a dictionary, list, bytes, or response tuple"}'),
+ ('POST', '/input', b'{', 400, b'{"error":"Invalid JSON body"}'),
+ ('POST', '/input', b'', 200, b'{}'),
+ ('GET', '/alice/missing', b'', 404, b'{"error":"Not Found"}'),
+ ('DELETE', '/alice/created', b'', 404, b'{"error":"Not Found"}'),
+ ]
+ for method, path, request_body, status, body in cases:
+ with self.subTest(path=path):
+ async def receive():
+ return {'type': 'http.request', 'body': request_body}
+
+ messages = await call_app(app, {'type': 'http', 'path': path, 'method': method}, receive)
+ self.assertEqual(messages, [
+ {'type': 'http.response.start', 'status': status,
+ 'headers': [(b'content-type', b'application/json')]},
+ {'type': 'http.response.body', 'body': body},
+ ])
+
+ async def test_route_priority_and_method_registration(self):
+ app = App()
+
+ @app.route('/people/{name}')
+ async def person(request, name):
+ return {'name': name}
+
+ @app.route('/people/{name}', methods=['POST'])
+ async def update_person(request, name):
+ return {'updated': name}
+
+ @app.route('/people/alice')
+ async def alice(request):
+ return {'fixed': True}
+
+ cases = [
+ ('GET', '/people/alice', b'{"fixed":true}'),
+ ('GET', '/people/bob', b'{"name":"bob"}'),
+ ('POST', '/people/alice', b'{"updated":"alice"}'),
+ ]
+ for method, path, body in cases:
+ with self.subTest(method=method, path=path):
+ async def receive():
+ raise AssertionError('These handlers do not read the body')
+
+ messages = await call_app(app, {'type': 'http', 'path': path, 'method': method}, receive)
+ self.assertEqual(messages[0]['status'], 200)
+ self.assertEqual(messages[1]['body'], body)
+
+ async def test_streamed_body_is_cached_and_headers_are_available(self):
+ app = App()
+ reads = 0
+ chunks = iter([
+ {'type': 'http.request', 'body': b'{"name":', 'more_body': True},
+ {'type': 'http.request', 'more_body': True},
+ {'type': 'http.request', 'body': b'"Alice"}', 'more_body': False},
+ ])
+
+ @app.route('/stream', methods=['POST'])
+ async def stream(request):
+ first = await request.body()
+ parsed = await request.json()
+ cached = await request.body()
+ return {'raw': first.decode(), 'cached': cached.decode(),
+ 'parsed': parsed, 'header': request.headers['x-example']}
+
+ async def receive():
+ nonlocal reads
+ reads += 1
+ return next(chunks)
+
+ messages = await call_app(app, {
+ 'type': 'http', 'path': '/stream', 'method': 'POST',
+ 'headers': [(b'x-example', b'hello')],
+ }, receive)
+ self.assertEqual(reads, 3)
+ self.assertEqual(messages[0]['status'], 200)
+ self.assertEqual(orjson.loads(messages[1]['body']), {
+ 'raw': '{"name":"Alice"}', 'cached': '{"name":"Alice"}',
+ 'parsed': {'name': 'Alice'}, 'header': 'hello',
+ })
+
+ async def test_disconnect_stops_body_reading(self):
+ app = App()
+ reads = 0
+
+ @app.route('/input', methods=['POST'])
+ async def input_data(request):
+ return await request.json()
+
+ async def receive():
+ nonlocal reads
+ reads += 1
+ if reads > 1:
+ raise AssertionError('Body reading continued after disconnect')
+ return {'type': 'http.disconnect'}
+
+ messages = await call_app(app, {'type': 'http', 'path': '/input', 'method': 'POST'}, receive)
+ self.assertEqual(messages, [
+ {'type': 'http.response.start', 'status': 400,
+ 'headers': [(b'content-type', b'application/json')]},
+ {'type': 'http.response.body', 'body': b'{"error":"Client disconnected"}'},
+ ])
+
+ async def test_unsupported_scope_does_not_read_http_fields(self):
+ app = App()
+
+ async def receive():
+ raise AssertionError('An unsupported scope should not be read')
+
+ messages = await call_app(app, {'type': 'unsupported'}, receive)
+ self.assertEqual(messages, [
+ {'type': 'http.response.start', 'status': 400,
+ 'headers': [(b'content-type', b'application/json')]},
+ {'type': 'http.response.body', 'body': b'{"error":"Unsupported scope type"}'},
+ ])
+
+ async def test_file_read_failure_returns_404(self):
+ app = App()
+ with tempfile.TemporaryDirectory() as directory:
+ (Path(directory) / 'private.txt').write_text('private')
+
+ @app.route('/')
+ async def file(request):
+ return await app.static('private.txt', root=directory)
+
+ async def receive():
+ raise AssertionError('File serving should not read the request body')
+
+ # Simulate the OS refusing a real file read, including when tests run as root.
+ with patch.object(Path, 'read_bytes', side_effect=PermissionError('denied')):
+ messages = await call_app(app, {'type': 'http', 'path': '/', 'method': 'GET'}, receive)
+ self.assertEqual(messages, [
+ {'type': 'http.response.start', 'status': 404,
+ 'headers': [(b'content-type', b'text/plain; charset=utf-8')]},
+ {'type': 'http.response.body', 'body': b'Not Found'},
+ ])
+
+ async def test_raw_and_custom_response_bodies(self):
+ app = App()
+
+ @app.route('/')
+ async def raw(request):
+ return response
+
+ cases = [
+ (b'\x00\xff', 200, b'application/octet-stream', b'\x00\xff', []),
+ ((b'<p>Hello</p>', 'text/html; charset=utf-8', 202),
+ 202, b'text/html; charset=utf-8', b'<p>Hello</p>', []),
+ (([], 'application/json', 201), 201, b'application/json', b'[]', []),
+ ((b'bad', 'text/plain'), 500, b'application/json',
+ b'{"error":"Response must be a dictionary, list, bytes, or response tuple"}', []),
+ (({'error': 'Unauthorized'}, 'application/json', 401,
+ [('WWW-Authenticate', 'Bearer'), ('Cache-Control', 'no-store')]),
+ 401, b'application/json', b'{"error":"Unauthorized"}',
+ [(b'www-authenticate', b'Bearer'), (b'cache-control', b'no-store')]),
+ ((b'', 'text/plain', 303,
+ [('Location', '/ui'), ('X-Label', 'caf\u00e9'),
+ ('Set-Cookie', 'a=1; Secure'), ('Set-Cookie', 'b=2; Secure')]),
+ 303, b'text/plain', b'',
+ [(b'location', b'/ui'), (b'x-label', b'caf\xe9'),
+ (b'set-cookie', b'a=1; Secure'), (b'set-cookie', b'b=2; Secure')]),
+ (([], 'application/json', 200, []), 200, b'application/json', b'[]', []),
+ ]
+ for response, status, content_type, body, headers in cases:
+ with self.subTest(response=response):
+ async def receive():
+ raise AssertionError('These responses do not use a request body')
+
+ messages = await call_app(app, {'type': 'http', 'path': '/', 'method': 'GET'}, receive)
+ self.assertEqual(messages, [
+ {'type': 'http.response.start', 'status': status,
+ 'headers': [(b'content-type', content_type)] + headers},
+ {'type': 'http.response.body', 'body': body},
+ ])
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/wred.py b/wred.py
new file mode 100644
index 0000000..0465300
--- /dev/null
+++ b/wred.py
@@ -0,0 +1,166 @@
+"""
+Wred: An ultra micro ASGI API framework.
+
+Wred is designed for lightweight JSON APIs, offering a minimalistic way to build
+ASGI-compatible web services. It prioritizes simplicity and performance over all.
+
+
+Copyright (c) 2026, Harrison Erd.
+"""
+
+__author__ = 'Harrison Erd'
+__email__ = '[email protected]'
+__version__ = '0.01'
+__license__ = 'BSD3'
+
+from collections.abc import Awaitable, Callable
+from asyncio import to_thread
+from mimetypes import guess_type
+from pathlib import Path
+from typing import Any
+import orjson
+
+
+class WredResponseError(Exception):
+ """Raised when a handler returns a non-JSON response."""
+ pass
+
+class WredRequestError(Exception):
+ """Raised when a request cannot be processed (e.g., invalid body)."""
+ pass
+
+
+class Request:
+ """A request object encapsulating ASGI scope, receive, and send."""
+ def __init__(self, scope: dict[str, Any], receive: Callable[[], Awaitable[dict[str, Any]]], send: Callable[[dict[str, Any]], Awaitable[None]]):
+ self.scope = scope
+ self.receive = receive
+ self.send = send
+ self.path = scope['path']
+ self.method = scope['method']
+ self.headers = {k.decode(): v.decode() for k, v in scope.get('headers', [])}
+ self._body = None
+
+ async def body(self) -> bytes:
+ """Read the request body from the ASGI receive stream."""
+ if self._body is None:
+ self._body = b""
+ while True:
+ message = await self.receive()
+ if message['type'] != 'http.request':
+ raise WredRequestError("Client disconnected")
+ self._body += message.get('body', b'')
+ if not message.get('more_body', False):
+ break
+ return self._body
+
+ async def json(self) -> dict[str, Any]:
+ """Parse the request body as JSON."""
+ try:
+ body = await self.body()
+ return orjson.loads(body) if body else {}
+ except orjson.JSONDecodeError:
+ raise WredRequestError("Invalid JSON body")
+
+
+class App:
+ """A minimal ASGI application framework."""
+ def __init__(self):
+ self.routes: dict[str, dict[str, Callable]] = {} # Path -> Method -> Handler
+ self.pattern_routes: dict[str, dict[str, Callable]] = {}
+
+ async def __call__(self, scope: dict[str, Any], receive: Callable[[], Awaitable[dict[str, Any]]], send: Callable[[dict[str, Any]], Awaitable[None]]) -> None:
+ """Handle ASGI requests."""
+ if scope['type'] != 'http':
+ await self.send_response(send, {"error": "Unsupported scope type"}, status=400)
+ return
+
+ request = Request(scope, receive, send)
+ await self.handle_http(request)
+
+ async def handle_http(self, request: Request) -> None:
+ """Handle HTTP requests with routing and JSON response support."""
+ path = request.path
+ method = request.method
+ try:
+ # Check exact matches first
+ if path in self.routes and method in self.routes[path]:
+ handler = self.routes[path][method]
+ response = await handler(request)
+ await self.send_response(request.send, response)
+ return
+
+ # Check dynamic routes
+ for pattern, method_handlers in self.pattern_routes.items():
+ params = self.match_path(pattern, path)
+ if params is not None and method in method_handlers:
+ handler = method_handlers[method]
+ response = await handler(request, **params)
+ await self.send_response(request.send, response)
+ return
+
+ # No match found
+ await self.send_response(request.send, {"error": "Not Found"}, status=404)
+
+ except (WredResponseError, WredRequestError) as e:
+ status = 400 if isinstance(e, WredRequestError) else 500
+ await self.send_response(request.send, {"error": str(e)}, status=status)
+
+ async def send_response(self, send: Callable[[dict[str, Any]], Awaitable[None]], response: Any, status: int = 200) -> None:
+ """Send JSON or (body, content_type, status[, headers]) with string header pairs."""
+ body, content_type = response, 'application/octet-stream'
+ headers = []
+ if isinstance(response, tuple) and len(response) in (3, 4):
+ body, content_type, status = response[:3]
+ if len(response) == 4:
+ headers = [(name.lower().encode('ascii'), value.encode('latin-1')) for name, value in response[3]]
+ if isinstance(body, (dict, list)):
+ body, content_type = orjson.dumps(body), 'application/json'
+ elif not isinstance(body, bytes):
+ raise WredResponseError("Response must be a dictionary, list, bytes, or response tuple")
+ await send({
+ 'type': 'http.response.start',
+ 'status': status,
+ 'headers': [(b'content-type', content_type.encode('ascii'))] + headers,
+ })
+ await send({'type': 'http.response.body', 'body': body})
+
+ def match_path(self, pattern: str, path: str) -> dict[str, str] | None:
+ """Match a path against a pattern and extract parameters."""
+ pattern_parts = pattern.strip('/').split('/')
+ path_parts = path.strip('/').split('/')
+ if len(pattern_parts) != len(path_parts):
+ return None
+
+ params = {}
+ for pattern_part, path_part in zip(pattern_parts, path_parts):
+ if pattern_part.startswith('{') and pattern_part.endswith('}'):
+ param_name = pattern_part[1:-1]
+ params[param_name] = path_part
+ elif pattern_part != path_part:
+ return None
+ return params
+
+ async def static(self, filename: str | Path, *, root: str | Path) -> tuple[bytes, str, int]:
+ """Serve a small file inside root; read its entire body off the event loop."""
+ def read() -> tuple[bytes, str, int]:
+ try:
+ directory = Path(root).resolve()
+ path = (directory / filename).resolve()
+ if not path.is_relative_to(directory) or not path.is_file():
+ return b'Not Found', 'text/plain; charset=utf-8', 404
+ body = path.read_bytes()
+ except (OSError, ValueError, RuntimeError):
+ return b'Not Found', 'text/plain; charset=utf-8', 404
+ return body, guess_type(path.name)[0] or 'application/octet-stream', 200
+
+ return await to_thread(read)
+
+ def route(self, path: str, methods: list[str] = ['GET']) -> Callable:
+ """Decorator to register routes with specific HTTP methods."""
+ def decorator(handler: Callable) -> Callable:
+ routes = self.pattern_routes if '{' in path and '}' in path else self.routes
+ routes.setdefault(path, {}).update({m.upper(): handler for m in methods})
+ return handler
+ return decorator
+