MostMinimalWebFramework.py

import json
import re
import traceback
from dataclasses import dataclass, field
from socket import AF_INET, SHUT_WR, SO_REUSEADDR, SOCK_STREAM, SOL_SOCKET, socket
from typing import Any, Callable, Dict, List, Tuple
from urllib.parse import parse_qs, urlparse


@dataclass
class Request:
    method: str
    headers: Dict[str, str]
    path: str
    query_params: List[Dict[str, List[str]]] = field(default_factory=list)
    body: Any = None


@dataclass
class Response:
    body: Any
    status_code: int = 200
    content_type: str = "text/html"


class JSONResponse:
    def __new__(cls, *args, **kwargs):
        return Response(content_type="application/json", *args, **kwargs)


class ApiException(Response, BaseException):
    pass


class MostMinimalWebFramework:
    route_table: List[Tuple[re.Pattern, Callable]] = []

    def route(self, path: str) -> Callable:
        def decorator(func: Callable):
            def __inner():
                return func()

            self.route_table.append((re.compile(path + "$"), func))
            return __inner

        return decorator

    def get_route_function(self, searched_path: str) -> Callable:
        return next(r for r in self.route_table if r[0].match(searched_path))[1]

    def request_parser(self, request_str: str) -> Request:
        request_lines = request_str.split("\r\n")
        method, url, _ = request_lines[0].split(" ")  # first line has method and url

        headers = {}
        for i, line in enumerate(request_lines[1:], 1):

            if line == "":  # under empty line, whole data is body
                try:
                    body = json.loads("".join(request_lines[i + 1 :]))
                except json.JSONDecodeError:
                    body = "".join(request_lines[i + 1 :])
                break

            j = line.find(":")  # left part of : will key, right part will be value
            headers[line[:j].upper()] = line[j + 2 :]

        url = urlparse(url)
        return Request(method, headers, url.path, parse_qs(url.query), body)

    def build_response(self, r: Response) -> str:
        body = r.body if isinstance(r.body, str) else json.dumps(r.body)
        return (
            f"HTTP/1.1 {r.status_code}\r\nContent-Type: {r.content_type}; charset=utf-8"
            f"\r\nContent-Length: {len(body)}\r\nConnection: close\r\n\r\n{body}"
        )

    def run(self, address: str, port: int):
        serversocket = socket(AF_INET, SOCK_STREAM)
        serversocket.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
        try:
            serversocket.bind((address, port))
            serversocket.listen(5)
            while True:
                clientsocket, _ = serversocket.accept()
                request = clientsocket.recv(4096).decode()
                try:
                    parsed_req = self.request_parser(request)
                    response = self.get_route_function(parsed_req.path)(parsed_req)
                except ApiException as e:
                    response = e
                except Exception:
                    print(traceback.format_exc())
                    response = Response({"msg": "500 - server error"}, 500)
                print(response.status_code, parsed_req.method, parsed_req.path)
                clientsocket.sendall(self.build_response(response).encode())
                clientsocket.shutdown(SHUT_WR)
        finally:
            serversocket.close()


if __name__ == "__main__":

    app = MostMinimalWebFramework()

    @app.route("/")
    def f(request):
        return Response("Hello World")

    @app.route("/json-response/")
    def json_response(request):
        return JSONResponse({"msg": "Hello World"})

    @app.route("/method-handling/")
    def method_handling(request):
        if request.method == "GET":
            return Response("Your method is GET")

        elif request.method == "POST":
            return Response("Your method is POST")

    @app.route("/status-code/")
    def status_code(request):
        return Response("Different status code", status_code=202)

    @app.route("/raise-exception/")
    def exception_raising(request):
        raise ApiException({"msg": "custom_exception"}, status_code=400)

    @app.route("/body-handle/")
    def body_handling(request):
        try:
            name = request.body["name"]
        except (KeyError, TypeError):
            raise ApiException({"msg": "name field required"}, status_code=400)

        return JSONResponse({"request__name": name})

    @app.route("/query-param-handling/")
    def query_param_handling(request):
        try:
            q_parameter = request.query_params["q"][0]
        except (KeyError, TypeError):
            raise ApiException({"msg": "q query paramter required"}, status_code=400)

        return JSONResponse({"your_q_parameter": q_parameter})

    @app.route("/header-handling/")
    def header_handling(request):
        try:
            token = request.headers["X-TOKEN"]
        except (KeyError, TypeError):
            raise ApiException({"msg": "Un authorized"}, status_code=403)

        return Response(f"your token {token}")

    @app.route("/user/[^/]*/posts")
    def varialbe_path(request):
        user_id = request.path[len("/user/") : -len("/posts")]
        return Response(f"posts for {user_id}", status_code=201)

    @app.route("/.*")
    def func_404(request):
        return Response("404", status_code=404)

    app.run("0.0.0.0", 8080)