Move slowAPI to submodule

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "libs"]
+	path = libs
+	url = http://chai.bokhary.ir/0880/SlowAPI.git

app.py

@@ -8,7 +8,7 @@ from itertools import product
 from pathlib import Path
 from typing import Any
 
-from slow import JSONAPI, App, HTTPResponse, JSONResponse, Request, render
+from libs.slow import JSONAPI, App, HTTPResponse, JSONResponse, Request, render
 
 
 class Coord:

slow/__init__.py

@@ -1,11 +0,0 @@
-from .slow import JSONAPI, App, Headers, HTTPResponse, JSONResponse, Request, render
-
-__all__ = [
-    "JSONAPI",
-    "App",
-    "HTTPResponse",
-    "JSONResponse",
-    "render",
-    "Request",
-    "Headers",
-]

slow/slow.py

@@ -1,260 +0,0 @@
-import asyncio
-import http.client
-import json
-import re
-import urllib.parse
-from pathlib import Path
-from typing import Any, Awaitable, Callable, Optional
-
-PR = re.compile(r"\<([a-zA-Z_][a-zA-Z0-9_]*)\>")
-
-
-class Headers:
-    def __init__(self):
-        self._d: dict[str, str] = {}
-
-    def get(self, key: str, default: Optional[Any] = None) -> str | Any:
-        return self._d.get(key.lower(), default)
-
-    def set(self, key: str, value: str) -> None:
-        self._d[key.lower()] = value
-
-    def __str__(self):
-        return str(self._d)
-
-
-class Request:
-    def __init__(self, method: str, path: str, headers: Headers, body: bytes):
-        self.method = method
-        self.path = path
-        self.headers = headers
-        self.body = body
-
-    def __str__(self):
-        return str(
-            {
-                "method": self.method,
-                "path": self.path,
-                "headers": self.headers,
-                "body": self.body,
-            }
-        )
-
-
-async def _default_404_route(request: Request):
-    return "HTTP/1.1 404 Not Found\r\nContent-Type: text/html\r\n\r\n404 Not Found contact admin".encode(
-        encoding="utf-8"
-    )
-
-
-async def _default_405_route(request: Request):
-    return "HTTP/1.1 405 Method Not Allowed\r\nContent-Type: text/html\r\n\r\n405 Method Not Allowed".encode(
-        encoding="utf-8"
-    )
-
-
-class App:
-    def __init__(self):
-        self.routes: dict[
-            re.Pattern[str], dict[str, Callable[[Request, ...], Awaitable[bytes]]]
-        ] = {}
-        self.error_routes: dict[int, Callable[[Request, ...], Awaitable[bytes]]] = {
-            404: _default_404_route,
-            405: _default_405_route,
-        }
-
-    def _pattern_to_regex(self, temp) -> re.Pattern[str]:
-        re_temp = temp
-        iter = PR.finditer(temp)
-        for m in iter:
-            name = m[1]
-            re_temp = re.sub(
-                m[0], r"(?P<" + name + r">[a-zA-Z0-9\-._~/:%&=]+)", re_temp
-            )
-        return re.compile(re_temp)
-
-    def _serve(
-        self, path: str, method: str, func: Callable[[Request, ...], Awaitable[bytes]]
-    ):
-        if method not in ["GET", "POST", "PUT", "DELETE"]:
-            raise RuntimeError(f'Invalid method "{method}".')
-        pat = self._pattern_to_regex(path)
-        if pat not in self.routes:
-            self.routes[pat] = {}
-        if method in self.routes[pat]:
-            raise RuntimeWarning(f'Path "{path}" already exists.')
-        self.routes[pat][method] = func
-
-    def GET(self, path: str):
-        """Decorator to register a GET HTTP route."""
-
-        def decorator(func: Callable[[Request, ...], Awaitable[bytes]]):
-            self._serve(path, "GET", func)
-            return func
-
-        return decorator
-
-    def POST(self, path: str):
-        """Decorator to register a POST HTTP route."""
-
-        def decorator(func: Callable[[Request, ...], Awaitable[bytes]]):
-            self._serve(path, "POST", func)
-            return func
-
-        return decorator
-
-    def PUT(self, path: str):
-        """Decorator to register a PUT HTTP route."""
-
-        def decorator(func: Callable[[Request, ...], Awaitable[bytes]]):
-            self._serve(path, "PUT", func)
-            return func
-
-    def DELETE(self, path: str):
-        """Decorator to register a DELETE HTTP route."""
-
-        def decorator(func: Callable[[Request, ...], Awaitable[bytes]]):
-            self._serve(path, "DELETE", func)
-            return func
-
-        return decorator
-
-    def OPTIONS(self, path: str):
-        """Decorator to register a OPTIONS HTTP route."""
-
-        def decorator(func: Callable[[Request, ...], Awaitable[bytes]]):
-            self._serve(path, "OPTIONS", func)
-            return func
-
-        return decorator
-
-    def error(self, code):
-        """Decorator to register an error route."""
-
-        def decorator(func):
-            self.error_routes[code] = func
-            return func
-
-        return decorator
-
-    def resolve(self, path, method) -> tuple[Callable[..., Awaitable[bytes]], dict]:
-        for pattern, route in self.routes.items():
-            if m := pattern.fullmatch(path):
-                if method not in route:
-                    return self.error_routes[405], {}
-                return route[method], {
-                    k: urllib.parse.unquote(v) for k, v in m.groupdict().items()
-                }
-        return self.error_routes[404], {}
-
-    async def handle_client(
-        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
-    ):
-        """Handle an incoming connection (HTTP or WebSocket)."""
-        try:
-            # Read the initial HTTP request line
-            request_line = await reader.readline()
-            if not request_line:
-                return
-
-            # Parse request line
-            parts = request_line.decode(encoding="utf-8").strip().split()
-            if len(parts) < 3:
-                return
-
-            method, path, protocol = parts[0], parts[1], parts[2]
-
-            assert protocol == "HTTP/1.1"
-
-            headers: Headers = Headers()
-            while True:
-                line = await reader.readline()
-                if line == b"\r\n" or line == b"\n" or not line:  # End of headers
-                    break
-                line = line.decode("utf-8").strip()
-                if ":" in line:
-                    key, value = line.split(":", 1)
-                    headers.set(key.strip(), value.strip())
-
-            content_length = int(headers.get("Content-Length", 0))
-            body = await reader.read(content_length) if content_length else b""
-
-            route, kwargs = self.resolve(path, method)
-
-            response = await route(
-                request=Request(
-                    method=method,
-                    path=path,
-                    headers=headers,
-                    body=body,
-                ),
-                **kwargs,
-            )
-            writer.write(response)
-
-            await writer.drain()
-        except Exception as e:
-            print(f"Internal Server Error: {e}")
-        finally:
-            writer.close()
-            await writer.wait_closed()
-
-    async def run(self, host="127.0.0.1", port=8000):
-        """Start the async server."""
-
-        server = await asyncio.start_server(self.handle_client, host, port)
-
-        print(f"Serving on http://{host}:{port}")
-        async with server:
-            await server.serve_forever()
-
-
-AccessControlAllowOrigin = "*"
-
-
-def HTTPResponse(
-    content: str, status=200, content_type="text/plain; charset=utf-8", headers=[]
-) -> bytes:
-    global AccessControlAllowOrigin
-    head: str = f"HTTP/1.1 {status} {http.client.responses.get(status, 'Unkown Status Code')}\r\nContent-Type: {content_type}\r\n"
-    head += f"Access-Control-Allow-Origin: {AccessControlAllowOrigin}\r\n"  # CORS
-    head += "Access-Control-Allow-Methods: GET, POST, PUT, DELETE, OPTIONS\r\n"  # CORS
-    head += "\r\n".join(headers) + ("\r\n" if len(headers) > 0 else "")
-    head += "\r\n"
-    return (head + content).encode(encoding="utf-8")
-
-
-_value_pattern = re.compile(r"\{\{\s*([a-zA-Z_][a-zA-Z_0-9]*)\s*\}\}")
-
-
-def render(file: str | Path, variables: dict[str, Any] = {}) -> bytes:
-    if isinstance(file, str):
-        file = Path(file)
-    content: str = file.read_text(encoding="utf-8")
-    for m in _value_pattern.findall(content):
-        if m in variables:
-            content = re.sub(r"\{\{\s*" + m + r"\s*\}\}", variables[m], content)
-    return HTTPResponse(content, content_type="text/html; charset=utf-8")
-
-
-def JSONResponse(d: dict, status=200) -> bytes:
-    return HTTPResponse(
-        json.dumps(d), status=status, content_type="text/json; charset=utf-8"
-    )
-
-
-def JSONAPI(func):
-    async def wrapper(*args, **kwargs):
-        result = await func(*args, **kwargs)
-        if not isinstance(result, dict):
-            if (
-                isinstance(result, tuple)
-                and len(result) == 2
-                and isinstance(result[1], dict)
-                and isinstance(result[0], int)
-            ):
-                return JSONResponse(result[1], result[0])
-            raise RuntimeError("Return value of JSONAPI route is not a dictionary")
-        return JSONResponse(result)
-
-    return wrapper