example2.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright 2024 Josua Krause
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""Another example quick server."""
from time import monotonic, sleep
from typing import Any, TypedDict

from quick_server import (
    create_server,
    msg,
    PreventDefaultResponse,
    QuickServerRequestHandler,
    ReqArgs,
    ReqNext,
    Response,
    setup_restart,
    WorkerArgs,
)


ResUptime = TypedDict('ResUptime', {
    "uptime": Any,
})


def run() -> None:
    """Runs the example quick server."""
    addr = ""
    port = 8000
    server = create_server((addr, port))
    server.bind_path("/", "..")
    server.add_default_white_list()
    server.favicon_fallback = "../src/quick_server/favicon.ico"
    server.suppress_noise = True
    server.report_slow_requests = True

    server.link_worker_js("/js/worker.js")
    server.max_file_size = 78

    mcs = server.max_chunk_size
    start = monotonic()
    count_uptime = 0

    @server.json_get("/api/uptime/", 1)
    def _uptime(
            req: QuickServerRequestHandler, args: ReqArgs) -> dict[str, Any]:
        nonlocal count_uptime

        # request has one mandatory additional path segment
        sleep(int(args["paths"][0]))
        count_uptime += 1
        res: dict[str, Any] = {
            "uptime": req.log_elapsed_time_string(
                (monotonic() - start) * 1000.0).strip(),
        }

        def convert(value: Any) -> Any:
            try:
                return float(value)
            except ValueError:
                return value

        for (key, value) in args["query"].items():
            res[key] = [
                convert(v) for v in f"{value}".split(",")
            ] if "," in f"{value}" else convert(value)
        return res

    @server.json_post("/api/upload")
    def _upload_file(
            _req: QuickServerRequestHandler, args: ReqArgs) -> dict[int, str]:
        ix = 0
        res = {}
        for (k, v) in sorted(args["post"].items(), key=lambda e: e[0]):
            res[ix] = f"{k} is {v}"
            ix += 1
        for (name, f) in sorted(args["files"].items(), key=lambda e: e[0]):
            bfcontent = f.read()
            size = len(bfcontent)
            try:
                fcontent = bfcontent.decode("utf-8")
            except UnicodeDecodeError:
                fcontent = repr(bfcontent)
            res[ix] = f"{name} is {size} bytes"
            ix += 1
            for line in fcontent.split("\n"):
                res[ix] = line
                ix += 1
        return res

    @server.json_worker("/api/uptime_worker")
    def uptime_worker(args: WorkerArgs) -> ResUptime:
        nonlocal count_uptime

        msg(f"sleep {int(args['time'])}")
        sleep(int(args["time"]))
        count_uptime += 1
        return {
            "uptime": (monotonic() - start) * 1000.0,
        }

    @server.json_worker("/api/message")
    def _message(args: WorkerArgs) -> str:
        if args["split"]:
            server.max_chunk_size = 10
        else:
            server.max_chunk_size = mcs
        sleep(2)
        return "1234567890 the quick brown fox jumps over the lazy dog"

    def check_login(
            _req: QuickServerRequestHandler,
            args: ReqArgs,
            okay: ReqNext) -> ReqNext | Response | dict[str, str]:
        token = args["query"].get("token")
        if token == "secret":
            args["meta"]["username"] = "user"
            return okay
        if token == "default":
            return {
                "name": "other",
            }
        if token == "except":
            raise PreventDefaultResponse(403, "Forbidden")
        return Response("Authentication Required", 401)

    @server.json_get("/api/user_details")
    @server.middleware(check_login)
    def _user_details(
            _req: QuickServerRequestHandler, args: ReqArgs) -> dict[str, str]:
        return {
            "name": args["meta"]["username"],
        }

    def complete_requests(_args: list[str], text: str) -> list[str]:
        return ["uptime"] if "uptime".startswith(text) else []

    @server.cmd(1, complete_requests)
    def requests(args: list[str]) -> None:
        if args[0] != 'uptime':
            msg(f"unknown request: {args[0]}")
        else:
            msg(f"requests made to {args[0]}: {count_uptime}")

    msg(f"starting server at {addr if addr else 'localhost'}:{port}")
    server.serve_forever()
    msg("shutting down..")
    server.server_close()


if __name__ == "__main__":
    setup_restart()
    run()