/MostMinimalWebFramework

Most Minimal Python Web Framework

Primary LanguagePythonGNU General Public License v3.0GPL-3.0

What?

Most Minimal Python Web Framework (in terms of number of line).

Key Importance Points Of Implementation

importancenote
under 100 lines of code⚫ (MUST)formatted with black
no thirth party⚫ (MUST)only built-in python modules
readablity🟢 (IMPORTANT)anyone should easily change this code with her needs
easy to use🟢 (IMPORTANT)should feels like flask
for personal use🟡 (Meh)if it is not essential service for you, personal use is okay
performance🟡 (Meh)not given importance but probably good because of simple implementation
standards🟡 (Meh)not always follows, for instance not use `CONTENT-TYPE` header
featureful⭕ (NO)only the basics
production🔴 (NEVER)

Why?

  1. Why not?
  2. For learning.
  3. I use this framework for my custom RSS feed generators. I don’t want to build docker images, pypi download workflows, etc. I have a small server with a small disk. So dependency-free framework is good. No need to venv, no need docker.

    Just run python MostMinimalWebFramework.py without worrying about dependencies, virtual envs, or docker image building process.

  4. Also if you like docker, you can use the official python docker image without building a new image, without pushing it somewhere or without waiting for build processes etc.
    docker run -it --rm \
       -v "$PWD":/app \
       -w /app \
       -p 8080:8080 \
       python:3-alpine \
       python MostMinimalWebFramework.py
        

Examples

Hello world example:

app = MostMinimalWebFramework()

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

app.run("0.0.0.0", 8080)

Json Response:

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

Method handling:

@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")

Different status code:

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

Exception raising:

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

Request body handling:

@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})

Query param handling:

@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})

Header handling:

@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(token)

Variable path

@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)

Framework FULL Code:

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()