Reference

class App(ViewApp):
    """Public view.py app object."""

    def __init__(self, config: Config) -> None:
        """
        Args:
            config: Configuration object to be used. Automatically generated by `new_app`.
        """
        supply_parsers(self)
        self.config = config
        self._set_dev_state(config.dev)
        self._manual_routes: list[Route] = []
        self.routes: list[Route] = []
        self.loaded: bool = False
        self.running = False
        self._docs: DocsType = {}
        self.loaded_routes: list[Route] = []

        Service.log.setLevel(
            config.log.level
            if not isinstance(config.log.level, str)
            else config.log.level.upper()
        )

        if config.dev:
            if os.environ.get("VIEW_PROD") is not None:
                Service.warning("VIEW_PROD is set but dev is set to true")

            format_warnings()
            weakref.finalize(self, self._finalize)

            if config.log.pretty_tracebacks:
                install(show_locals=True)

            rich_handler = sys.excepthook

            def _hook(tp: type[B], value: B, traceback: Traceback) -> None:
                rich_handler(tp, value, traceback)
                os.environ["_VIEW_CANCEL_FINALIZERS"] = "1"

                if isinstance(value, ViewError):
                    if value.hint:
                        print(value.hint)

            sys.excepthook = _hook
            with suppress(UnsupportedOperation):
                faulthandler.enable()
        else:
            os.environ["VIEW_PROD"] = "1"

        if config.log.level == "debug":
            enable_debug()

        self.running = False

    def _finalize(self) -> None:
        if os.environ.get("_VIEW_CANCEL_FINALIZERS"):
            return

        if self.loaded:
            return

        warnings.warn(
            "load() was never called (did you forget to start the app?)"
        )
        split = self.config.app.app_path.split(":", maxsplit=1)

        if len(split) != 2:
            return

        app_name = split[1]

        print(
            make_hint(
                "Add this to your code",
                split[0],
                line=-1,
                prepend=f"\n{app_name}.run()",
            )
        )

    def _push_route(self, route: Route) -> None:
        if route in self._manual_routes:
            return

        self._manual_routes.append(route)

    def _method_wrapper(
        self,
        path: str,
        doc: str | None,
        target: Callable[..., Any],
        # i dont really feel like typing this properly
    ) -> Callable[[RouteOrCallable], Route]:
        def inner(route: RouteOrCallable) -> Route:
            new_route = target(path, doc)(route)
            self._push_route(new_route)
            return new_route

        return inner

    def get(self, path: str, *, doc: str | None = None):
        """Set a GET route."""
        return self._method_wrapper(path, doc, get)

    def post(self, path: str, *, doc: str | None = None):
        """Set a POST route."""
        return self._method_wrapper(path, doc, post)

    def delete(self, path: str, *, doc: str | None = None):
        """Set a DELETE route."""
        return self._method_wrapper(path, doc, delete)

    def patch(self, path: str, *, doc: str | None = None):
        """Set a PATCH route."""
        return self._method_wrapper(path, doc, patch)

    def put(self, path: str, *, doc: str | None = None):
        """Set a PUT route."""
        return self._method_wrapper(path, doc, put)

    def options(self, path: str, *, doc: str | None = None):
        """Set a OPTIONS route."""
        return self._method_wrapper(path, doc, options)

    def query(
        self,
        name: str,
        *tps: type[V],
        doc: str | None = None,
        default: V | None | _NoDefaultType = _NoDefault,
    ):
        """Set a query parameter.

        Args:
            name: Name of the parameter.
            tps: Types that can be passed to the server. If empty, any is used.
            doc: Description of this query parameter.
            default: Default value to be used if not supplied.
        """

        def inner(func: RouteOrCallable) -> Route:
            route = query_impl(name, *tps, doc=doc, default=default)(func)
            self._push_route(route)
            return route

        return inner

    def body(
        self,
        name: str,
        *tps: type[V],
        doc: str | None = None,
        default: V | None | _NoDefaultType = _NoDefault,
    ):
        """Set a body parameter.

        Args:
            name: Name of the parameter.
            tps: Types that can be passed to the server. If empty, any is used.
            doc: Description of this body parameter.
            default: Default value to be used if not supplied.
        """

        def inner(func: RouteOrCallable) -> Route:
            route = body_impl(name, *tps, doc=doc, default=default)(func)
            self._push_route(route)
            return route

        return inner

    async def _app(self, scope, receive, send) -> None:
        return await self.asgi_app_entry(scope, receive, send)

    def load(self, routes: list[Route] | None = None) -> None:
        """Load the app. This is automatically called most of the time and should only be called manually during manual loading.

        Args:
            routes: Routes to load into the app.
        """
        if self.loaded:
            if routes:
                finalize(routes, self)
            Internal.warning("load called twice")
            return

        if self.config.app.loader == "filesystem":
            if routes:
                warnings.warn(_ROUTES_WARN_MSG)
            load_fs(self, self.config.app.loader_path)
        elif self.config.app.loader == "simple":
            if routes:
                warnings.warn(_ROUTES_WARN_MSG)
            load_simple(self, self.config.app.loader_path)
        else:
            finalize([*(routes or ()), *self._manual_routes], self)

        self.loaded = True

        for r in self.loaded_routes:
            if not r.path:
                continue

            body = {}
            query = {}

            for i in r.inputs:
                target = body if i.is_body else query
                target[i.name] = InputDoc(
                    i.doc or "No description provided.", i.tp, i.default
                )

            self._docs[(r.method.name, r.path)] = RouteDoc(
                r.doc or "No description provided.", body, query
            )

    async def _spawn(self, coro: Coroutine[Any, Any, Any]):
        Internal.info(f"using event loop: {asyncio.get_event_loop()}")
        Internal.info(f"spawning {coro}")

        task = asyncio.create_task(coro)
        if self.config.log.hijack:
            if self.config.server.backend == "uvicorn":
                Internal.info("hijacking uvicorn")
                for log in (
                    logging.getLogger("uvicorn.error"),
                    logging.getLogger("uvicorn.access"),
                ):
                    log.addFilter(UvicornHijack())
            else:
                Internal.info("hijacking hypercorn")

        if self.config.log.fancy:
            if not self.config.log.hijack:
                raise ValueError("hijack must be enabled for fancy mode")

            enter_server()

        self.running = True
        Internal.debug("here we go!")
        await task
        self.running = False

        if self.config.log.fancy:
            exit_server()

        Internal.info("server closed")

    def _run(self, start_target: Callable[..., Any] | None = None) -> Any:
        self.load()
        Internal.info("starting server!")
        server = self.config.server.backend
        uvloop_enabled = False

        if self.config.app.uvloop is True:
            uvloop = attempt_import("uvloop")
            uvloop.install()
            uvloop_enabled = True
        elif self.config.app.uvloop == "decide":
            with suppress(MissingLibraryError):
                uvloop = attempt_import("uvloop")
                uvloop.install()
                uvloop_enabled = True

        start = start_target or asyncio.run

        if server == "uvicorn":
            uvicorn = attempt_import("uvicorn")

            config = uvicorn.Config(
                self._app,
                port=self.config.server.port,
                host=str(self.config.server.host),
                log_level="debug" if self.config.dev else "info",
                lifespan="on",
                factory=False,
                interface="asgi3",
                loop="uvloop" if uvloop_enabled else "asyncio",
                **self.config.server.extra_args,
            )
            server = uvicorn.Server(config)

            return start(self._spawn(server.serve()))

        elif server == "hypercorn":
            hypercorn = attempt_import("hypercorn")
            conf = hypercorn.Config()
            conf.loglevel = "debug" if self.config.dev else "info"
            conf.bind = [
                f"{self.config.server.host}:{self.config.server.port}",
            ]

            for k, v in self.config.server.extra_args.items():
                setattr(conf, k, v)

            return start(
                importlib.import_module("hypercorn.asyncio").serve(
                    self._app, conf
                )
            )
        else:
            raise NotImplementedError("viewserver is not implemented yet")

    def run(self, *, fancy: bool | None = None) -> None:
        """Run the app."""
        if fancy is not None:
            self.config.log.fancy = fancy

        frame = inspect.currentframe()
        assert frame, "failed to get frame"
        assert frame.f_back, "frame has no f_back"

        back = frame.f_back
        base = os.path.basename(back.f_code.co_filename)
        app_path = self.config.app.app_path
        fname = app_path.split(":")[0]
        if base != fname:
            warnings.warn(
                f"ran app from {base}, but app path is {fname} in config",
            )

        if (not os.environ.get("_VIEW_RUN")) and (
            back.f_globals.get("__name__") == "__main__"
        ):
            self._run()
        else:
            Internal.info("called run, but env or scope prevented startup")

    def run_threaded(self, *, daemon: bool = True) -> Thread:
        """Run the app in a thread."""
        thread = Thread(target=self._run, daemon=daemon)
        thread.start()
        return thread

    def run_async(
        self,
        loop: asyncio.AbstractEventLoop | None = None,
    ) -> None:
        """Run the app in an event loop."""
        self._run((loop or asyncio.get_event_loop()).run_until_complete)

    def run_task(
        self,
        loop: asyncio.AbstractEventLoop | None = None,
    ) -> asyncio.Task[None]:
        """Run the app as a task."""
        return self._run((loop or asyncio.get_event_loop()).create_task)

    start = run

    def __repr__(self) -> str:
        return f"App(config={self.config!r})"

    @asynccontextmanager
    async def test(self):
        """Open the testing context."""
        self.load()
        ctx = TestingContext(self.asgi_app_entry)
        try:
            yield ctx
        finally:
            await ctx.stop()

    @overload
    def docs(self, file: None = None) -> str:
        ...

    @overload
    def docs(self, file: TextIO) -> None:
        ...

    @overload
    def docs(
        self,
        file: Path,
        *,
        encoding: str = "utf-8",
        overwrite: bool = True,
    ) -> None:
        ...

    @overload
    def docs(
        self,
        file: str,
        *,
        encoding: str = "utf-8",
        overwrite: bool = True,
    ) -> None:
        ...

    def docs(
        self,
        file: str | TextIO | Path | None = None,
        *,
        encoding: str = "utf-8",
        overwrite: bool = True,
    ) -> str | None:
        """Generate documentation for the app."""
        self.load()
        md = markdown_docs(self._docs)

        if not file:
            return md

        if isinstance(file, str):
            if not overwrite:
                Path(file).write_text(md, encoding=encoding)
            else:
                with open(file, "w", encoding=encoding) as f:
                    f.write(md)
        elif isinstance(file, Path):
            if overwrite:
                with open(file, "w", encoding=encoding) as f:
                    f.write(md)
            else:
                file.write_text(md)
        else:
            file.write(md)

