html_compose.gallery.server

View Source

  1import http.server
  2import importlib
  3import mimetypes
  4import os.path
  5import sys
  6import threading
  7from pathlib import Path
  8from time import sleep
  9from typing import cast
 10from urllib.parse import parse_qs, urlparse
 11
 12from ..live import WatchCond, Watcher, livereload_server
 13from . import app
 14from .impl import (
 15    ShowcaseResult,
 16    env_registry,
 17    show_showcases,
 18    showcase_registry,
 19)
 20
 21
 22class Settings:
 23    resource_prefix = "/"
 24    static_dir = Path.cwd() / "static"
 25
 26    @staticmethod
 27    def set_static_dir(path: str) -> None:
 28        Settings.static_dir = Path(path)
 29
 30
 31# Common web mimetypes if the system mimetypes db fails
 32WEB_MIMETYPES = {
 33    ".html": "text/html",
 34    ".css": "text/css",
 35    ".js": "application/javascript",
 36    ".png": "image/png",
 37    ".jpg": "image/jpeg",
 38    ".gif": "image/gif",
 39    ".svg": "image/svg+xml",
 40}
 41
 42
 43class GalleryRequestHandler(http.server.BaseHTTPRequestHandler):
 44    content: list[ShowcaseResult] = []
 45
 46    def _handle(self) -> None:
 47        parsed = urlparse(self.path)
 48        path = parsed.path
 49        query = parse_qs(parsed.query)
 50        content = GalleryRequestHandler.content
 51
 52        if path == "/" or path == "/index.html":
 53            self.send_response(200)
 54            self.send_header("Content-Type", "text/html; charset=utf-8")
 55            self.end_headers()
 56            self.wfile.write(app.main_route(content).encode("utf-8"))
 57            return
 58
 59        for c in content:
 60            if path == f"/showcase/{c.func_module}/{c.func_name}":
 61                # Filter on name if provided
 62                test_name = query.get("name", None)
 63                if test_name:
 64                    if c.name != test_name[0]:
 65                        continue
 66
 67                if not c.success:
 68                    self.send_response(500)
 69                    self.send_header("Content-Type", "text/plain")
 70                    self.end_headers()
 71                    self.wfile.write(f"Showcase broken: {c.result}".encode())
 72                    return
 73
 74                # Serve the showcase
 75                self.send_response(200)
 76                self.send_header("Content-Type", "text/html; charset=utf-8")
 77                self.end_headers()
 78                self.wfile.write(app.generate_showcase(c).encode("utf-8"))
 79                return
 80
 81        # Try to serve static files
 82        if self._serve_static(path):
 83            return
 84
 85        # Nothing matched
 86        self.send_response(404)
 87        self.end_headers()
 88
 89    def do_GET(self):
 90        # This method handles all GET requests
 91        self._handle()
 92
 93    def _serve_static(self, path: str) -> bool:
 94        if not Settings.static_dir:
 95            return False
 96        prefix = Settings.resource_prefix
 97        if path.startswith(prefix):
 98            normalized = path.removeprefix(prefix).lstrip("/")
 99        else:
