From: Christian Heller <c.heller@plomlompom.de>
Date: Sat, 18 Jan 2025 02:25:29 +0000 (+0100)
Subject: Add web stuff.
X-Git-Url: https://plomlompom.com/repos/%7B%7Bdb.prefix%7D%7D/static/process?a=commitdiff_plain;h=refs%2Fheads%2Fmaster;p=plomlib

Add web stuff.
---

diff --git a/web.py b/web.py
new file mode 100644
index 0000000..629e63b
--- /dev/null
+++ b/web.py
@@ -0,0 +1,107 @@
+"""HTTP services."""
+from http.server import HTTPServer, BaseHTTPRequestHandler
+from json import loads as json_loads
+from pathlib import Path
+from typing import Any, Generic, Optional, Type, TypeVar
+from urllib.parse import parse_qs, urlparse
+from jinja2 import (
+        Environment as JinjaEnv, FileSystemLoader as JinjaFSLoader)
+
+
+MIME_APP_JSON = 'application/json'
+
+
+Server = TypeVar('Server', bound='PlomHttpServer')
+QueryMap = TypeVar('QueryMap', bound='PlomQueryMap')
+
+
+class PlomHttpServer(HTTPServer):
+    """Basic HTTP server."""
+
+    def __init__(self, templates_dir: Path, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        self.jinja = JinjaEnv(loader=JinjaFSLoader(templates_dir))
+
+    def serve(self) -> None:
+        """Do .serve_forever on .server_port/.server_address until ^C."""
+        print(f'running at port {self.server_port}')
+        try:
+            self.serve_forever()
+        except KeyboardInterrupt:
+            print('aborted due to keyboard interrupt')
+        self.server_close()
+
+
+class PlomHttpHandler(BaseHTTPRequestHandler, Generic[Server, QueryMap]):
+    """Basic HTTP request handler for use with PlomHttpServer."""
+    server: Server
+    params: QueryMap
+    path_toks: tuple[str, ...]
+    postvars: QueryMap
+    mapper: Type[QueryMap]
+    pagename: str
+
+    def parse_request(self) -> bool:
+        """Extend by setup of .params, .path_toks."""
+        if not super().parse_request():
+            return False
+        url = urlparse(self.path)
+        self.params = self.mapper(url.query)
+        self.path_toks = Path(url.path).parts
+        self.pagename = self.path_toks[1] if len(self.path_toks) > 1 else ''
+        self.postvars = self.mapper(
+                self.rfile.read(int(self.headers.get('content-length', 0))
+                                ).decode('utf8'),
+                self.headers['content-type'])
+        return True
+
+    def send_http(self,
+                  content: bytes = b'',
+                  headers: Optional[list[tuple[str, str]]] = None,
+                  code: int = 200
+                  ) -> None:
+        """Send HTTP response."""
+        headers = headers if headers else []
+        self.send_response(code)
+        for header in headers:
+            self.send_header(*header)
+        self.end_headers()
+        if content:
+            self.wfile.write(content)
+
+    def send_rendered(self,
+                      tmpl_name: Path,
+                      ctx: dict[str, Any],
+                      code: int = 200
+                      ) -> None:
+        """Send rendered page of tmpl_name template, with ctx data."""
+        tmpl = self.server.jinja.get_template(str(tmpl_name))
+        self.send_http(bytes(tmpl.render(**ctx), encoding='utf8'), code=code)
+
+    def redirect(self, target: Path) -> None:
+        """Redirect browser to target."""
+        self.send_http(headers=[('Location', str(target))], code=302)
+
+
+class PlomQueryMap:
+    """Convenience wrapper over somewhat dict-shaped HTTP inputs."""
+
+    def __init__(self, serialized: str, content_type: str = '') -> None:
+        if content_type == MIME_APP_JSON:
+            self.as_dict = json_loads(serialized)
+        else:
+            self.as_dict = parse_qs(serialized, keep_blank_values=True,
+                                    strict_parsing=True, errors='strict')
+
+    def all(self, key: str) -> Optional[list[str]]:
+        """Return all values mapped to key, None if key not present."""
+        return self.as_dict.get(key, None)
+
+    def first(self, key: str) -> Optional[str]:
+        """Return first value mapped to key, if found, else None."""
+        values = self.all(key)
+        return values[0] if values else None
+
+    def keys_prefixed(self, prefix: str) -> tuple[str, ...]:
+        """Return all present keys starting with prefix."""
+        return tuple(k for k in self.as_dict if k.startswith(prefix))