Skip to content

SSR

Http class to create an emuleo app

Arguments

pages: dict[str, Path] Dictionary of routes and their corresponding pages server: dict[str, Path] Dictionary of routes and their corresponding server files static: dict[str, Path] Dictionary of routes and their corresponding static files scripts: dict[str, Path] Dictionary of routes and their corresponding scripts styles: dict[str, Path] Dictionary of routes and their corresponding styles

Attributes:

Name Type Description
pages dict[str, Path]

Dictionary of routes and their corresponding pages

server dict[str, Path]

Dictionary of routes and their corresponding server files

static dict[str, Path]

Dictionary of routes and their corresponding static files

scripts dict[str, Path]

Dictionary of routes and their corresponding scripts

styles dict[str, Path]

Dictionary of routes and their corresponding styles

Source code in emuleo/http.py
class SSR:
    """
    Http class to create an emuleo app

    Arguments
    ---------
    pages: dict[str, Path]
        Dictionary of routes and their corresponding pages
    server: dict[str, Path]
        Dictionary of routes and their corresponding server files
    static: dict[str, Path]
        Dictionary of routes and their corresponding static files
    scripts: dict[str, Path]
        Dictionary of routes and their corresponding scripts
    styles: dict[str, Path]
        Dictionary of routes and their corresponding styles

    Attributes
    ----------
    pages: dict[str, Path]
        Dictionary of routes and their corresponding pages
    server: dict[str, Path]
        Dictionary of routes and their corresponding server files
    static: dict[str, Path]
        Dictionary of routes and their corresponding static files
    scripts: dict[str, Path]
        Dictionary of routes and their corresponding scripts
    styles: dict[str, Path]
        Dictionary of routes and their corresponding styles
    """

    def __init__(
        self,
        pages: dict[str, Path],
        server: dict[str, Path],
        static: dict[str, Path],
        scripts: dict[str, Path],
        styles: dict[str, Path],
    ) -> None:
        self.pages = pages
        self.server = server
        self.static = static
        self.scripts = scripts
        self.styles = styles

    async def __call__(
        self,
        scope: dict[str, t.Any],
        receive: Callable[..., t.Any],
        send: Callable[..., t.Any],
    ) -> None:
        """
        The main function of the app

        Arguments
        ---------
        scope: dict[str, typing.Any]
            The scope of the request
        receive: collections.abc.Callable[..., t.Any]
            The receive function
        send: collections.abc.Callable[..., t.Any]
            The send function

        Returns
        -------
        None

        Raises
        ------
        Exception
            If an error occurs

        Notes
        -----
        This function is called by uvicorn
        """
        assert scope["type"] == "http"
        console.print(
            f"[#0EA5E9]🔗 {scope['client'][0]}:{scope['client'][1]} {scope['method']} {scope['path']}[/#0EA5E9]",
        )
        route: str = scope["path"]
        try:
            if route.startswith("/static"):
                body = self.serve_static(route)
                if body:
                    await send_response(
                        200,
                        body[0],
                        [[b"content-type", body[1].encode() if body else b"text/html"]],
                        send,
                    )
                    emulog("success", scope, 200)
                else:
                    await self.render_not_found(send)
                    emulog("fail", scope, 404)
            elif route == "/favicon.ico":
                favicon = self.serve_static("/static/favicon.ico")
                if favicon:
                    await send_response(
                        200,
                        favicon[0],
                        [
                            [
                                b"content-type",
                                favicon[1].encode() if favicon else b"text/html",
                            ]
                        ],
                        send,
                    )
                    emulog("success", scope, 200)
                else:
                    await self.render_not_found(send)
                    emulog("fail", scope, 404)
            elif route.startswith("/scripts"):
                body = await self.serve_script(route)  # type: ignore[assignment]
                if body:
                    await send_response(
                        200, body, [[b"content-type", b"text/javascript"]], send
                    )
                    emulog("success", scope, 200)
                else:
                    await self.render_not_found(send)
                    emulog("fail", scope, 404)
            elif route.startswith("/styles"):
                body = await self.serve_styles(route)  # type: ignore[assignment]
                if body:
                    await send_response(
                        200, body, [[b"content-type", b"text/css"]], send
                    )
                    emulog("success", scope, 200)
                else:
                    await self.render_not_found(send)
                    emulog("fail", scope, 404)
            else:
                try:
                    body = return_template(self.pages[route])  # type: ignore[assignment]
                    status = 200
                    headers: list[list[str | bytes]] = [[b"content-type", b"text/html"]]
                    if self.server.get(route):
                        mod = await load_server(self.server[route])
                        if mod:
                            data = await get_load_data(mod, await receive())
                            if data and body and isinstance(body, str):
                                body = render_template(body, data.body)
                                if isinstance(body, Exception):
                                    await self.render_error(send)
                                    raise body
                                status = data.status
                                headers: list[list[str | bytes]] = data.headers  # type: ignore[no-redef]
                            else:
                                await self.render_error(send)
                            emulog("fail", scope, 500)
                        else:
                            await self.render_error(send)
                            emulog("fail", scope, 500)
                    if body:
                        await send_response(status, body, headers, send)
                        emulog("success", scope, status)
                    else:
                        await self.render_not_found(send)
                    emulog("fail", scope, 404)
                except KeyError:
                    await self.render_not_found(send)
                    emulog("fail", scope, 404)
        except Exception:
            await self.render_error(send)
            emulog("fail", scope, 500)
            console.print_exception()

    async def run(
        self,
        host: str = "localhost",
        port: int = 8000,
        dev: bool = False,
        reload_dirs: list[Path] = [],
    ) -> None:
        """
        Run the app

        Arguments
        ---------
        host: str
            The host of the app
        port: int
            The port of the app
        dev: bool
            Whether to run in development mode
        reload_dirs: list[Path]
            The directories to reload

        Returns
        -------
        None
        """
        config = uvicorn.Config(
            self,
            host=host,
            port=port,
            reload=dev,
            log_level="critical",
            reload_dirs=[path.as_posix() for path in reload_dirs],
        )
        server = uvicorn.Server(config)
        try:
            console.print(
                f"[#8B5CF6 bold]✅ Server running at http://{host}:{port}[/#8B5CF6 bold]",
                (f"[#D97706 bold]🚀 dev mode: {dev}[/#D97706 bold]\n"),
            )
            await server.serve()
        except KeyboardInterrupt:
            console.print("[#8B5CF6 bold]\n🛑 Server stopped[/#8B5CF6 bold]\n")
        except Exception as e:
            console.print(f"[#FF0000 bold]🚨 {e}[/#FF0000 bold]\n")

    def serve_static(
        self, route: str
    ) -> tuple[bytes, str] | tuple[bytes, t.Literal["text/plain"]] | None:
        """
        Serve the static files

        Arguments
        ---------
        route: str
            The route to the static file

        Returns
        -------
        tuple[bytes, str] | tuple[bytes, t.Literal["text/plain"]] | None
            The static file or None
        """
        try:
            with open(self.static[route.replace("/static", "")], "rb") as file:
                mime_type, _ = mimetypes.guess_type(
                    self.static[route.replace("/static", "")]
                )
                if mime_type:
                    return file.read(), mime_type
                return file.read(), "text/plain"
        except (FileNotFoundError, KeyError):
            return None

    async def render_not_found(self, send: Callable[..., t.Any]) -> None:
        """
        Render the 404 page

        Arguments
        ---------
        send: collections.abc.Callable[..., t.Any]
            The send function

        Returns
        -------
        None
        """
        try:
            body = return_template(self.pages["/404"])
            await send_response(
                404,
                body if body else "404 Not Found",
                [[b"content-type", b"text/html"]],
                send,
            )
        except KeyError:
            await send_response(
                404, "404 Not Found", [[b"content-type", b"text/html"]], send
            )

    async def render_error(self, send: Callable[..., t.Any]) -> None:
        """
        Render the 500 page

        Arguments
        ---------
        send: collections.abc.Callable[..., t.Any]
            The send function

        Returns
        -------
        None
        """
        try:
            body = return_template(self.pages.get("/500") or self.pages["/500"])
            await send_response(
                500,
                body if body else "500 Internal Server Error",
                [[b"content-type", b"text/html"]],
                send,
            )
        except KeyError:
            await send_response(
                500,
                "500 Internal Server Error",
                [[b"content-type", b"text/html"]],
                send,
            )

    async def serve_script(self, route: str) -> str | None:
        """
        Serve the scripts

        Arguments
        ---------
        route: str
            The route to the script

        Returns
        -------
        str | None
            The script or None
        """
        try:
            with open(self.scripts[route.replace("/scripts", "")], "r") as file:
                return file.read()
        except (FileNotFoundError, KeyError):
            return None

    async def serve_styles(self, route: str) -> str | None:
        """
        Serve the styles

        Arguments
        ---------
        route: str
            The route to the style

        Returns
        -------
        str | None
            The style or None
        """
        try:
            with open(self.styles[route.replace("/styles", "")], "r") as file:
                return file.read()
        except (FileNotFoundError, KeyError):
            return None

