--- /dev/null
+"""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))