git.mar77i.info Git - hublib/blob - hub/app.py

   1 import socket
   2 import sys
   3 from argparse import ArgumentParser
   4 from itertools import chain
   5 from pathlib import Path
   6 from secrets import token_urlsafe
   7 from subprocess import run
   8 from traceback import print_exception
   9 from urllib.parse import urlunsplit
  10
  11 from falcon.asgi import App as FalconApp
  12 from falcon.constants import MEDIA_HTML
  13 from jinja2 import Environment, FileSystemLoader, select_autoescape
  14 from uvicorn import Config, Server
  15
  16 from .hubapp import HubApp, RootApp
  17
  18
  19 class App(FalconApp):
  20     def __init__(self, secret, **kwargs):
  21         kwargs.setdefault("media_type", MEDIA_HTML)
  22         super().__init__(**kwargs)
  23         self.base_dir = Path(__file__).parents[1] / "webroot"
  24         self.env = Environment(
  25             loader=FileSystemLoader(self.base_dir),
  26             autoescape=select_autoescape,
  27             extensions=["hub.utils.StaticTag"],
  28         )
  29         self.secret = secret or token_urlsafe(64)
  30         self.hubapps = {}
  31         RootApp(self, self.base_dir)
  32         for base_dir in self.base_dir.iterdir():
  33             if not base_dir.is_dir() or HubApp.is_ignored_filename(base_dir):
  34                 continue
  35             HubApp(self, base_dir)
  36         self.add_error_handler(Exception, self.print_exception)
  37
  38     def get_hubapp_by_name(self, name):
  39         if name == "root":
  40             name = ""
  41         return self.hubapps[name]
  42
  43     async def print_exception(self, req, resp, ex, params):
  44         print_exception(*sys.exc_info())
  45
  46
  47 class HubServer(Server):
  48     def __init__(self, *args, browser, **kwargs):
  49         self.browser = browser
  50         super().__init__(*args, **kwargs)
  51
  52     async def startup(self, sockets: list[socket.socket] | None = None) -> None:
  53         await super().startup(sockets)
  54         config = self.config
  55         protocol_name = "https" if config.ssl else "http"
  56         host = "0.0.0.0" if config.host is None else config.host
  57         if ":" in host:
  58             # It's an IPv6 address.
  59             host = f"[{host.rstrip(']').lstrip('[')}]"
  60
  61         port = config.port
  62         if port == 0:
  63             try:
  64                 port = next(
  65                     chain.from_iterable(
  66                         (server.sockets for server in getattr(self, "servers", ()))
  67                     )
  68                 ).getsockname()[1]
  69             except StopIteration:
  70                 pass
  71         if {"http": 80, "https": 443}[protocol_name] != port:
  72             host = f"{host}:{port}"
  73         app = config.loaded_app.app
  74         print("Secret:", app.secret)
  75         for key, value in app.root_app.files_per_uris.items():
  76             if Path(value.path.name).stem == "index.html":
  77                 url = urlunsplit((protocol_name, host, key, "", ""))
  78                 print("URL:", url)
  79                 if self.browser:
  80                     run([self.browser, url])
  81
  82
  83 app: App
  84
  85
  86 async def main():
  87     global app
  88     ap = ArgumentParser()
  89     ap.add_argument("--secret")
  90     ap.add_argument("--browser", default="firefox")
  91     args = ap.parse_args()
  92     app = App(args.secret)
  93     config = Config("hub.app:app", port=5000, log_level="info")
  94     config.setup_event_loop()
  95     hs = HubServer(config, browser=args.browser)
  96     await hs.serve()