Dummy API Server with boom style error response

dummy_server.py

#!/usr/bin/env python3

import html
import json
import socketserver
import sys
from datetime import datetime
from http import HTTPStatus
from http.server import SimpleHTTPRequestHandler
from time import mktime
from wsgiref.handlers import format_date_time

PORT = 8282


class ErrorResponse:
    """
    https://hapi.dev/module/boom/api/
    """

    code: int
    error: str
    message: str
    ext: dict[str, any]

    def __init__(
        self,
        code: int,
        message: str | None = None,
        explain: str | None = None,
        **kwargs,
    ) -> None:
        try:
            shortmsg, longmsg = SimpleHTTPRequestHandler.responses[code]
        except KeyError:
            shortmsg, longmsg = "???", "???"
        self.code = code
        self.error = shortmsg if message is None else message
        self.message = longmsg if explain is None else explain
        self.ext = kwargs

    @property
    def payload(self) -> dict[str, str]:
        return {
            "statusCode": self.code,
            "error": self.error,
            "message": self.message,
        } | {key: self.convert(self.ext[key]) for key in self.ext}

    def __str__(self) -> str:
        return html.escape(json.dumps(self.payload), quote=False)

    @classmethod
    def convert(cls, value: any) -> str:
        if isinstance(value, datetime):
            return format_date_time(mktime(value.timetuple()))
        return str(value)


class Handler(SimpleHTTPRequestHandler):
    error_content_type = "application/json; charset=utf-8"

    @classmethod
    def format_datetime(cls, value: datetime | int | float) -> str:
        if isinstance(value, datetime):
            value = mktime(value.timetuple())
        return format_date_time(value)

    def send_error(
        self,
        code: int,
        message: str | None = None,
        explain: str | None = None,
        headers: dict[str, str] | None = None,
        **kwargs,
    ) -> None:
        payload = ErrorResponse(code, message, explain, **kwargs)

        self.log_error("code %d, message %s", payload.code, payload.error)
        self.send_response(payload.code, payload.error)
        self.send_header("Connection", "close")
        if isinstance(headers, dict):
            for key, value in headers.items():
                self.send_header(key, value)

        body = str(payload).encode("UTF-8", "replace")
        if code >= 200 and code not in (
            HTTPStatus.NO_CONTENT,
            HTTPStatus.RESET_CONTENT,
            HTTPStatus.NOT_MODIFIED,
        ):
            self.send_header("Content-Type", self.error_content_type)
            self.send_header("Content-Length", str(len(body)))
        self.end_headers()

        if self.command != "HEAD" and body:
            self.wfile.write(body)

    def do_GET(self):
        after = datetime(2023, 10, 9, 20)
        headers = {
            "Retry-After": self.format_datetime(after),
        }
        self.send_error(503, headers=headers, retryAfter=headers["Retry-After"])
        return None


def run(address, port=PORT):
    address = "127.0.0.1" if address in ["127.0.0.1", "localhost"] else ""
    with socketserver.TCPServer((address, port), Handler) as httpd:
        print(f"Serving at http://{address}:{port}")
        httpd.serve_forever()


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser()
    parser.add_argument(
        "port", nargs="?", default=PORT, type=int, help=f"port number (default: {PORT})"
    )
    params = parser.parse_args()

    try:
        run("localhost", params.port)
    except (RuntimeError, OSError) as exc:
        print(exc, file=sys.stderr)
        sys.exit(1)
    except KeyboardInterrupt:
        print("Terminated.")