patx/wred
Condense framework implementation while retaining type hints and docstrings
Commit ea52887 · patx · 2026-10-17T23:02:53-04:00
Comments
No comments yet.
Diff
diff --git a/README.md b/README.md
index 75f0f72..95d7b3c 100644
--- a/README.md
+++ b/README.md
@@ -166,6 +166,7 @@ Exceptions propagate to your ASGI server or middleware. Malformed JSON raises
`orjson.JSONDecodeError`, invalid raw UTF-8 query bytes raise `UnicodeDecodeError`,
client disconnects during body reads raise `ConnectionError`, and unsupported
response types raise `TypeError`. Wred does not turn exceptions into responses.
+Malformed response tuples can raise `ValueError` during unpacking.
Return response tuples for expected application errors. Catch parsing errors
in your handler when you want a JSON 400 response:
diff --git a/tests/test_wred.py b/tests/test_wred.py
index 095f822..4643a96 100644
--- a/tests/test_wred.py
+++ b/tests/test_wred.py
@@ -234,7 +234,7 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
('POST', '/input', b'{', b'', None, orjson.JSONDecodeError),
('GET', '/search', b'', b'\xff', None, UnicodeDecodeError),
('GET', '/invalid', b'', b'', 'unsupported response', TypeError),
- ('GET', '/invalid', b'', b'', (b'bad', 'text/plain'), TypeError),
+ ('GET', '/invalid', b'', b'', (b'bad', 'text/plain'), ValueError),
]
for method, path, body, query_string, response, error in cases:
with self.subTest(path=path, response=response):
diff --git a/wred.py b/wred.py
index 3b044e0..b6e03e8 100644
--- a/wred.py
+++ b/wred.py
@@ -1,18 +1,11 @@
"""
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]'
+__author__, __email__, __license__ = 'Harrison Erd', '[email protected]', 'BSD-3-Clause'
__version__ = '0.08'
-__license__ = 'BSD-3-Clause'
-
from collections.abc import Awaitable, Callable
from asyncio import to_thread
from mimetypes import guess_type
@@ -21,20 +14,22 @@ from typing import Any
from urllib.parse import parse_qsl
import orjson
-
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]]):
+ def __init__(self, scope: dict[str, Any], receive: Callable[[], Awaitable[dict[str, Any]]], send: Callable[[dict[str, Any]], Awaitable[None]]) -> None:
+ """Store the ASGI connection and decode request headers."""
self.scope, self.receive, self.send = scope, receive, send
self.path, self.method = scope['path'], scope['method']
- self.headers = {}
- for k, v in scope.get('headers', []): k = k.decode('latin-1').lower(); v = v.decode('latin-1'); self.headers[k] = self.headers[k] + '; ' + v if k == 'cookie' and k in self.headers else v
+ self.headers: dict[str, str] = {}
+ for k, v in scope.get('headers', []):
+ k, v = k.decode('latin-1').lower(), v.decode('latin-1')
+ self.headers[k] = self.headers[k] + '; ' + v if k == 'cookie' and k in self.headers else v
self._body = 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:
- self._query = dict(parse_qsl(self.scope.get('query_string', b'').decode('utf-8'), keep_blank_values=True))
+ self._query = dict(parse_qsl(self.scope.get('query_string', b'').decode(), keep_blank_values=True))
return self._query
async def body(self) -> bytes:
@@ -43,11 +38,9 @@ class Request:
chunks = []
while True:
message = await self.receive()
- if message['type'] != 'http.request':
- raise ConnectionError("Client disconnected")
+ if message['type'] != 'http.request': raise ConnectionError("Client disconnected")
chunks.append(message.get('body', b''))
- if not message.get('more_body', False):
- break
+ if not message.get('more_body'): break
self._body = b''.join(chunks)
return self._body
@@ -56,72 +49,56 @@ class Request:
body = await self.body()
return orjson.loads(body) if body else {}
-
class App:
"""A minimal ASGI application framework."""
- def __init__(self):
+ def __init__(self) -> None:
+ """Create fixed and pre-split dynamic route tables."""
self.routes: dict[str, dict[str, Callable]] = {}
self.pattern_routes: dict[tuple[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':
- raise RuntimeError(f"Unsupported ASGI scope type: {scope['type']}")
- request = Request(scope, receive, send)
- await self.handle_http(request)
+ if scope['type'] != 'http': raise RuntimeError(f"Unsupported ASGI scope type: {scope['type']}")
+ await self.handle_http(Request(scope, receive, send))
async def handle_http(self, request: Request) -> None:
"""Handle HTTP requests with routing and JSON response support."""
path = request.path
- root_path = request.scope.get('root_path', '').rstrip('/')
- if root_path and (path == root_path or path.startswith(root_path + '/')):
- path = path[len(root_path):] or '/'
+ root = request.scope.get('root_path', '').rstrip('/')
+ if root and (path == root or path.startswith(root + '/')): path = path[len(root):] or '/'
path = path.rstrip('/') or '/'
method = request.method
if path in self.routes and method in self.routes[path]:
- await self.dispatch_response(request, await self.routes[path][method](request))
- return
- path_parts = path.split('/')
- for pattern, method_handlers in self.pattern_routes.items():
- params = self.match_path(pattern, path_parts)
- if params is not None and method in method_handlers:
- await self.dispatch_response(request, await method_handlers[method](request, **params))
- return
- await self.send_response(request.send, {"error": "Not Found"}, status=404)
+ return await self.dispatch_response(request, await self.routes[path][method](request))
+ parts = path.split('/')
+ for pattern, handlers in self.pattern_routes.items():
+ if (params := self.match_path(pattern, parts)) is not None and method in handlers:
+ return await self.dispatch_response(request, await handlers[method](request, **params))
+ await self.send_response(request.send, {"error": "Not Found"}, 404)
async def dispatch_response(self, request: Request, response: Any) -> None:
"""Invoke an ASGI response or serialize an ordinary handler result."""
- if callable(response):
- await response(request.scope, request.receive, request.send)
- else:
- await self.send_response(request.send, response)
+ if callable(response): await response(request.scope, request.receive, request.send)
+ else: await self.send_response(request.send, response)
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, headers = response, 'application/octet-stream', []
+ if isinstance(response, tuple):
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 TypeError("Response must be a dictionary, list, bytes, or response tuple")
+ if len(response) == 4: headers = [(n.lower().encode('ascii'), v.encode('latin-1')) for n, v in response[3]]
+ if isinstance(body, (dict, list)): body, content_type = orjson.dumps(body), 'application/json'
+ elif not isinstance(body, bytes): raise TypeError("Response must be dict, list, bytes, or tuple")
await send({'type': 'http.response.start', 'status': status, 'headers': [(b'content-type', content_type.encode('latin-1'))] + headers})
await send({'type': 'http.response.body', 'body': body})
- def match_path(self, pattern_parts: tuple[str, ...], path_parts: list[str]) -> dict[str, str] | None:
+ def match_path(self, pattern: tuple[str, ...], parts: list[str]) -> dict[str, str] | None:
"""Match normalized paths and extract parameters."""
- if len(pattern_parts) != len(path_parts):
- return None
+ if len(pattern) != len(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
+ for p, part in zip(pattern, parts):
+ if p.startswith('{') and p.endswith('}'): params[p[1:-1]] = part
+ elif p != part: return None
return params
async def static(self, filename: str | Path, *, root: str | Path | None = 'static') -> tuple[bytes, str, int]:
@@ -130,21 +107,18 @@ class App:
try:
directory = Path(root or 'static').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
+ if not path.is_relative_to(directory) or not path.is_file(): return b'Not Found', 'text/plain; charset=utf-8', 404
+ return path.read_bytes(), guess_type(path.name)[0] or 'application/octet-stream', 200
+ except (OSError, ValueError, RuntimeError): return b'Not Found', 'text/plain; charset=utf-8', 404
return await to_thread(read)
def route(self, path: str, methods: list[str] | tuple[str, ...] = ('GET',)) -> Callable:
"""Decorator to register routes with specific HTTP methods."""
path = path.rstrip('/') or '/'
def decorator(handler: Callable) -> Callable:
- routes = self.pattern_routes if '{' in path and '}' in path else self.routes
- key = tuple(path.split('/')) if routes is self.pattern_routes else path
- routes.setdefault(key, {}).update({m.upper(): handler for m in methods})
+ target = self.pattern_routes if '{' in path and '}' in path else self.routes
+ key = tuple(path.split('/')) if target is self.pattern_routes else path
+ target.setdefault(key, {}).update({m.upper(): handler for m in methods})
return handler
return decorator