MostMinimalWebFramework

• What?

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

• Key Importance Points Of Implementation

| | importance | note | |-------------------------+----------------+-------------------------------------------------------------------------| | 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.

   #+begin_src bash docker run -it --rm
   -v "$PWD":/app
   -w /app
   -p 8080:8080
   python:3-alpine
   python MostMinimalWebFramework.py #+end_src

• Examples

Hello world example:

#+begin_src python app = MostMinimalWebFramework()

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

app.run("0.0.0.0", 8080) #+end_src

Json Response:

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

Method handling:

#+begin_src python @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")

#+end_src

Different status code:

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

Exception raising:

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

Request body handling:

#+begin_src python @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})

#+end_src

Query param handling:

#+begin_src python @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})

#+end_src

Header handling:

#+begin_src python @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)

#+end_src

Variable path

#+begin_src python @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) #+end_src

• Framework FULL Code:

#+begin_src python 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()

#+end_src