__call__(scope, receive, send) async

The main function of the app

Arguments

scope: dict[str, typing.Any] The scope of the request receive: collections.abc.Callable[..., t.Any] The receive function send: collections.abc.Callable[..., t.Any] The send function

Returns:

Type Description
None

Raises:

Type Description
Exception

If an error occurs

Notes

This function is called by uvicorn

Source code in emuleo/http.py
async def __call__(
    self,
    scope: dict[str, t.Any],
    receive: Callable[..., t.Any],
    send: Callable[..., t.Any],
) -> None:
    """
    The main function of the app

    Arguments
    ---------
    scope: dict[str, typing.Any]
        The scope of the request
    receive: collections.abc.Callable[..., t.Any]
        The receive function
    send: collections.abc.Callable[..., t.Any]
        The send function

    Returns
    -------
    None

    Raises
    ------
    Exception
        If an error occurs

    Notes
    -----
    This function is called by uvicorn
    """
    assert scope["type"] == "http"
    console.print(
        f"[#0EA5E9]🔗 {scope['client'][0]}:{scope['client'][1]} {scope['method']} {scope['path']}[/#0EA5E9]",
    )
    route: str = scope["path"]
    try:
        if route.startswith("/static"):
            body = self.serve_static(route)
            if body:
                await send_response(
                    200,
                    body[0],
                    [[b"content-type", body[1].encode() if body else b"text/html"]],
                    send,
                )
                emulog("success", scope, 200)
            else:
                await self.render_not_found(send)
                emulog("fail", scope, 404)
        elif route == "/favicon.ico":
            favicon = self.serve_static("/static/favicon.ico")
            if favicon:
                await send_response(
                    200,
                    favicon[0],
                    [
                        [
                            b"content-type",
                            favicon[1].encode() if favicon else b"text/html",
                        ]
                    ],
                    send,
                )
                emulog("success", scope, 200)
            else:
                await self.render_not_found(send)
                emulog("fail", scope, 404)
        elif route.startswith("/scripts"):
            body = await self.serve_script(route)  # type: ignore[assignment]
            if body:
                await send_response(
                    200, body, [[b"content-type", b"text/javascript"]], send
                )
                emulog("success", scope, 200)
            else:
                await self.render_not_found(send)
                emulog("fail", scope, 404)
        elif route.startswith("/styles"):
            body = await self.serve_styles(route)  # type: ignore[assignment]
            if body:
                await send_response(
                    200, body, [[b"content-type", b"text/css"]], send
                )
                emulog("success", scope, 200)
            else:
                await self.render_not_found(send)
                emulog("fail", scope, 404)
        else:
            try:
                body = return_template(self.pages[route])  # type: ignore[assignment]
                status = 200
                headers: list[list[str | bytes]] = [[b"content-type", b"text/html"]]
                if self.server.get(route):
                    mod = await load_server(self.server[route])
                    if mod:
                        data = await get_load_data(mod, await receive())
                        if data and body and isinstance(body, str):
                            body = render_template(body, data.body)
                            if isinstance(body, Exception):
                                await self.render_error(send)
                                raise body
                            status = data.status
                            headers: list[list[str | bytes]] = data.headers  # type: ignore[no-redef]
                        else:
                            await self.render_error(send)
                        emulog("fail", scope, 500)
                    else:
                        await self.render_error(send)
                        emulog("fail", scope, 500)
                if body:
                    await send_response(status, body, headers, send)
                    emulog("success", scope, status)
                else:
                    await self.render_not_found(send)
                emulog("fail", scope, 404)
            except KeyError:
                await self.render_not_found(send)
                emulog("fail", scope, 404)
    except Exception:
        await self.render_error(send)
        emulog("fail", scope, 500)
        console.print_exception()