100            return False
101
102        # Collapse .. to prevent directory traversal without resolving symlinks
103        candidate = Path(os.path.normpath(Settings.static_dir / normalized))
104
105        # Not relative to static dir
106        if not candidate.is_relative_to(
107            Path(os.path.normpath(Settings.static_dir))
108        ):
109            return False
110
111        # Resolve symlinks for the actual file read
112        # as a developer server, we assume your symlinks are safe
113        candidate = candidate.resolve()
114
115        if not candidate.is_file():
116            return False
117
118        mime, _ = mimetypes.guess_type(candidate.name)
119        if not mime:
120            # Fall through if the mimetype db failed to identify
121            # basic web types
122            ext = candidate.suffix
123            # fallback to application/octet-stream if we really
124            # have no idea
125            mime = WEB_MIMETYPES.get(ext, "application/octet-stream")
126
127        self.send_response(200)
128        self.send_header("Content-Type", mime)
129        self.send_header("Content-Length", str(candidate.stat().st_size))
130        self.end_headers()
131
132        # Stream the file in chunks to avoid loading large files into memory
133        with candidate.open("rb") as f:
134            while True:
135                data = f.read(8192)
136                if not data:
137                    break
138                self.wfile.write(data)
139
140        return True
141
142
143def run(
144    module_path: str | Path,
145    host: str,
146    port: int,
147    livereload_port: int,
148    livereload_host: str,
149    force_polling: bool = False,
150    extra_paths: list[str] | None = None,
151):
152    # Resolve to absolute before accessing parts to handle absolute file paths safely
153    # If the user passed an absolute path, we inject its directory into sys.path
154    mp_raw = Path(module_path)
155    if mp_raw.is_absolute():
156        if str(mp_raw.parent) not in sys.path:
157            sys.path.insert(0, str(mp_raw.parent))
158        mp = Path(mp_raw.name)
159    else:
160        mp = mp_raw
161
162    if mp.suffix == ".py":
163        mp = mp.with_suffix("")
164
165    showcase_registry.clear()
166    module_name = ".".join(mp.parts)
167
168    if module_name in sys.modules:
169        del sys.modules[module_name]
170
171    importlib.invalidate_caches()
172
173    app.Settings.livereload_host = livereload_host
174    app.Settings.livereload_port = livereload_port
175
176    try:
177        module = importlib.import_module(module_name)
178    except Exception as e:
179        print(f"Error importing module {module_name}: {e}")
180        raise
181    assert module is not None
182
183    watch_globs = []
184    ignore_glob = []
185    if extra_paths:
186        for ep in extra_paths:
187            watch_globs.append(ep)
188    if hasattr(module, "__path__"):
189        pathlist = module.__path__
190    elif hasattr(module, "__file__"):
191        pathlist = [os.path.dirname(cast(str, module.__file__))]
192    else:
193        pathlist = []
194        print("Warning: could not determine module path for file watching")
195
196    for watch_path in pathlist:
197        loaded_mod_path = Path(watch_path)
198        watch_globs.extend([f"{loaded_mod_path / '**/*.py'}"])
199        ignore_glob.extend(
200            [
201                f"{loaded_mod_path / venv}"
202                for venv in [
203                    "venv",
204                    ".venv",
205                    ".virtualenv",
206                    "virtualenv",
207                    "env",
208                    ".env",
209                ]
210            ]
211        )
212    fs_watcher = WatchCond(
213        path_glob=watch_globs, ignore_glob=ignore_glob, action=None
214    )
215
216    watcher = Watcher([fs_watcher], force_polling=force_polling)
217
218    def serve():
219        server_address = (host, port)
220        httpd = http.server.ThreadingHTTPServer(
221            server_address, GalleryRequestHandler
222        )
223        print(f"Serving HTTP on http://{server_address[0]}:{server_address[1]}")
224        httpd.serve_forever()
225
226    server_thread = threading.Thread(target=serve, daemon=True)
227    server_thread.start()
228    livereload_thread = threading.Thread(
229        target=livereload_server.live_reloader,
230        args=(livereload_host, livereload_port),
231        daemon=True,
232    )
233    livereload_thread.start()
234    healthy = True
235    backoff = 1.0
236    last_error_str = None
237
238    GalleryRequestHandler.content = show_showcases()
239    changed_files = []
240    while True:
241        if healthy:
242            while True:
243                changed_files = watcher.changed()
244                if changed_files:
245                    print("Changes detected, reloading module...")
246                    break
247                sleep(0.1)
248        else:
249            # Poll for changes but with progressive sleep limit (backoff)
250            waited = 0.0
251            while waited < backoff:
252                changed_files = watcher.changed()
253                if changed_files:
254                    break
255                sleep(0.1)
256                waited += 0.1
257
258        # Remove the module and all its submodules so they reimport
259        prefix = module_name + "."
260        stale = [
261            key
262            for key in sys.modules
263            if key == module_name or key.startswith(prefix)
264        ]
265        for key in stale:
266            del sys.modules[key]
267
268        importlib.invalidate_caches()
269
270        try:
271            showcase_registry.clear()
272            env_registry.clear()
273            module = importlib.import_module(module_name)
274            GalleryRequestHandler.content.clear()
275            GalleryRequestHandler.content.extend(show_showcases())
276            healthy = True
277            backoff = 1.0
278            last_error_str = None
279
280            if changed_files:
281                livereload_server.reload_because(
282                    [f.path for f in changed_files]
283                )
284        except Exception as e:
285            healthy = False
286            current_error_str = f"Error re-importing module {module_name}: {e}"
287            if current_error_str != last_error_str:
288                print(current_error_str)
289                last_error_str = current_error_str
290            backoff = min(backoff * 2, 10.0)

