POST requests

pyproject.toml

@@ -17,7 +17,8 @@ dependencies = [
   "PyYAML>=6.0",
   "docutils>=0.20",
   "starlette>=0.37",
-  "uvicorn>=0.30"
+  "uvicorn>=0.30",
+  "python-multipart>=0.0.9"
 ]
 keywords = [
   "Materials Discovery",

src/httk/web/engine/site_engine.py

@@ -7,6 +7,7 @@
 from httk.web.model.config import SiteConfig
 from httk.web.model.errors import FunctionInjectionError, NotFoundError
 from httk.web.model.page import PageResult, ResolvedRoute
+from httk.web.model.request import HttpRequestContext
 from httk.web.renderers import RENDERERS_BY_SUFFIX
 from httk.web.templating import JinjaTemplateEngine, TemplateRenderInput
 
@@ -20,7 +21,12 @@ def __init__(self, config: SiteConfig) -> None:
     def resolve(self, route: str) -> ResolvedRoute:
         return resolve_route(config=self.config, route=route)
 
-    def render(self, route: str, query: dict[str, str] | None = None) -> PageResult:
+    def render(
+        self,
+        route: str,
+        query: dict[str, str] | None = None,
+        request: HttpRequestContext | None = None,
+    ) -> PageResult:
         resolved = self.resolve(route)
 
         if resolved.kind == "missing" or resolved.source_path is None:
@@ -30,19 +36,40 @@ def render(self, route: str, query: dict[str, str] | None = None) -> PageResult:
             content_type = guess_type(str(resolved.source_path))[0] or "application/octet-stream"
             return PageResult(status_code=200, content_type=content_type, body=resolved.source_path.read_bytes())
 
-        query_params = dict(query or {})
+        if request is None:
+            request_context = HttpRequestContext(query=dict(query or {}))
+        else:
+            request_context = request
+            if query is not None:
+                request_context = HttpRequestContext(
+                    method=request.method,
+                    query=dict(query),
+                    postvars=request.postvars,
+                    headers=request.headers,
+                )
+
+        query_params = dict(request_context.query)
+        postvars = dict(request_context.postvars)
+        request_params = dict(query_params)
+        request_params.update(postvars)
         rendered_html, metadata = self._render_content_without_templates(resolved)
         route_key = normalize_route(route)
         warnings: list[str] = []
 
         template_name = self._metadata_string(metadata, "template", default="default")
         base_template_name = self._metadata_string(metadata, "base_template", default="base_default")
 
-        context = self._build_template_context(route_key=route_key, metadata=metadata, query=query_params)
+        context = self._build_template_context(
+            route_key=route_key,
+            metadata=metadata,
+            query=query_params,
+            postvars=postvars,
+            request=request_context,
+        )
         self._apply_function_injections(
             metadata=metadata,
             context=context,
-            query=query_params,
+            params=request_params,
             route_key=route_key,
             warnings=warnings,
         )
@@ -82,6 +109,8 @@ def _build_template_context(
         route_key: str,
         metadata: dict[str, object],
         query: dict[str, str],
+        postvars: dict[str, str],
+        request: HttpRequestContext,
     ) -> dict[str, object]:
         context: dict[str, object] = dict(metadata)
         page_cache: dict[str, tuple[str, dict[str, object]]] = {}
@@ -144,6 +173,8 @@ def pages(path: str, field: str) -> object:
         context["listdir"] = listdir
         context["pages"] = pages
         context["query"] = dict(query)
+        context["postvars"] = dict(postvars)
+        context["request"] = request
         context["page"] = {
             "relurl": route_key,
             "absurl": self._absolute_url(route_key),
@@ -156,7 +187,7 @@ def _apply_function_injections(
         *,
         metadata: dict[str, object],
         context: dict[str, object],
-        query: dict[str, str],
+        params: dict[str, str],
         route_key: str,
         warnings: list[str],
     ) -> None:
@@ -173,16 +204,16 @@ def _apply_function_injections(
                 function_name, arg_specs, function_template = self._parse_function_spec(raw_spec)
                 required = [x.strip() for x in arg_specs.split(",") if x.strip()]
 
-                if not self._function_args_satisfied(required, query):
+                if not self._function_args_satisfied(required, params):
                     context[output_name] = ""
                     metadata[output_name] = ""
                     warnings.append(
-                        f"Function '{function_name}' on route '{route_key}' skipped: query constraints not satisfied."
+                        f"Function '{function_name}' on route '{route_key}' skipped: input parameter constraints not satisfied."
                     )
                     del metadata[function_key]
                     continue
 
-                result = self.function_handler.execute(function_name=function_name, query=query, global_data=context)
+                result = self.function_handler.execute(function_name=function_name, params=params, global_data=context)
                 joint_context = dict(context)
                 joint_context["result"] = result
 

src/httk/web/functions/python_module.py

@@ -12,15 +12,15 @@ def __init__(self, functions_dir: Path) -> None:
         self._module_cache: dict[Path, ModuleType] = {}
         self._cache_lock = threading.Lock()
 
-    def execute(self, *, function_name: str, query: dict[str, str], global_data: dict[str, object]) -> Any:
+    def execute(self, *, function_name: str, params: dict[str, str], global_data: dict[str, object]) -> Any:
         module_path = self._resolve_function_path(function_name)
         module = self._load_module(module_path)
 
         execute_fn = getattr(module, "execute", None)
         if execute_fn is None or not callable(execute_fn):
             raise ValueError(f"Function module missing callable execute(): {module_path}")
 
-        callargs: dict[str, object] = dict(query)
+        callargs: dict[str, object] = dict(params)
         callargs["global_data"] = global_data
         return execute_fn(**callargs)
 

src/httk/web/model/__init__.py

@@ -1,6 +1,7 @@
 from .config import SiteConfig
 from .errors import FunctionInjectionError, NotFoundError, WebError
 from .page import PageResult, PublishReport, ResolvedRoute
+from .request import HttpRequestContext
 
 __all__ = [
     "SiteConfig",
@@ -10,4 +11,5 @@
     "ResolvedRoute",
     "PageResult",
     "PublishReport",
+    "HttpRequestContext",
 ]

src/httk/web/model/request.py

@@ -0,0 +1,9 @@
+from dataclasses import dataclass, field
+
+
+@dataclass(frozen=True)
+class HttpRequestContext:
+    method: str = "GET"
+    query: dict[str, str] = field(default_factory=dict)
+    postvars: dict[str, str] = field(default_factory=dict)
+    headers: dict[str, str] = field(default_factory=dict)

src/httk/web/runtime/asgi.py

@@ -1,25 +1,101 @@
+import json
+from urllib.parse import parse_qsl
+
 from starlette.applications import Starlette
+from starlette.datastructures import UploadFile
 from starlette.requests import Request
 from starlette.responses import Response
 from starlette.routing import Route
 
 from httk.web.engine.site_engine import SiteEngine
 from httk.web.model.errors import WebError
+from httk.web.model.request import HttpRequestContext
+
+MAX_POST_BODY_BYTES = 1_000_000
 
 
 async def _handle_request(request: Request) -> Response:
     engine: SiteEngine = request.app.state.engine
     route = request.path_params.get("path", "")
-    query = dict(request.query_params)
     try:
-        result = engine.render(route, query=query)
+        request_context = HttpRequestContext(
+            method=request.method,
+            query=dict(request.query_params),
+            postvars=await _extract_postvars(request),
+            headers={k.lower(): v for k, v in request.headers.items()},
+        )
+        result = engine.render(route, request=request_context)
     except WebError as exc:
         return Response(content=str(exc), status_code=exc.status_code, media_type="text/plain")
 
     return Response(content=result.body, status_code=result.status_code, media_type=result.content_type)
 
 
 def create_app(*, engine: SiteEngine, debug: bool = False) -> Starlette:
-    app = Starlette(debug=debug, routes=[Route("/{path:path}", _handle_request, methods=["GET"])])
+    app = Starlette(debug=debug, routes=[Route("/{path:path}", _handle_request, methods=["GET", "POST"])])
     app.state.engine = engine
     return app
+
+
+async def _extract_postvars(request: Request) -> dict[str, str]:
+    if request.method.upper() != "POST":
+        return {}
+
+    content_type = request.headers.get("content-type", "").lower()
+    content_length = request.headers.get("content-length")
+    if content_length is not None:
+        try:
+            if int(content_length) > MAX_POST_BODY_BYTES:
+                raise WebError(
+                    f"POST body too large (>{MAX_POST_BODY_BYTES} bytes).",
+                    status_code=413,
+                )
+        except ValueError:
+            # Ignore invalid content-length values and attempt best-effort parsing.
+            pass
+
+    if "application/x-www-form-urlencoded" in content_type:
+        raw = await request.body()
+        if len(raw) > MAX_POST_BODY_BYTES:
+            raise WebError(
+                f"POST body too large (>{MAX_POST_BODY_BYTES} bytes).",
+                status_code=413,
+            )
+        body = raw.decode("utf-8", errors="replace")
+        return {k: v for k, v in parse_qsl(body, keep_blank_values=True)}
+
+    if "application/json" in content_type:
+        raw = await request.body()
+        if len(raw) > MAX_POST_BODY_BYTES:
+            raise WebError(
+                f"POST body too large (>{MAX_POST_BODY_BYTES} bytes).",
+                status_code=413,
+            )
+        try:
+            payload = json.loads(raw.decode("utf-8", errors="replace"))
+        except json.JSONDecodeError:
+            return {}
+        if isinstance(payload, dict):
+            out_json: dict[str, str] = {}
+            for key, value in payload.items():
+                if isinstance(value, (str, int, float, bool)) or value is None:
+                    out_json[str(key)] = "" if value is None else str(value)
+            return out_json
+        return {}
+
+    if "multipart/form-data" in content_type:
+        try:
+            form = await request.form()
+        except Exception:
+            return {}
+        out_form: dict[str, str] = {}
+        for key, value in form.multi_items():
+            normalized_key = str(key)
+            if isinstance(value, UploadFile):
+                # Keep uploads explicit and compact in postvars.
+                out_form[normalized_key] = value.filename or ""
+            else:
+                out_form[normalized_key] = str(value)
+        return out_form
+
+    return {}