render_error(send) async

Render the 500 page

Arguments

send: collections.abc.Callable[..., t.Any] The send function

Returns:

Type Description
None
Source code in emuleo/http.py
async def render_error(self, send: Callable[..., t.Any]) -> None:
    """
    Render the 500 page

    Arguments
    ---------
    send: collections.abc.Callable[..., t.Any]
        The send function

    Returns
    -------
    None
    """
    try:
        body = return_template(self.pages.get("/500") or self.pages["/500"])
        await send_response(
            500,
            body if body else "500 Internal Server Error",
            [[b"content-type", b"text/html"]],
            send,
        )
    except KeyError:
        await send_response(
            500,
            "500 Internal Server Error",
            [[b"content-type", b"text/html"]],
            send,
        )

render_not_found(send) async

Render the 404 page

Arguments

send: collections.abc.Callable[..., t.Any] The send function

Returns:

Type Description
None
Source code in emuleo/http.py
async def render_not_found(self, send: Callable[..., t.Any]) -> None:
    """
    Render the 404 page

    Arguments
    ---------
    send: collections.abc.Callable[..., t.Any]
        The send function

    Returns
    -------
    None
    """
    try:
        body = return_template(self.pages["/404"])
        await send_response(
            404,
            body if body else "404 Not Found",
            [[b"content-type", b"text/html"]],
            send,
        )
    except KeyError:
        await send_response(
            404, "404 Not Found", [[b"content-type", b"text/html"]], send
        )

