Skip to content
Back to top

# Source code for xonsh.webconfig.main

    #!/usr/bin/env python3
    import contextlib
    import json
    import os
    import socketserver
    import string
    import sys
    import typing as tp
    from email.message import EmailMessage
    from http import HTTPStatus, server
    from pathlib import Path
    from pprint import pprint
    from urllib import parse

    from xonsh.built_ins import XSH
    from xonsh.webconfig import tags as t
    from xonsh.webconfig.file_writes import insert_into_xonshrc
    from xonsh.webconfig.routes import Routes

    RENDERERS: list[tp.Callable] = []

    class XonshConfigHTTPRequestHandler(server.SimpleHTTPRequestHandler):
        def __init__(self, *args, **kwargs):
            kwargs.setdefault("directory", os.path.dirname(__file__))
            super().__init__(*args, **kwargs)

        def _write_headers(self, *headers: "tuple[str, str]"):
            for name, val in headers:
                self.send_header(name, val)
            self.end_headers()

        def _write_data(self, data: "bytes|dict|str"):
            if isinstance(data, bytes):
                content_type = "text/html"
            elif isinstance(data, dict):
                content_type = "application/json"
                data = json.dumps(data).encode()
            else:
                content_type = "text/html"
                data = str(data).encode()
            self._write_headers(
                ("Content-type", content_type),
                ("Content-Length", str(len(data))),
            )
            self.wfile.write(data)
            self.wfile.flush()

        def _send(
            self,
            data: "bytes|dict|str|None" = None,
            status: "None|int" = None,
            redirect: "str|None" = None,
        ):
            status = status or (HTTPStatus.FOUND if redirect else HTTPStatus.OK)
            self.send_response(status)
            if data:
                self._write_data(data)
            elif redirect:
                self._write_headers(
                    ("Location", redirect),
                )

        def _read(self):
            content_len = int(self.headers.get("content-length", 0))
            return self.rfile.read(content_len)

        def render_get(self, route):
            try:
                webconfig = Path(__file__).parent
            except Exception:
                # in case of thread missing __file__ definition
                webconfig = Path.cwd()
            path = webconfig / "index.html"
            tmpl = string.Template(path.read_text())
            navlinks = t.to_str(route.get_nav_links())
            msgs = t.to_str(route.get_err_msgs())
            body = t.to_str(route.get())  # type: ignore
            data = tmpl.substitute(navlinks=navlinks, body=msgs + body)
            return self._send(data)

        def _get_route(self, method: str):
            url = parse.urlparse(self.path)
            route_cls = Routes.registry.get(url.path)
            if route_cls and hasattr(route_cls, method):
                params = parse.parse_qs(url.query)
                return route_cls(url=url, params=params, xsh=XSH)

        def do_GET(self) -> None:
            route = self._get_route("get")
            if route is not None:
                return self.render_get(route)
            return super().do_GET()

        def _read_form(self):
            msg = EmailMessage()
            msg["Content-Type"] = self.headers.get("content-type")
            if msg.get_content_type() == "application/x-www-form-urlencoded":
                data = parse.parse_qs(self._read(), keep_blank_values=True)
                for name, values in data.items():
                    value = None
                    if isinstance(name, bytes):
                        name = name.decode(encoding="utf-8")
                    if isinstance(values, list) and values:
                        if isinstance(values[0], bytes):
                            value = values[0].decode(encoding="utf-8")
                    yield name, value

        def do_POST(self):
            """Reads post request body"""
            route = self._get_route("post")
            if route is not None:
                # redirect after form submission
                data = dict(self._read_form())

                new_route = route.post(data) or route
                return self._send(redirect=new_route.path)
            post_body = self._read()
            config = json.loads(post_body)
            print("Web Config Values:")
            pprint(config)
            fname = insert_into_xonshrc(config)
            print("Wrote out to " + fname)
            self._send(b"received post request:<br>" + post_body)

    def bind_server_to(
        port: int = 8421, handler_cls=XonshConfigHTTPRequestHandler, browser=False
    ):
        cls = socketserver.TCPServer
        # cls = socketserver.ThreadingTCPServer  # required ctrl+c twice ?

        cls.allow_reuse_address = True

        while port <= 9310:
            try:
                cls.allow_reuse_address = True

                httpd = cls(("", port), handler_cls)
                url = f"http://localhost:{port}"
                print(f"Web config started at '{url}'. Hit Ctrl+C to stop.")
                if browser:
                    import webbrowser

                    webbrowser.open(url)
                return httpd
            except OSError:
                type, value = sys.exc_info()[:2]
                if "Address already in use" not in str(value):
                    raise
            except KeyboardInterrupt:
                break
            port += 1

    def serve(browser=False):
        httpd = bind_server_to(browser=browser)

        with contextlib.suppress(KeyboardInterrupt):
            with httpd:
                httpd.serve_forever()

    def main(browser=False):
        """standalone entry point for webconfig."""
        from xonsh.main import setup

        setup()
        serve(browser)

    if __name__ == "__main__":
        # watchexec -r -e py -- python -m xonsh.webconfig --no-browser
        main()