class Settings: View Source

23class Settings:
24    resource_prefix = "/"
25    static_dir = Path.cwd() / "static"
26
27    @staticmethod
28    def set_static_dir(path: str) -> None:
29        Settings.static_dir = Path(path)

resource_prefix = '/'

static_dir = PosixPath('/home/runner/work/html-compose/html-compose/static')

@staticmethod

def set_static_dir(path: str) -> None: View Source

27    @staticmethod
28    def set_static_dir(path: str) -> None:
29        Settings.static_dir = Path(path)

WEB_MIMETYPES = {'.html': 'text/html', '.css': 'text/css', '.js': 'application/javascript', '.png': 'image/png', '.jpg': 'image/jpeg', '.gif': 'image/gif', '.svg': 'image/svg+xml'}

class GalleryRequestHandler(http.server.BaseHTTPRequestHandler): View Source

 44class GalleryRequestHandler(http.server.BaseHTTPRequestHandler):
 45    content: list[ShowcaseResult] = []
 46
 47    def _handle(self) -> None:
 48        parsed = urlparse(self.path)
 49        path = parsed.path
 50        query = parse_qs(parsed.query)
 51        content = GalleryRequestHandler.content
 52
 53        if path == "/" or path == "/index.html":
 54            self.send_response(200)
 55            self.send_header("Content-Type", "text/html; charset=utf-8")
 56            self.end_headers()
 57            self.wfile.write(app.main_route(content).encode("utf-8"))
 58            return
 59
 60        for c in content:
 61            if path == f"/showcase/{c.func_module}/{c.func_name}":
 62                # Filter on name if provided
 63                test_name = query.get("name", None)
 64                if test_name:
 65                    if c.name != test_name[0]:
 66                        continue
 67
 68                if not c.success:
 69                    self.send_response(500)
 70                    self.send_header("Content-Type", "text/plain")
 71                    self.end_headers()
 72                    self.wfile.write(f"Showcase broken: {c.result}".encode())
 73                    return
 74
 75                # Serve the showcase
 76                self.send_response(200)
 77                self.send_header("Content-Type", "text/html; charset=utf-8")
 78                self.end_headers()
 79                self.wfile.write(app.generate_showcase(c).encode("utf-8"))
 80                return
 81
 82        # Try to serve static files
 83        if self._serve_static(path):
 84            return
 85
 86        # Nothing matched
 87        self.send_response(404)
 88        self.end_headers()
 89
 90    def do_GET(self):
 91        # This method handles all GET requests
 92        self._handle()
 93
 94    def _serve_static(self, path: str) -> bool:
 95        if not Settings.static_dir:
 96            return False
 97        prefix = Settings.resource_prefix
 98        if path.startswith(prefix):
 99            normalized = path.removeprefix(prefix).lstrip("/")