run(host='localhost', port=8000, dev=False, reload_dirs=[]) async

Run the app

Arguments

host: str The host of the app port: int The port of the app dev: bool Whether to run in development mode reload_dirs: list[Path] The directories to reload

Returns:

Type Description
None
Source code in emuleo/http.py
async def run(
    self,
    host: str = "localhost",
    port: int = 8000,
    dev: bool = False,
    reload_dirs: list[Path] = [],
) -> None:
    """
    Run the app

    Arguments
    ---------
    host: str
        The host of the app
    port: int
        The port of the app
    dev: bool
        Whether to run in development mode
    reload_dirs: list[Path]
        The directories to reload

    Returns
    -------
    None
    """
    config = uvicorn.Config(
        self,
        host=host,
        port=port,
        reload=dev,
        log_level="critical",
        reload_dirs=[path.as_posix() for path in reload_dirs],
    )
    server = uvicorn.Server(config)
    try:
        console.print(
            f"[#8B5CF6 bold]✅ Server running at http://{host}:{port}[/#8B5CF6 bold]",
            (f"[#D97706 bold]🚀 dev mode: {dev}[/#D97706 bold]\n"),
        )
        await server.serve()
    except KeyboardInterrupt:
        console.print("[#8B5CF6 bold]\n🛑 Server stopped[/#8B5CF6 bold]\n")
    except Exception as e:
        console.print(f"[#FF0000 bold]🚨 {e}[/#FF0000 bold]\n")

serve_script(route) async

Serve the scripts

Arguments

route: str The route to the script

Returns:

Type Description
str | None

The script or None

Source code in emuleo/http.py
async def serve_script(self, route: str) -> str | None:
    """
    Serve the scripts

    Arguments
    ---------
    route: str
        The route to the script

    Returns
    -------
    str | None
        The script or None
    """
    try:
        with open(self.scripts[route.replace("/scripts", "")], "r") as file:
            return file.read()
    except (FileNotFoundError, KeyError):
        return None

serve_static(route)

Serve the static files

Arguments

route: str The route to the static file

Returns:

Type Description
tuple[bytes, str] | tuple[bytes, Literal['text/plain']] | None

The static file or None

Source code in emuleo/http.py
def serve_static(
    self, route: str
) -> tuple[bytes, str] | tuple[bytes, t.Literal["text/plain"]] | None:
    """
    Serve the static files

    Arguments
    ---------
    route: str
        The route to the static file

    Returns
    -------
    tuple[bytes, str] | tuple[bytes, t.Literal["text/plain"]] | None
        The static file or None
    """
    try:
        with open(self.static[route.replace("/static", "")], "rb") as file:
            mime_type, _ = mimetypes.guess_type(
                self.static[route.replace("/static", "")]
            )
            if mime_type:
                return file.read(), mime_type
            return file.read(), "text/plain"
    except (FileNotFoundError, KeyError):
        return None

serve_styles(route) async

Serve the styles

Arguments

route: str The route to the style

Returns:

Type Description
str | None

The style or None

Source code in emuleo/http.py
async def serve_styles(self, route: str) -> str | None:
    """
    Serve the styles

    Arguments
    ---------
    route: str
        The route to the style

    Returns
    -------
    str | None
        The style or None
    """
    try:
        with open(self.styles[route.replace("/styles", "")], "r") as file:
            return file.read()
    except (FileNotFoundError, KeyError):
        return None