Understanding and Creating your Own ASGI Web Framework [Part-1]

> O.S interface
---> Signal handling - Respond to OS signals (SIGTERM, SIGHUP, etc.) for control operations
---> File descriptor management - Efficiently handle large numbers of open connections
> Network interface
---> Socket binding - Bind to specific IP addresses and ports
---> TCP/UDP support - Handle different transport protocols
> Process Management
---> Workers lifecycle, Spawn process and/or threads to run the web application

> When a request arrives:
---> it must make sure we will send it to the correct handler
---> it must make sure we only receive methods that we have mapped [GET, POST, etc...]
---> it must allow you have access to headers
---> it must allow you have access to query string
---> it must allow you have access to body data
> When a request is completed
---> Make sure we are sending the response to the client with the data we need
---> Repassing status code, headers, payloads and etc

class App:
    
    # Every time a request comes to the server, the server will call the __call__ method of the class
    # The __call__ method is the entry point of the application
    async def __call__(self, scope, receive, send) -> None:
        # scope - as described in the ASGI spec it holds the request information as path, query strings, headers etc... in this case a dict var
        # receive - its an async generator that yields the request body
        # send - its an async function that sends the response back to the client
        pass

app = App()
# install uvicorn `pip install uvicorn`

/ (root)
- > orders
  - > user_ID
    - > pending 

from dataclasses import dataclass
from typing import Any, List, Union, Dict

# this will represent our endpoint/route
@dataclass
class _Route:
    handler: Any
    path: str

# this will make the magic of association between path similarities
class _NodeRoute:
    """
    Visual example
    routes
     -> / -> home handler
     -> /users -> users handler
     -> /files/documents -> documents handler

    root [segment = /, children = Users, Files, handler = home/index handler]

        -> users [segment = users, children = ..., handler =  users handler]

        -> files [segment = files, children = documents, handler =  None]
            -> documents [segment = documents, children = ..., handler =  documents handler]
    ...
    and that continue as required, allowing to add more items and nesting as required
    """
    __slots__ = ["segment", "children", "handler"]

    def __init__(self, segment: str = ""):
        self.segment: str = segment  # the part of the route we will have
        self.children: Dict[str, _NodeRoute] = {}  # dict segment and its node route
        self.handler: Any = None  # the handler if exists

    def __repr__(self) -> str:
        return f"NodeRoute({self.segment}) children = {len(self.children)}"


# this will be the interface with our users
class Router:
    __slots__ = ["root"]

    def __init__(self):
        self.root = _NodeRoute()

    @staticmethod
    def get_segments(path: str) -> List[str]:
        return path.split("/")

    def add_route(self, path: str, handler: Any) -> None:
        current = self.root
        segments = self.get_segments(path)
        # insertion DFS like
        for segment in segments:
            if segment not in current.children:
                current.children[segment] = _NodeRoute(segment)

            # swp to the next nest level
            current = current.children[segment]

        current.handler = handler

    def get_route(self, path: str) -> Union[_NodeRoute, None]:
        current = self.root
        segments = self.get_segments(path)
        for segment in segments:
            if segment not in current.children:
                # print("not found")
                return None

            current = current.children[segment]

        return current

from router import Router


async def home():
    print("home triggered")

async def users():
    print("users triggered")

class App:

    def __init__(self):
        self.router = Router()
        self.router.add_route("/", home)
        self.router.add_route("/users", users)

    async def __call__(self, scope, receive, send):
        target = self.router.get_route(scope['path'])

        # following the specs, in order to provide a response to our caller/client
        # we must follow the two steps you see below
        # !keep a closer eye to the type of each send call
        # with http.response.start we can provide status code, headers [considering we are dealing with http requests]
        # with the http.response.body we can provide a flag finalizing if we are planning to send more data [streaming like] and the payload/chunk
        # this is enough for now, we will get back to it after to improve and add more details and fix a few things

        if target is None:
            print("handler not found, ignoring")
            await send({"type": "http.response.start", "status": 404})
            await send({"type": "http.response.body", "body": b"not found"})
            return

        print("handler found")
        await target.handler()
        await send({"type": "http.response.start", "status": 200})
        await send({"type": "http.response.body", "body": b"ok"})


app = App()