from argparse import ArgumentParser from itertools import chain from secrets import token_urlsafe import socket from subprocess import run import sys from traceback import print_exception from typing import List, Optional from urllib.parse import urlunsplit from falcon.asgi import App from falcon.constants import MEDIA_HTML from uvicorn import Config, Server from .staticresource import StaticResource from .hub import Hub class HubApp(App): def __init__(self, secret, browser, hubapp, *args, **kwargs): kwargs.setdefault("media_type", MEDIA_HTML) super().__init__(*args, **kwargs) self.secret = secret or token_urlsafe(64) self.hub = Hub(self.secret) self.sr = StaticResource(self.secret, hubapp, self.hub.master_ws_uri) self.browser = browser self.hub.add_routes(self) self.sr.add_routes(self) self.add_error_handler(Exception, self.print_exception) async def print_exception(self, req, resp, ex, params): print_exception(*sys.exc_info()) def get_app(): ap = ArgumentParser() ap.add_argument("--secret") ap.add_argument("--browser", default="firefox") ap.add_argument("--hubapp", default="first") args = ap.parse_args() return HubApp(args.secret, args.browser, args.hubapp) class HubServer(Server): async def startup(self, sockets: Optional[List[socket.socket]] = None) -> None: await super().startup(sockets) config = self.config protocol_name = "https" if config.ssl else "http" host = "0.0.0.0" if config.host is None else config.host if ":" in host: # It's an IPv6 address. host = f"[{host.rstrip(']').lstrip('[')}]" port = config.port if port == 0: try: port = next( chain.from_iterable((server.sockets for server in getattr(self, "servers", ()))) ).getsockname()[1] except StopIteration: pass if {"http": 80, "https": 443}[protocol_name] != port: host = f"{host}:{port}" app = config.loaded_app.app print("master_uri", app.sr.master_uri) master_url = urlunsplit((protocol_name, host, app.sr.master_uri, "", "")) print("secret:", app.secret) if not app.browser: print("master url", master_url) else: run([app.browser, master_url]) def main(): HubServer( Config("hub.app:get_app", factory=True, port=5000, log_level="info") ).run()