patx/wred
Release v0.03 with query helper and clearer examples and docs
Commit 54cdc98 · patx · 2026-10-13T09:41:17-04:00
Comments
No comments yet.
Diff
diff --git a/README.md b/README.md
index 3a0495b..bb67aaa 100644
--- a/README.md
+++ b/README.md
@@ -1,44 +1,41 @@
# Wred
-Wred handles HTTP routing, requests, and JSON responses. Need CORS, sessions,
-authentication, or compression? Wrap it with existing ASGI middleware.
-No need to reinvent what ASGI already makes possible.
+A tiny Python/ASGI backend for HTML and JavaScript apps. One file. JSON from
+`orjson`, server from Granian. Pronounced "red".
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])`.
+- Async handlers, explicit HTTP methods, and fixed or parameterized routes.
+- Headers, query parameters, cached body reads, and JSON parsing.
+- JSON responses, custom response tuples, and a small-file helper.
+- Callable ASGI response support and routing relative to mounted paths.
## 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.
-
-Your application chooses those pieces. Use existing libraries without making
-them everyone else's dependency. Wred doesn't need to know they exist.
+or built-in streaming.
-## Use it
+Use existing ASGI middleware for CORS, authentication, sessions, or compression.
+[Starlette's middleware](https://starlette.dev/middleware/#using-middleware-in-other-frameworks)
+can wrap Wred directly. Your app chooses its dependencies; Wred doesn't need to
+know they exist.
-Python 3.11+. Install from PyPI:
+## Install and run
-Procure your goods, yeah, use `uv` not `pip`, its 2026:
+Python 3.11+. For a new project:
```bash
-$ uv init my-app
-$ cd my-app
-$ uv add wred
+uv init my-app
+cd my-app
+uv add wred
```
-Already have a uv project? Just run `uv add wred`. Granian and orjson come with it.
+In an existing uv project, just `uv add wred`. Granian and orjson are included.
-Put this in your pipe (`app.py`):
+Save as `app.py`:
```python
from wred import App
@@ -46,7 +43,7 @@ from wred import App
app = App()
@app.route('/')
-async def say_hello(request):
+async def hello(request):
return {'hello': 'world'}
```
@@ -56,21 +53,15 @@ And smoke it:
uv run granian --interface asginl app:app
```
-`asginl` disables lifespan handling. Wred is HTTP-only and rejects lifespan,
-WebSocket, and other unsupported scopes with `RuntimeError`, without sending
-HTTP messages. This follows the [ASGI protocol guidance](https://asgi.readthedocs.io/en/latest/specs/main.html#protocol-specifications).
-
-The server can run a middleware wrapper around Wred as its `app` too.
-[Starlette's middleware](https://starlette.dev/middleware/#using-middleware-in-other-frameworks)
-works with other ASGI applications: CORS, GZip, TrustedHost, HTTPS redirects,
-and cookie sessions are available without changing Wred. Install only what
-your app needs; Starlette isn't a Wred dependency. Choose standard ASGI wrappers
-rather than integrations that require another framework's application internals.
+Open <http://127.0.0.1:8000>. Wred is HTTP-only; `asginl` disables lifespan.
+Unsupported protocols raise `RuntimeError` without sending HTTP messages,
+following [ASGI guidance](https://asgi.readthedocs.io/en/latest/specs/main.html#protocol-specifications).
+You can also export a middleware wrapper as `app`.
## Routes
-Handlers are async. The first argument is the request. `methods` defaults to
-`['GET']`; path parameters arrive as strings, passed by keyword.
+Handlers are async and receive a request. Methods default to `['GET']`;
+path parameters are strings passed by keyword.
```python
@app.route('/users/{user_id}')
@@ -86,96 +77,64 @@ 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.
+Different methods can have separate handlers on the same path. Fixed routes
+win over dynamic routes for that method; dynamic routes follow 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.
-Mounted apps resolve routes relative to the ASGI `root_path`: a Wred route
-`/users` mounted at `/api` handles `/api/users`. `request.path` and the original
-ASGI scope keep the full incoming path for middleware and response objects.
+Mounting at `/api` makes a `/users` route handle `/api/users`. Routing respects
+`root_path`; `request.path` and the original scope keep the full incoming path.
## Requests
-| API | What you get |
+| API | Returns |
| --- | --- |
-| `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. |
+| `request.path` | Path without the query string. |
+| `request.method` | HTTP method. |
+| `request.headers` | Decoded header dict; use lowercase keys. |
+| `request.scope` | Original ASGI scope, including client, scheme, and middleware state. |
+| `request.query()` | Cached dict of string keys and values. No `await`. |
+| `await request.body()` | Cached raw bytes. |
| `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}
-
@app.route('/search')
async def search(request):
- query = parse_qs(request.scope.get('query_string', b'').decode('utf-8'))
- return {'q': query.get('q', [''])[0]}
+ return {'q': request.query().get('q', '')}
```
-Read a bearer credential with `request.headers.get('authorization', '')`.
-Check it in your application or let authentication middleware populate the ASGI
-scope, then enforce access in your handlers. Middleware can also add session
-data: Starlette's SessionMiddleware exposes it through `request.scope['session']`,
-not a Wred-specific session property. Its cookies are signed, not encrypted.
+Query parameters preserve blank values; repeated keys keep the last value.
+JSON can be an object, array, or scalar. Validate inputs and convert types yourself.
-## Responses
+Authentication middleware can populate the scope for handlers to check.
+Starlette's SessionMiddleware uses `request.scope['session']`; its cookies are
+signed, not encrypted. Wred doesn't supply a session property.
-Return a dict or list for JSON with HTTP 200. Return bytes for
-`application/octet-stream`. No `json()` helper. No ceremony.
-
-Handlers can also return callable ASGI response objects. Wred awaits them with
-`(scope, receive, send)`, letting the response handle its own headers, status,
-and body. Response classes from ASGI libraries can work directly; Wred doesn't
-need a dependency or adapter for them.
-
-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}]
+## Responses
[email protected]('/items', methods=['POST'])
-async def create_item(request):
- return {'id': 3}, 'application/json', 201, [('Location', '/items/3')]
+Dicts and lists become JSON with HTTP 200. Bytes default to
+`application/octet-stream`. Custom responses use
+`(body, content_type, status[, headers])`:
[email protected]('/old')
-async def redirect(request):
- return b'', 'text/plain', 303, [('Location', '/')]
+| Handler return | Response |
+| --- | --- |
+| `{'hello': 'world'}` | JSON object, 200. |
+| `[{'id': 1}, {'id': 2}]` | JSON array, 200. |
+| `b'hello', 'text/plain', 200` | Text response. |
+| `{'id': 3}, 'application/json', 201, [('Location', '/items/3')]` | Created resource with a custom header. |
+| `b'', 'text/plain', 303, [('Location', '/')]` | Redirect. |
+| `b'', 'application/json', 204` | No content. |
[email protected]('/items/{item_id}', methods=['DELETE'])
-async def delete_item(request, item_id):
- return b'', 'application/json', 204
-```
+Dicts and lists always use `application/json`, including inside tuples.
+Other tuple bodies must be bytes. Headers are string pairs; repeated
+`Set-Cookie` headers are supported.
-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.
-For behavior across every route, an ASGI wrapper can intercept requests and
-responses to apply CORS or compression. Keep that plumbing out of your handlers.
+Return a callable ASGI response object to let it manage status, headers, and
+body through `(scope, receive, send)`. Reuse external response classes for
+streaming, downloads, or SSE. Use ASGI middleware for behavior across routes.
## 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
@@ -190,22 +149,24 @@ 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.
+`static(filename, root=...)` returns `(bytes, content_type, status)`.
+The root is required and accepts a string or `Path`. MIME types are guessed;
+missing, unreadable, or outside-root files return plain-text 404, including
+symlink escapes. 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.
+Malformed JSON, invalid UTF-8 query strings, and `WredRequestError` produce
+JSON 400 responses: `{'error': message}`. Raise `WredRequestError` from a handler
+for application validation failures. Unsupported response types produce JSON
+500 errors; other exceptions propagate.
## Example app
-The [`example_app`](https://gitman.io/patx/wred/src/example_app) notebook demonstrates signup, login,
-refresh cookies, logout, protected endpoints, note CRUD, and rate limiting.
-Those features belong to the example, not the framework.
-Clone the repository to try it:
+The [notebook demo](https://gitman.io/patx/wred/src/example_app) includes signup,
+login, refresh cookies, logout, protected note CRUD, and rate limiting.
+Those features belong to the app, not Wred.
```bash
git clone https://gitman.io/git/patx/wred
@@ -214,12 +175,12 @@ uv sync
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.
+Open <http://127.0.0.1:8000>. Data lives in memory and disappears on restart.
## Benchmarks
-`GET /` returning `{"hello": "world"}`, measured on October 13, 2026.
-Three runs per framework, with the run order rotated each round.
+`GET /` returning `{"hello": "world"}` on October 13, 2026. Three runs per
+framework, rotating their order each round.
| Framework | Median requests/sec | Min | Max |
| --- | ---: | ---: | ---: |
@@ -228,10 +189,9 @@ Three runs per framework, with the run order rotated each round.
| MicroPie 0.32 | 207,720 | 197,520 | 216,010 |
Python 3.14.4, Granian 2.8.4, eight workers, HTTP/1 keep-alive, asyncio,
-and access logging disabled. `wrk -t4 -c256 -d20s --latency`, after a
-five-second warmup for each run. Client and server shared the same machine
-over loopback. All nine measured runs had zero socket errors or non-success responses.
+no access logging. `wrk -t4 -c256 -d20s --latency`, five-second warmups.
+Client and server shared a machine over loopback. All nine measured runs had
+zero socket errors or non-success responses.
-Wred delivered about 31% more requests/sec than MicroPie in this workload.
-It's a tiny JSON response, not your database, auth, or entire application.
-Benchmark your own shit.
+About 31% more requests/sec than MicroPie for this workload. It's a tiny JSON
+response, not your whole application. Benchmark your own shit.
diff --git a/example_app/app.py b/example_app/app.py
index aeb1b94..2f78837 100644
--- a/example_app/app.py
+++ b/example_app/app.py
@@ -1,57 +1,18 @@
-"""Wred Notebook: a complete, commented application built on Wred.
+"""Run the notebook demo from the repository root:
-From the repository root:
uv sync
- uv run granian --interface asginl --workers 1 examples.app:app
+ uv run granian --interface asginl --workers 1 example_app.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.
+Open http://127.0.0.1:8000. Users, tokens, and notes disappear on restart.
+Login is restored with an HttpOnly refresh cookie; API calls use bearer tokens.
+Storage and rate limits are process-local. Production needs shared storage and HTTPS.
"""
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
index 367c3fb..a8df158 100644
--- a/example_app/auth.py
+++ b/example_app/auth.py
@@ -1,4 +1,5 @@
"""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
@@ -10,274 +11,210 @@ 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.
+token_lifetime = 3600
+refresh_lifetime = 7 * 24 * 3600
+refresh_cookie_name = 'notebook_refresh'
-def same_origin_request(request): # Protect endpoints that issue or use browser cookies.
+def same_origin_request(request):
# 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.
+ expected = f'{request.scope.get("scheme", "http")}://{request.headers.get("host", "")}'
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.
+def refresh_cookie(request, value, max_age):
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.
+def read_refresh_cookie(request):
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.
+def issue_access_token(username, login_id):
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.
+ tokens[digest] = {
+ 'user_id': username,
+ 'login_id': login_id,
+ 'expires_at': monotonic() + token_lifetime,
+ }
+ return {
+ 'access_token': token,
+ 'token_type': 'bearer',
+ 'expires_in': token_lifetime,
+ 'user': {'id': username},
+ }
+
+
+def revoke_login(login_id):
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.
+async def read_credentials(request):
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.
+async def hash_password(password, salt):
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.
+async def validate_access_token(token):
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.
+def protected(handler):
@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.
+ async def wrapper(request, *args, **kwargs):
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.
+ return (
+ {'error': 'Unauthorized'},
+ 'application/json',
+ 401,
+ [('WWW-Authenticate', 'Bearer')],
+ )
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.
+ async def signup(request):
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.
+ async def login(request):
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.
+ return (
+ {'error': 'A same-origin request with X-Notebook-Request: 1 is required'},
+ 'application/json',
+ 403,
+ )
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)]
-
+ refresh_tokens[sha256(refresh.encode()).hexdigest()] = {
+ 'user_id': username,
+ 'login_id': login_id,
+ 'expires_at': monotonic() + refresh_lifetime,
+ }
+ 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.
+ async def refresh(request):
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.
+ return (
+ {'error': 'A same-origin request with X-Notebook-Request: 1 is required'},
+ 'application/json',
+ 403,
+ )
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)]
+ 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')]
+ 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.
+ async def me(request):
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.
+ async def logout(request):
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
index f21b2ad..1f8efaa 100644
--- a/example_app/notes.py
+++ b/example_app/notes.py
@@ -1,182 +1,127 @@
"""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.
+async def read_note(request, partial=False):
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.
+ if 'title' in data and (
+ not isinstance(data['title'], str) or not 1 <= len(data['title'].strip()) <= 100
+ ):
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.
+def owned_note(note_id, user):
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.
+ async def list_notes(request):
+ # Keep this demo's query budget at 20 fields, including repeated keys.
+ if request.scope.get('query_string', b'').count(b'&') >= 20:
raise WredRequestError('Invalid query string')
- # Treat omitted pagination values as a request for the first five notes.
+ query = request.query()
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.
+ limit = int(query.get('limit') or '5')
+ offset = int(query.get('offset') or '0')
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.
+ search = query.get('q', '').strip().casefold()
+ pinned = query.get('pinned', '')
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.
+ items = [
+ note for note in items if search in (note['title'] + '\n' + note['body']).casefold()
+ ]
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}
-
+ 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.
+ async def create_note(request):
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.
+ note = {
+ 'id': note_id,
+ 'owner': request.user['id'],
+ **data,
+ 'created_at': now,
+ 'updated_at': now,
+ }
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.
+ async def get_note(request, note_id):
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.
+ async def update_note(request, note_id):
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.
+ async def delete_note(request, note_id):
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.
+ async def export_notes(request):
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"')]
+ return (
+ items,
+ 'application/json',
+ 200,
+ [('Content-Disposition', 'attachment; filename="notes.json"')],
+ )
diff --git a/example_app/pages.py b/example_app/pages.py
index 5bdb70a..128b826 100644
--- a/example_app/pages.py
+++ b/example_app/pages.py
@@ -1,50 +1,34 @@
"""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).
+ async def home(request):
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().
+ async def assets(request, filename):
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.
+ async def old_ui(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.
+ async def health(request):
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.
+ async def rate_limit_demo(request):
return {'message': 'Request accepted'}
diff --git a/example_app/rate_limit.py b/example_app/rate_limit.py
index 68b74ce..22ecccb 100644
--- a/example_app/rate_limit.py
+++ b/example_app/rate_limit.py
@@ -1,4 +1,5 @@
"""Application-owned sliding-window rate limiting, independent of Wred."""
+
from collections import deque
from functools import wraps
from math import ceil
@@ -7,41 +8,33 @@ 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 rate_limit(requests, seconds):
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')]
+ 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/tests/test_wred.py b/tests/test_wred.py
index c345f03..f91c12f 100644
--- a/tests/test_wred.py
+++ b/tests/test_wred.py
@@ -237,6 +237,36 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
self.assertEqual(messages[0]['status'], 200)
self.assertEqual(messages[1]['body'], body)
+ async def test_query_parameters_decode_to_strings_without_reading_body(self):
+ app = App()
+
+ @app.route('/search')
+ async def search(request):
+ return {'params': request.query(), 'q': request.query().get('q', '')}
+
+ cases = [
+ (None, 200, {'params': {}, 'q': ''}),
+ (b'', 200, {'params': {}, 'q': ''}),
+ (b'q=coffee+ideas&limit=20', 200,
+ {'params': {'q': 'coffee ideas', 'limit': '20'}, 'q': 'coffee ideas'}),
+ (b'q=C%2B%2B&label=caf%C3%A9&x%26y=a%3Db', 200,
+ {'params': {'q': 'C++', 'label': 'café', 'x&y': 'a=b'}, 'q': 'C++'}),
+ (b'q=&bare&tag=first&tag=last', 200,
+ {'params': {'q': '', 'bare': '', 'tag': 'last'}, 'q': ''}),
+ (b'\xff', 400, {'error': 'Invalid query string'}),
+ ]
+ for query_string, status, body in cases:
+ with self.subTest(query_string=query_string):
+ async def receive():
+ raise AssertionError('Query parsing should not read the request body')
+
+ scope = {'type': 'http', 'path': '/search', 'method': 'GET'}
+ if query_string is not None:
+ scope['query_string'] = query_string
+ messages = await call_app(app, scope, receive)
+ self.assertEqual(messages[0]['status'], status)
+ self.assertEqual(orjson.loads(messages[1]['body']), body)
+
async def test_streamed_body_is_cached_and_headers_are_available(self):
app = App()
reads = 0
diff --git a/wred.py b/wred.py
index 1cadc09..eaac925 100644
--- a/wred.py
+++ b/wred.py
@@ -10,7 +10,7 @@ Copyright (c) 2026, Harrison Erd.
__author__ = 'Harrison Erd'
__email__ = '[email protected]'
-__version__ = '0.02'
+__version__ = '0.03'
__license__ = 'BSD-3-Clause'
from collections.abc import Awaitable, Callable
@@ -18,6 +18,7 @@ from asyncio import to_thread
from mimetypes import guess_type
from pathlib import Path
from typing import Any
+from urllib.parse import parse_qsl
import orjson
@@ -40,6 +41,16 @@ class Request:
self.method = scope['method']
self.headers = {k.decode(): v.decode() for k, v in scope.get('headers', [])}
self._body = None
+ self._query = None
+
+ def query(self) -> dict[str, str]:
+ """Parse and cache query parameters; repeated keys keep their last value."""
+ if self._query is None:
+ try:
+ self._query = dict(parse_qsl(self.scope.get('query_string', b'').decode('utf-8'), keep_blank_values=True))
+ except UnicodeDecodeError:
+ raise WredRequestError("Invalid query string")
+ return self._query
async def body(self) -> bytes:
"""Read the request body from the ASGI receive stream."""