100        else:
101            return False
102
103        # Collapse .. to prevent directory traversal without resolving symlinks
104        candidate = Path(os.path.normpath(Settings.static_dir / normalized))
105
106        # Not relative to static dir
107        if not candidate.is_relative_to(
108            Path(os.path.normpath(Settings.static_dir))
109        ):
110            return False
111
112        # Resolve symlinks for the actual file read
113        # as a developer server, we assume your symlinks are safe
114        candidate = candidate.resolve()
115
116        if not candidate.is_file():
117            return False
118
119        mime, _ = mimetypes.guess_type(candidate.name)
120        if not mime:
121            # Fall through if the mimetype db failed to identify
122            # basic web types
123            ext = candidate.suffix
124            # fallback to application/octet-stream if we really
125            # have no idea
126            mime = WEB_MIMETYPES.get(ext, "application/octet-stream")
127
128        self.send_response(200)
129        self.send_header("Content-Type", mime)
130        self.send_header("Content-Length", str(candidate.stat().st_size))
131        self.end_headers()
132
133        # Stream the file in chunks to avoid loading large files into memory
134        with candidate.open("rb") as f:
135            while True:
136                data = f.read(8192)
137                if not data:
138                    break
139                self.wfile.write(data)
140
141        return True

HTTP request handler base class.

The following explanation of HTTP serves to guide you through the code as well as to expose any misunderstandings I may have about HTTP (so you don't need to read the code to figure out I'm wrong :-).

HTTP (HyperText Transfer Protocol) is an extensible protocol on top of a reliable stream transport (e.g. TCP/IP). The protocol recognizes three parts to a request:

One line identifying the request type and path
An optional set of RFC-822-style headers
An optional data part

The headers and data are separated by a blank line.

The first line of the request has the form

where is a (case-sensitive) keyword such as GET or POST, is a string containing path information for the request, and should be the string "HTTP/1.0" or "HTTP/1.1". is encoded using the URL encoding scheme (using %xx to signify the ASCII character with hex code xx).

The specification specifies that lines are separated by CRLF but for compatibility with the widest range of clients recommends servers also handle LF. Similarly, whitespace in the request line is treated sensibly (allowing multiple spaces between components and allowing trailing whitespace).

Similarly, for output, lines ought to be separated by CRLF pairs but most clients grok LF characters just fine.

If the first line of the request has the form

(i.e. is left out) then this is assumed to be an HTTP 0.9 request; this form has no optional headers and data part and the reply consists of just the data.

The reply form of the HTTP 1.x protocol again has three parts:

One line giving the response code
An optional set of RFC-822-style headers
The data

Again, the headers and data are separated by a blank line.

The response code line has the form

where is the protocol version ("HTTP/1.0" or "HTTP/1.1"), is a 3-digit response code indicating success or failure of the request, and is an optional human-readable string explaining what the response code means.

This server parses the request and the headers, and then calls a function specific to the request type (). Specifically, a request SPAM will be handled by a method do_SPAM(). If no such method exists the server sends an error response to the client. If it exists, it is called with no arguments:

do_SPAM()

Note that the request name is case sensitive (i.e. SPAM and spam are different requests).

The various request details are stored in instance variables:

client_address is the client IP address in the form (host, port);
command, path and version are the broken-down request line;
headers is an instance of email.message.Message (or a derived class) containing the header information;
rfile is a file object open for reading positioned at the start of the optional input data part;
wfile is a file object open for writing.

IT IS IMPORTANT TO ADHERE TO THE PROTOCOL FOR WRITING!

The first thing to be written must be the response line. Then follow 0 or more header lines, then a blank line, and then the actual data (if any). The meaning of the header lines depends on the command executed by the server; in most cases, when data is returned, there should be at least one header line of the form

Content-type: /

where and should be registered MIME types, e.g. "text/html" or "text/plain".

content: list[html_compose.gallery.impl.ShowcaseResult] = []

def do_GET(self): View Source

90    def do_GET(self):
91        # This method handles all GET requests
92        self._handle()

def run( module_path: str | pathlib.Path, host: str, port: int, livereload_port: int, livereload_host: str, force_polling: bool = False, extra_paths: list[str] | None = None): View Source

