patx/micropie
Enhance _asgi_app_http to support middleware-driven routing
Commit f7f95d9 · patx · 2025-06-05T00:07:02-04:00
Enhance _asgi_app_http to support middleware-driven routing Update _asgi_app_http to: - Allow underscore-prefixed handler names when set via request._route_handler, while restricting them for URL-derived paths. - Explicitly check for _route_handler using hasattr for robust middleware integration. - Respect path_params set by middleware, improving routing flexibility.
Comments
No comments yet.
Diff
diff --git a/MicroPie.py b/MicroPie.py
index 9b07567..c041e80 100644
--- a/MicroPie.py
+++ b/MicroPie.py
@@ -249,12 +249,18 @@ class App:
# Parse path and find handler
path: str = scope["path"].lstrip("/")
parts: List[str] = path.split("/") if path else []
- func_name: str = parts[0] if parts else "index"
- if func_name.startswith("_"):
- await self._send_response(send, 404, "404 Not Found")
- return
+ # Check if request._route_handler has been set by middleware
+ if hasattr(request, "_route_handler"):
+ func_name: str = request._route_handler
+ else:
+ func_name: str = parts[0] if parts else "index"
+ if func_name.startswith("_"):
+ await self._send_response(send, 404, "404 Not Found")
+ return
- request.path_params = parts[1:] if len(parts) > 1 else []
+ # Respect path_params set in middleware
+ if not request.path_params:
+ request.path_params = parts[1:] if len(parts) > 1 else []
handler = getattr(self, func_name, None) or getattr(self, "index", None)
if not handler:
await self._send_response(send, 404, "404 Not Found")
diff --git a/examples/middleware/router.py b/examples/middleware/router.py
index e6bb02d..fd45094 100644
--- a/examples/middleware/router.py
+++ b/examples/middleware/router.py
@@ -1,102 +1,47 @@
import re
-import inspect
-import json
-from typing import Callable, Dict, List, Optional, Tuple, Any
+from typing import Dict, List, Optional, Tuple, Any
from MicroPie import App, HttpMiddleware, Request
-class ExplicitRoutingMiddleware(HttpMiddleware):
+class ExplicitRouter(HttpMiddleware):
def __init__(self):
- # Registry to map route patterns to handler callables and HTTP methods
- self.routes: Dict[str, Tuple[Callable, str, str]] = {}
+ # Map route paths to (method, regex pattern, handler name)
+ self.routes: Dict[str, Tuple[str, str, str]] = {}
- def add_route(self, path: str, handler: Callable, method: str = "GET") -> None:
+ def add_route(self, path: str, handler_name: str, method: str = "GET") -> None:
"""
- Register an explicit route with its handler callable and HTTP method.
+ Register an explicit route with its handler method name and HTTP method.
Args:
path: The route pattern (e.g., "/api/users/{user}/records/{record}")
- handler: The handler method callable (e.g., app.get_record)
+ handler_name: The handler method name (e.g., "api")
method: The HTTP method (e.g., "GET", "POST")
"""
- # Convert path pattern to regex (e.g., "/api/users/{user}/records/{record}" -> "^/api/users/([^/]+)/records/([^/]+)$")
pattern = re.sub(r"{([^}]+)}", r"([^/]+)", path)
pattern = f"^{pattern}$"
- self.routes[path] = (handler, method, pattern)
+ self.routes[path] = (method, pattern, handler_name)
async def before_request(self, request: Request) -> Optional[Dict]:
"""
- Match the request path against registered routes and dispatch to the handler.
+ Match the request path and set path parameters for MicroPie routing.
Args:
request: The MicroPie Request object
Returns:
- Optional response dict if handled, None to continue to implicit routing
+ None to let MicroPie handle parsing and routing
"""
path = request.scope["path"]
- request_method = request.method
- for route_path, (handler, method, pattern) in self.routes.items():
- if method != request_method:
+ for route_path, (method, pattern, handler_name) in self.routes.items():
+ if request.method != method:
continue
match = re.match(pattern, path)
if match:
- # Extract path parameters
- path_params = list(match.groups())
-
- # Build arguments based on handler signature
- sig = inspect.signature(handler)
- func_args = []
- path_params_copy = path_params[:]
- for param in sig.parameters.values():
- if param.name == "self":
- continue
- if param.kind == inspect.Parameter.VAR_POSITIONAL:
- func_args.extend(path_params_copy)
- path_params_copy = []
- continue
- if path_params_copy:
- func_args.append(path_params_copy.pop(0))
- elif param.default is not param.empty:
- func_args.append(param.default)
- else:
- return {
- "status_code": 400,
- "body": f"400 Bad Request: Missing required parameter '{param.name}'",
- "headers": []
- }
-
- try:
- # Call handler with path parameters
- result = await handler(*func_args) if inspect.iscoroutinefunction(handler) else handler(*func_args)
- status_code = 200
- response_body = result
- extra_headers = []
-
- # Handle tuple response (body, status_code, headers)
- if isinstance(result, tuple):
- status_code, response_body = result[0], result[1]
- extra_headers = result[2] if len(result) > 2 else []
-
- # Convert dict/list to JSON if needed
- if isinstance(response_body, (dict, list)):
- response_body = json.dumps(response_body)
- extra_headers.append(("Content-Type", "application/json"))
-
- return {
- "status_code": status_code,
- "body": response_body,
- "headers": extra_headers
- }
- except Exception as e:
- print(f"Handler error: {e}")
- return {
- "status_code": 500,
- "body": "500 Internal Server Error",
- "headers": []
- }
+ # Ensure path parameters are strings
+ request.path_params = [str(param) for param in match.groups()]
+ request._route_handler = handler_name
+ return None
- # No matching explicit route, proceed to implicit routing
return None
async def after_request(
@@ -106,58 +51,51 @@ class ExplicitRoutingMiddleware(HttpMiddleware):
response_body: Any,
extra_headers: List[Tuple[str, str]]
) -> Optional[Dict]:
- """
- Pass through the response unchanged.
-
- Args:
- request: The MicroPie Request object
- status_code: HTTP status code
- response_body: Response body
- extra_headers: List of response headers
-
- Returns:
- None to pass through the response unchanged
- """
return None
-# Example usage
+
class MyApp(App):
def __init__(self):
super().__init__()
- self.router = ExplicitRoutingMiddleware()
+ self.router = ExplicitRouter()
self.middlewares.append(self.router)
# Register explicit routes
- self.router.add_route("/api/users/{user}/records/{record}", self.get_record, "GET")
- self.router.add_route("/api/users/{user}/records", self.create_record, "POST")
- self.router.add_route("/api/users/{user}/records/{record}/details/subdetails", self.get_record_subdetails, "GET")
- # Implicit route handled by MicroPie's default routing
- # Note: /records/{user}/{record} will use implicit routing since not explicitly defined
+ self.router.add_route("/api/users/{user}/records/{record}", "_get_record", "GET")
+ self.router.add_route("/api/users/{user}/records", "_create_record", "POST")
+ self.router.add_route("/api/users/{user}/records/{record}/details/subdetails", "_get_record_subdetails", "GET")
- async def get_record(self, user: str, record: str):
+ async def _get_record(self, user: str, record: str):
try:
record_id = int(record)
- # Access request via self.request if needed
return {"user": user, "record": record_id}
except ValueError:
- return {"error": "Record must be an integer"}, 400
+ return {"error": "Record must be an integer"}
- async def create_record(self, user: str):
- data = self.request.get_json
- return {"user": user, "record": data.get("record_id"), "created": True}, 201
+ async def _create_record(self, user: str):
+ try:
+ data = self.request.get_json
+ return {"user": user, "record": data.get("record_id"), "created": True}
+ except Exception as e:
+ return {"error": f"Invalid JSON: {str(e)}"}
- async def get_record_subdetails(self, user: str, record: str):
+ async def _get_record_subdetails(self, user: str, record: str):
try:
record_id = int(record)
return {"user": user, "record": record_id, "subdetails": "more detailed info"}
except ValueError:
- return {"error": "Record must be an integer"}, 400
+ return {"error": "Record must be an integer"}
+ # Implicitly routed
async def records(self, user: str, record: str):
try:
record_id = int(record)
return {"user": user, "record": record_id, "implicit": True}
except ValueError:
- return {"error": "Record must be an integer"}, 400
+ return {"error": "Record must be an integer"}
+
+ # Private route, not exposed in any way
+ async def _private(self):
+ return {"viewing": "private"}
app = MyApp()