class AppNotFoundError(ViewError, FileNotFoundError):
    """Couldn't find the app from the given path."""

    ...

class EnvironmentError(ViewError):
    """An environment variable is missing."""

    ...

class InvalidBodyError(ViewError):
    """The specified type cannot be used as a view body."""

    ...

class LoaderWarning(ViewWarning):
    """A warning from the loader."""

    ...

class MissingLibraryError(ViewError):
    """A library is not installed."""

    ...

class MistakeError(ViewError):
    """The user made a mistake."""

    ...

class NotLoadedWarning(ViewWarning):
    """load() was never called"""

    ...

class ViewError(Exception):
    """Base class for exceptions in view.py"""

    def __init__(
        self,
        *args: object,
        hint: RenderableType | None = None,
    ) -> None:
        self.hint = hint
        super().__init__(*args)

class ViewWarning(UserWarning):
    """Base class for all warnings in view.py"""

    ...

class HTML(Response[str]):
    """HTML response wrapper."""

    def __init__(
        self,
        body: TextIO | str | Path | DOMNode,
        status: int = 200,
        headers: dict[str, str] | None = None,
    ) -> None:
        parsed_body = ""

        if isinstance(body, Path):
            parsed_body = body.read_text()
        elif isinstance(body, str):
            parsed_body = body
        elif isinstance(body, DOMNode):
            parsed_body = body.data
        else:
            try:
                parsed_body = body.read()
            except AttributeError:
                raise TypeError(
                    f"expected TextIO, str, Path, or DOMNode, not {type(body)}",  # noqa
                ) from None

        super().__init__(parsed_body, status, headers)
        self._raw_headers.append((b"content-type", b"text/html"))