144def run(
145    module_path: str | Path,
146    host: str,
147    port: int,
148    livereload_port: int,
149    livereload_host: str,
150    force_polling: bool = False,
151    extra_paths: list[str] | None = None,
152):
153    # Resolve to absolute before accessing parts to handle absolute file paths safely
154    # If the user passed an absolute path, we inject its directory into sys.path
155    mp_raw = Path(module_path)
156    if mp_raw.is_absolute():
157        if str(mp_raw.parent) not in sys.path:
158            sys.path.insert(0, str(mp_raw.parent))
159        mp = Path(mp_raw.name)
160    else:
161        mp = mp_raw
162
163    if mp.suffix == ".py":
164        mp = mp.with_suffix("")
165
166    showcase_registry.clear()
167    module_name = ".".join(mp.parts)
168
169    if module_name in sys.modules:
170        del sys.modules[module_name]
171
172    importlib.invalidate_caches()
173
174    app.Settings.livereload_host = livereload_host
175    app.Settings.livereload_port = livereload_port
176
177    try:
178        module = importlib.import_module(module_name)
179    except Exception as e:
180        print(f"Error importing module {module_name}: {e}")
181        raise
182    assert module is not None
183
184    watch_globs = []
185    ignore_glob = []
186    if extra_paths:
187        for ep in extra_paths:
188            watch_globs.append(ep)
189    if hasattr(module, "__path__"):
190        pathlist = module.__path__
191    elif hasattr(module, "__file__"):
192        pathlist = [os.path.dirname(cast(str, module.__file__))]
193    else:
194        pathlist = []
195        print("Warning: could not determine module path for file watching")
196
197    for watch_path in pathlist:
198        loaded_mod_path = Path(watch_path)
199        watch_globs.extend([f"{loaded_mod_path / '**/*.py'}"])
200        ignore_glob.extend(
201            [
202                f"{loaded_mod_path / venv}"
203                for venv in [
204                    "venv",
205                    ".venv",
206                    ".virtualenv",
207                    "virtualenv",
208                    "env",
209                    ".env",
210                ]
211            ]
212        )
213    fs_watcher = WatchCond(
214        path_glob=watch_globs, ignore_glob=ignore_glob, action=None
215    )
216
217    watcher = Watcher([fs_watcher], force_polling=force_polling)
218
219    def serve():
220        server_address = (host, port)
221        httpd = http.server.ThreadingHTTPServer(
222            server_address, GalleryRequestHandler
223        )
224        print(f"Serving HTTP on http://{server_address[0]}:{server_address[1]}")
225        httpd.serve_forever()
226
227    server_thread = threading.Thread(target=serve, daemon=True)
228    server_thread.start()
229    livereload_thread = threading.Thread(
230        target=livereload_server.live_reloader,
231        args=(livereload_host, livereload_port),
232        daemon=True,
233    )
234    livereload_thread.start()
235    healthy = True
236    backoff = 1.0
237    last_error_str = None
238
239    GalleryRequestHandler.content = show_showcases()
240    changed_files = []
241    while True:
242        if healthy:
243            while True:
244                changed_files = watcher.changed()
245                if changed_files:
246                    print("Changes detected, reloading module...")
247                    break
248                sleep(0.1)
249        else:
250            # Poll for changes but with progressive sleep limit (backoff)
251            waited = 0.0
252            while waited < backoff:
253                changed_files = watcher.changed()
254                if changed_files:
255                    break
256                sleep(0.1)
257                waited += 0.1
258
259        # Remove the module and all its submodules so they reimport
260        prefix = module_name + "."
261        stale = [
262            key
263            for key in sys.modules
264            if key == module_name or key.startswith(prefix)
265        ]
266        for key in stale:
267            del sys.modules[key]
268
269        importlib.invalidate_caches()
270
271        try:
272            showcase_registry.clear()
273            env_registry.clear()
274            module = importlib.import_module(module_name)
275            GalleryRequestHandler.content.clear()
276            GalleryRequestHandler.content.extend(show_showcases())
277            healthy = True
278            backoff = 1.0
279            last_error_str = None
280
281            if changed_files:
282                livereload_server.reload_because(
283                    [f.path for f in changed_files]
284                )
285        except Exception as e:
286            healthy = False
287            current_error_str = f"Error re-importing module {module_name}: {e}"
288            if current_error_str != last_error_str:
289                print(current_error_str)
290                last_error_str = current_error_str
291            backoff = min(backoff * 2, 10.0)