An extreme simple Python http server with auto reload and file-system based router

__init__.py

import os

from .app import start_cli_service


env = (os.environ.get("env") or "prod").lower()
is_dev = env == "dev" or env == "local"
port, autoreload_observer = start_cli_service(autoreload=is_dev)
if autoreload_observer:
    # move autoreload observer to the foreground so process won't exit
    # while reloading.
    autoreload_observer.join()

app.py

import importlib
import importlib.util
import logging
import sys
import threading
import time
from threading import Thread
from typing import Set

from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler

from . import httpserver
from .utils import file_path_to_module_name, is_port_in_use, root_path


logger = logging.getLogger("of.cli_service")
logger.handlers = [logging.StreamHandler()]


def start_server_thread(port: int):
    """Start the CLI http server in another thread"""
    try:
        importlib.reload(httpserver)  # always reload the module
    except SyntaxError:
        # ignore syntax error (which often means we are editting files)
        logger.exception("Reloading httpserver module failed")
        return
    server = httpserver.CliHttpServer(port)
    server_thread = Thread(target=lambda: server.start())
    server_thread.start()
    return server


class AutoreloadHandler(PatternMatchingEventHandler):
    """Auto reload handlers"""

    def __init__(
        self,
        server: httpserver.CliHttpServer,
        patterns=None,
        ignore_patterns=None,
        ignore_directories=False,
        case_sensitive=False,
    ):
        super().__init__(patterns, ignore_patterns, ignore_directories, case_sensitive)
        self.server = server
        self.needs_reload: Set[str] = set()  # modules that need to be reloaded

        # mark last update to make sure we only reload 1s after the last update

        self.last_updated_at = time.time()
        # start another thread to check if we need to reload
        threading.Thread(target=lambda: self.check_reload()).start()

    def reload_modules(self):
        logging.debug("%s modules updated", len(self.needs_reload))
        for mname in self.needs_reload:
            logging.debug(mname)
            if mname in sys.modules:
                importlib.reload(sys.modules[mname])
        self.needs_reload.clear()

    def check_reload(self):
        while True:
            if self.needs_reload and time.time() - self.last_updated_at > 1:
                logger.debug("Change detected, restarting CLI service...\n")
                self.reload_modules()
                self.server.stop()
                new_server = start_server_thread(self.server.server_port)
                if new_server:
                    self.server = new_server
            time.sleep(1)

    def on_any_event(self, event):
        self.needs_reload.add(file_path_to_module_name(event.src_path))
        self.last_updated_at = time.time()


def start_cli_service(autoreload=True):
    """Find available private (dynamic) port for the CLI service and start
    the service."""
    port, max_port = httpserver.PORT_RANGE
    while port < max_port and is_port_in_use(port):
        port += 1

    server = start_server_thread(port)
    observer = None

    if autoreload:
        # start autoreload watcher in another thread
        observer = Observer()
        observer.schedule(
            AutoreloadHandler(
                server=server,
                patterns=["*.py"],
                ignore_patterns=["__pycache__", "cli_service/app.py"],
            ),
            # monitor all "of.xxx" files
            root_path,
            recursive=True,
        )
        observer.start()
        observer.on_thread_stop = lambda: server.stop()

    return port, observer

httpserver.py

"""
An extreme simple http server
"""
from http import HTTPStatus
from importlib import import_module
import json
import logging
from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
from typing import Any

logger = logging.getLogger("of.cli_service")

# Allowed port range for the CLI service
# i.e., a subset of dynamic port range regulated by IANA RFC 6335
# RFC 6335 DPR starts from 49152, we use a larger port to reduce chances of
# collision.
PORT_RANGE = [49200, 65535]
ALLOWED_ORIGINS = ["http://localhost:8081"]

class CliServiceRequestHandler(BaseHTTPRequestHandler):
    def end_headers(self):
        self.send_cors_headers()
        BaseHTTPRequestHandler.end_headers(self)

    def send_cors_headers(self):
        origin = self.headers.get("origin")
        if origin in ALLOWED_ORIGINS:
            self.send_header("Access-Control-Allow-Origin", origin)

    def send_json(self, obj: Any):
        self.send_header("Content-Type", "application/json;charset=utf-8")
        self.end_headers()
        self.wfile.write(json.dumps(obj).encode("UTF-8", "replace"))

    def send_text(self, text: str, content_type: str = "text/plain"):
        self.send_header("Content-Type", f"{content_type};charset=utf-8")
        self.end_headers()
        self.wfile.write(text.encode("UTF-8", "replace"))

    def send_error(self, code: int, message: str = None, explain: str = None):
        try:
            shortmsg, longmsg = self.responses[code]
        except KeyError:
            shortmsg, longmsg = "???", "???"
        if message is None:
            message = shortmsg
        if explain is None:
            explain = longmsg
        self.log_error("code %d, message %s", code, message)
        self.send_response(code, message)
        return self.send_json({"error": message, "explain": explain})

    def handle_request(self) -> None:
        self.send_cors_headers()

        method, path = self.command, self.path
        mod = None
        try:
            mname = f'of.cli_service.routes{path.replace("/", ".")}'.strip(".")
            mod = import_module(mname)
        except ModuleNotFoundError:
            return self.send_error(HTTPStatus.NOT_FOUND)

        handler = mod and getattr(mod, method, None)
        if not handler:
            return self.send_error(HTTPStatus.METHOD_NOT_ALLOWED)

        self.send_response(HTTPStatus.ACCEPTED)
        response = handler(self)
        if response:
            if isinstance(response, str):
                self.send_text(response)
            else:
                self.send_json(response)

    def do_HEAD(self):
        self.handle_request()

    def do_GET(self):
        self.handle_request()

    def do_POST(self):
        self.handle_request()


class CliHttpServer(ThreadingHTTPServer):
    def __init__(self, port=PORT_RANGE[0]):
        super().__init__(("localhost", port), CliServiceRequestHandler)

    def start(self):
        logger.info(f"🚀 CLI service started at http://localhost:{self.server_port}")
        self.serve_forever()

    def stop(self):
        self.shutdown()
        self.server_close()

routers.__init__.py

def GET(handler):
  return "OK"

utils.py

import socket
import os

root_path = os.path.dirname(os.path.dirname(__file__))


def is_port_in_use(port: int) -> bool:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        return s.connect_ex(("localhost", port)) == 0


def file_path_to_module_name(path: str) -> str:
    """Convert file path to module name"""
    return (
        os.path.abspath(path)
        .replace(root_path, "of")
        .replace(".py", "")
        .replace("/__init__", "")
        .replace("/", ".")
    )

nice

Cool