class Response(Generic[T]):
    """Wrapper for responses."""

    def __init__(
        self,
        body: T | None = None,
        status: int = 200,
        headers: dict[str, str] | None = None,
        *,
        body_translate: BodyTranslateStrategy | None = _Find,
    ) -> None:
        self.body = body
        self.status = status
        self.headers = headers or {}
        self._raw_headers: list[tuple[bytes, bytes]] = []
        if body_translate:
            self.translate = body_translate
        else:
            self.translate = (
                "str" if not hasattr(body, "__view_result__") else "result"
            )

    def cookie(
        self,
        key: str,
        value: str = "",
        *,
        max_age: int | None = None,
        expires: int | DateTime | None = None,
        path: str | None = None,
        domain: str | None = None,
        http_only: bool = False,
        same_site: SameSite = "lax",
        partitioned: bool = False,
        secure: bool = False,
    ) -> None:
        """Set a cookie.

        Args:
            key: Cookie name.
            value: Cookie value.
            max_age: Max age of the cookies.
            expires: When the cookie expires.
            domain: Domain the cookie is valid at.
            http_only: Whether the cookie should be HTTP only.
            same_site: SameSite setting for the cookie.
            partitioned: Whether to tie it to the top level site.
            secure: Whether the cookie should enforce HTTPS."""
        cookie_str = f"{key}={value}; SameSite={same_site}".encode()

        if expires:
            dt = (
                expires
                if isinstance(expires, DateTime)
                else DateTime.fromtimestamp(expires)
            )
            ts = timestamp(dt)
            cookie_str += f"; Expires={ts}".encode()

        if http_only:
            cookie_str += b"; HttpOnly"

        if domain:
            cookie_str += f"; Domain={domain}".encode()

        if max_age:
            cookie_str += f"; Max-Age={max_age}".encode()

        if partitioned:
            cookie_str += b"; Partitioned"

        if secure:
            cookie_str += b"; Secure"

        if path:
            cookie_str += f"; Path={path}".encode()

        self._raw_headers.append((b"Set-Cookie", cookie_str))

    def _build_headers(self) -> tuple[tuple[bytes, bytes], ...]:
        headers: list[tuple[bytes, bytes]] = [*self._raw_headers]

        for k, v in self.headers.items():
            headers.append((k.encode(), v.encode()))

        return tuple(headers)

    def __view_result__(self):
        body: str = ""
        if self.translate == "str":
            body = str(self.body)
        elif self.translate == "repr":
            body = repr(self.body)
        else:
            view_result = getattr(self.body, "__view_result__", None)

            if not view_result:
                raise AttributeError(f"{self.body!r} has no __view_result__")

            body_res = view_result()
            if isinstance(body_res, str):
                body = body_res
            else:
                for i in body_res:
                    if isinstance(i, str):
                        body = i

        return body, self.status, self._build_headers()

@dataclass
class Route(Generic[P, T], LoadChecker):
    """Standard Route Wrapper"""

    func: Callable[P, T]
    path: str | None
    method: Method
    inputs: list[RouteInput]
    doc: str | None = None
    cache_rate: int = -1
    errors: dict[int, ViewRoute] | None = None
    extra_types: dict[str, Any] = field(default_factory=dict)
    parts: list[str | Part[Any]] = field(default_factory=list)

    def error(self, status_code: int):
        def wrapper(handler: ViewRoute):
            if not self.errors:
                self.errors = {}

            self.errors[status_code] = handler
            return handler

        return wrapper

    def __repr__(self):
        return f"Route({self.method.name}(\"{self.path or '/???'}\"))"  # noqa

    __str__ = __repr__

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> T:
        return self.func(*args, **kwargs)