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

   1 from argparse import ArgumentParser
   2 from itertools import chain
   3 from secrets import token_urlsafe
   4 import socket
   5 from subprocess import run
   6 import sys
   7 from traceback import print_exception
   8 from typing import List, Optional
   9 from urllib.parse import urlunsplit
  10
  11 from falcon.asgi import App
  12 from falcon.constants import MEDIA_HTML
  13 from uvicorn import Config, Server
  14
  15 from .staticresource import StaticResource
  16 from .hub import Hub
  17
  18
  19 class HubApp(App):
  20     def __init__(self, secret, browser, hubapp, *args, **kwargs):
  21         kwargs.setdefault("media_type", MEDIA_HTML)
  22         super().__init__(*args, **kwargs)
  23         self.secret = secret or token_urlsafe(64)
  24         self.hub = Hub(self.secret)
  25         self.sr = StaticResource(self.secret, hubapp)
  26         self.browser = browser
  27         self.hub.add_routes(self)
  28         self.sr.add_routes(self)
  29         self.hub.update_context_vars(self.sr.context_vars)
  30         self.add_error_handler(Exception, self.print_exception)
  31
  32     async def print_exception(self, req, resp, ex, params):
  33         print_exception(*sys.exc_info())
  34
  35
  36 def get_app():
  37     ap = ArgumentParser()
  38     ap.add_argument("--secret")
  39     ap.add_argument("--browser", default="firefox")
  40     ap.add_argument("--hubapp", default="first")
  41     args = ap.parse_args()
  42     return HubApp(args.secret, args.browser, args.hubapp)
  43
  44
  45 class HubServer(Server):
  46     async def startup(self, sockets: Optional[List[socket.socket]] = None) -> None:
  47         await super().startup(sockets)
  48         config = self.config
  49         protocol_name = "https" if config.ssl else "http"
  50         host = "0.0.0.0" if config.host is None else config.host
  51         if ":" in host:
  52             # It's an IPv6 address.
  53             host = f"[{host.rstrip(']').lstrip('[')}]"
  54
  55         port = config.port
  56         if port == 0:
  57             try:
  58                 port = next(
  59                     chain.from_iterable((server.sockets for server in getattr(self, "servers", ())))
  60                 ).getsockname()[1]
  61             except StopIteration:
  62                 pass
  63         if {"http": 80, "https": 443}[protocol_name] != port:
  64             host = f"{host}:{port}"
  65         app = config.loaded_app.app
  66         print("master_uri", app.sr.master_uri)
  67         master_url = urlunsplit((protocol_name, host, app.sr.master_uri, "", ""))
  68         print("secret:", app.secret)
  69         if not app.browser:
  70             print("master url", master_url)
  71         else:
  72             run([app.browser, master_url])
  73
  74
  75 def main():
  76     HubServer(
  77         Config("hub.app:get_app", factory=True, port=5000, log_level="info")
  78     ).run()