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         return self.hubapps[name]
  40
  41     async def print_exception(self, req, resp, ex, params):
  42         print_exception(*sys.exc_info())
  43
  44
  45 class HubServer(Server):
  46     def __init__(self, *args, browser, **kwargs):
  47         self.browser = browser
  48         super().__init__(*args, **kwargs)
  49
  50     async def startup(self, sockets: list[socket.socket] | None = None) -> None:
  51         await super().startup(sockets)
  52         config = self.config
  53         protocol_name = "https" if config.ssl else "http"
  54         host = "0.0.0.0" if config.host is None else config.host
  55         if ":" in host:
  56             # It's an IPv6 address.
  57             host = f"[{host.rstrip(']').lstrip('[')}]"
  58
  59         port = config.port
  60         if port == 0:
  61             try:
  62                 port = next(
  63                     chain.from_iterable(
  64                         (server.sockets for server in getattr(self, "servers", ()))
  65                     )
  66                 ).getsockname()[1]
  67             except StopIteration:
  68                 pass
  69         if {"http": 80, "https": 443}[protocol_name] != port:
  70             host = f"{host}:{port}"
  71         app = config.loaded_app.app
  72         print("Secret:", app.secret)
  73         for key, value in app.hubapps["root"].files_per_uris.items():
  74             if Path(value.path.name).stem == "index.html":
  75                 url = urlunsplit((protocol_name, host, key, "", ""))
  76                 print("URL:", url)
  77                 if self.browser:
  78                     run([self.browser, url])
  79
  80
  81 app: App
  82
  83
  84 async def main():
  85     global app
  86     ap = ArgumentParser()
  87     ap.add_argument("--secret")
  88     ap.add_argument("--browser", default="firefox")
  89     args = ap.parse_args()
  90     app = App(args.secret)
  91     config = Config("hub.app:app", port=5000, log_level="info")
  92     config.setup_event_loop()
  93     hs = HubServer(config, browser=args.browser)
  94     await hs.serve()