BriskAPI: A High-Performance Lightweight Python API Framework with Minimal Dependencies
import re
from pprint import pp
from urllib.parse import parse_qs

from briskapi.request import Request
from briskapi.response import JSONResponse
from briskapi.connection import WebSocketConnection
from briskapi.status import (
    HTTP_200_OK,
    HTTP_400_BAD_REQUEST,
    HTTP_404_NOT_FOUND,
    HTTP_500_INTERNAL_SERVER_ERROR,
    HTTP_422_UNPROCESSABLE_ENTITY,
)

from jsonschema import Draft7Validator


class BriskAPI:
    def __init__(self: "BriskAPI", debug: bool = True):
        self.debug = debug
        self._http_routes = []
        self._ws_routes = []

    # Exposed decorators
    def route(self, path, methods=["GET"], request_schema=None, response_schema=None):
        compiled_request_schema = None
        compiled_response_schema = None

        if path.endswith("/"):
            path = path[:-1]

        if request_schema:
            compiled_request_schema = Draft7Validator(request_schema)

        if response_schema:
            compiled_response_schema = Draft7Validator(response_schema)

        def decorator(handler):
            self._add_http_route(
                path,
                methods,
                handler,
                compiled_request_schema,
                compiled_response_schema,
            )
            return handler

        return decorator

    def ws(self, path):
        if path.endswith("/"):
            path = path[:-1]

        def decorator(handler):
            self._add_ws_route(
                path,
                handler,
            )
            return handler

        return decorator

    def _add_http_route(
        self, path, methods, handler, compiled_request_schema, compiled_response_schema
    ):
        param_re = r"{([a-zA-Z_][a-zA-Z0-9_]*)}"
        path_re = r"^" + re.sub(param_re, r"(?P<\1>\\w+)", path) + r"$"
        self._http_routes.append(
            (
                re.compile(path_re),
                methods,
                handler,
                compiled_request_schema,
                compiled_response_schema,
            )
        )
        return handler

    def _add_ws_route(
        self, path, handler
    ):
        param_re = r"{([a-zA-Z_][a-zA-Z0-9_]*)}"
        path_re = r"^" + re.sub(param_re, r"(?P<\1>\\w+)", path) + r"$"
        self._ws_routes.append(
            (
                re.compile(path_re),
                handler,
            )
        )
        return handler

    # Match path to handler + exctract slugs
    def _match_http(self, scope):
        path_info = scope["path"]
        # Remove trailing slash
        if path_info.endswith("/"):
            path_info = path_info[:-1]

        request_method = scope["method"]
        for (
            path,
            methods,
            handler,
            compiled_request_schema,
            compiled_response_schema,
        ) in self._http_routes:
            # Skip if invalid method
            if not request_method in methods:
                continue
            m = path.match(path_info)
            if m is not None:
                # Extract and return parameter values
                path_params = m.groupdict()
                return {
                    "path_params": path_params,
                    "handler": handler,
                    "compiled_request_schema": compiled_request_schema,
                    "compiled_response_schema": compiled_response_schema,
                }
            
    def _match_ws(self, scope):
        path_info = scope["path"]
        # Remove trailing slash
        if path_info.endswith("/"):
            path_info = path_info[:-1]

        for (
            path,
            handler,
        ) in self._ws_routes:
            m = path.match(path_info)
            if m is not None:
                # Extract and return parameter values
                path_params = m.groupdict()
                return {
                    "path_params": path_params,
                    "handler": handler,
                }

    # Parsers
    def _parse_qs(self, scope):
        query_string = scope["query_string"]
        if not query_string:
            return {}
        parsed = parse_qs(query_string)
        return {
            key: value[0] if len(value) == 1 else value for key, value in parsed.items()
        }

    def _parse_headers(self, scope):
        headers = dict()
        for header in scope["headers"]:
            headers[header[0].decode("utf-8")] = header[1].decode("utf-8")
        return headers

    # Body readers
    async def _read_http_body(self, receive):
        body = bytearray()
        while True:
            msg = await receive()
            body += msg["body"]
            if not msg.get("more_body"):
                break
        return bytes(body)

    def _validate_http_request(self, request: Request, compiled_request_schema):
        request_validation_errors = []
        # Validate request is a json
        try:
            request.json
        except ValueError:
            return ["Invalid request body"]
        # Validate request json
        if compiled_request_schema:
            for error in compiled_request_schema.iter_errors(request.json):
                request_validation_errors.append(error.message)
        return request_validation_errors

    def _validate_http_response(self, response: JSONResponse, compiled_response_schema):
        response_validation_errors = []
        # Validate request json
        if compiled_response_schema:
            for error in compiled_response_schema.iter_errors(response.data):
                response_validation_errors.append(error.message)
        return response_validation_errors

    async def _http_response(self, response, send):
        await send(
            {
                "type": "http.response.start",
                "status": response.status,
                "headers": [(k.encode(), v.encode()) for k, v in response.headers],
            }
        )
        await send(
            {
                "type": "http.response.body",
                "body": response.body,
            }
        )

    async def _http_handler(self, scope, receive, send):
        # Match to http router and extract slugs
        match = self._match_http(scope)
        if match is None:
            response = JSONResponse(
                status=HTTP_404_NOT_FOUND, data={"message": "Resource not found"}
            )
            await self._http_response(response, send)
            return

        path_params = match["path_params"]
        handler = match["handler"]

        # Exctract schemas
        compiled_request_schema = match["compiled_request_schema"]
        compiled_response_schema = match["compiled_response_schema"]

        # Parse request dependencies
        headers = self._parse_headers(scope)
        query_params = self._parse_qs(scope)
        body = await self._read_http_body(receive)

        # Prepare Request object
        path_info = scope["path"]
        request_method = scope["method"]

        request = Request()
        request.headers = headers
        request.method = request_method
        request.path = path_info
        request.slugs = path_params
        request.query = query_params
        request.body = body

        # Validate request
        request_validation_errors = self._validate_http_request(
            request, compiled_request_schema
        )

        if request_validation_errors:
            response = JSONResponse(
                status=HTTP_422_UNPROCESSABLE_ENTITY,
                data={
                    "message": "validation error",
                    "details": request_validation_errors,
                },
            )
            await self._http_response(response, send)
            return

        # Prepare Response
        handler_response = handler(request)

        # Check JSONResponse is returned
        if not isinstance(handler_response, JSONResponse):
            raise ValueError("Invalid response object")

        # Validate response
        response_validation_errors = self._validate_http_response(
            handler_response, compiled_response_schema
        )
        if response_validation_errors:
            raise ValueError("Response validation failed {response_validation_errors}")

        # Flow success
        await self._http_response(handler_response, send)

    async def _ws_handler(self, scope, receive, send):
        # Match to ws router and extract slugs
        match = self._match_ws(scope)
        if match is None:
            await send({
                "type": "websocket.close",
                "code": 1000
            })
        
        path_params = match["path_params"]
        handler = match["handler"]

        # Parse ws connection dependencies
        query_params = self._parse_qs(scope)

        # prepare websocket connection object
        connection = WebSocketConnection()
        connection.receive = receive
        connection.send = send
        connection.path = scope["path"]
        connection.query = query_params
        connection.slugs = path_params
        await handler(connection)


    async def __call__(self, scope, receive, send):

        if scope["type"] == "http":
            await self._http_handler(scope, receive, send)
        elif scope["type"] == "lifespan":
            await self.lifespan_handler(scope, receive, send)
        elif scope["type"] == "websocket":
            await self._ws_handler(scope, receive, send)

briskapi/applications.py
achaayb/BriskAPI
from typing import Any
class WebSocketConnection:
    
    _receive:Any
    _send: Any

    path: str
    query: dict[str, str]
    slugs: dict[str, str]

    async def accept(self):
        await self.send({"type": "websocket.accept"})
    
    async def receive_message(self):
        message = await self.receive()
        if message["type"] == "websocket.receive":
            return message.get("text", "")
        return None

    async def send_message(self, message):
        await self.send({
            "type": "websocket.send",
            "text": message,
        })

    async def close(self, code=1000):
        await self.send({
            "type": "websocket.close",
            "code": code,
        })
briskapi/connection.py
achaayb/BriskAPI
import json
from typing import Any, List, Tuple, Union

from briskapi.status import (
    HTTP_200_OK,
    HTTP_400_BAD_REQUEST,
    HTTP_404_NOT_FOUND,
    HTTP_500_INTERNAL_SERVER_ERROR,
)


class Request:
    path: str
    method: str
    headers: dict[str, str]
    query: dict[str, str]
    slugs: dict[str, str]
    body: bytes

    @property
    def text(self) -> str:
        return self.body.decode()

    @property
    def json(self) -> object:
        if not self.body:
            return {}
        return json.loads(self.body)

    @property
    def dict(self) -> object:
        return {
            "path": self.path,
            "method": self.method,
            "headers": self.headers,
            "query": self.query,
            "slugs": self.slugs,
            "body": self.text,
        }

briskapi/request.py
achaayb/BriskAPI
from briskapi.status import (
    HTTP_200_OK,
    HTTP_400_BAD_REQUEST,
    HTTP_404_NOT_FOUND,
    HTTP_500_INTERNAL_SERVER_ERROR,
)
import json
from typing import Any, List, Tuple, Union, Dict
from pprint import pp


class JSONResponse:
    def __init__(
        self,
        data: Union[Dict[str, Any], list[Dict[str, Any]]] = None,
        status: str = HTTP_200_OK,
        headers: dict[str, str] = [("Content-Type", "application/json")],
    ):
        self.data = data
        self.status = status
        self.headers = headers

    def set_header(self, key: str, value: str) -> None:
        # Check if the header already exists
        for i, (existing_key, _) in enumerate(self.headers):
            if existing_key == key:
                # Update the existing header
                self.headers[i] = (key, value)
                break
        else:
            # Add a new header if it doesn't exist
            self.headers.append((key, value))

    @property
    def body(self) -> bytes:
        if self.data is None:
            return b"{}"
        elif isinstance(self.data, (dict, list)):
            return json.dumps(self.data).encode()
        else:
            raise TypeError("Invalide response data type")

    @property
    def status_to_str(self) -> str:
        return str(self.status)

briskapi/response.py
achaayb/BriskAPI
HTTP_100_CONTINUE = 100
HTTP_101_SWITCHING_PROTOCOLS = 101
HTTP_102_PROCESSING = 102
HTTP_103_EARLY_HINTS = 103
HTTP_200_OK = 200
HTTP_201_CREATED = 201
HTTP_202_ACCEPTED = 202
HTTP_203_NON_AUTHORITATIVE_INFORMATION = 203
HTTP_204_NO_CONTENT = 204
HTTP_205_RESET_CONTENT = 205
HTTP_206_PARTIAL_CONTENT = 206
HTTP_207_MULTI_STATUS = 207
HTTP_208_ALREADY_REPORTED = 208
HTTP_226_IM_USED = 226
HTTP_300_MULTIPLE_CHOICES = 300
HTTP_301_MOVED_PERMANENTLY = 301
HTTP_302_FOUND = 302
HTTP_303_SEE_OTHER = 303
HTTP_304_NOT_MODIFIED = 304
HTTP_305_USE_PROXY = 305
HTTP_306_RESERVED = 306
HTTP_307_TEMPORARY_REDIRECT = 307
HTTP_308_PERMANENT_REDIRECT = 308
HTTP_400_BAD_REQUEST = 400
HTTP_401_UNAUTHORIZED = 401
HTTP_402_PAYMENT_REQUIRED = 402
HTTP_403_FORBIDDEN = 403
HTTP_404_NOT_FOUND = 404
HTTP_405_METHOD_NOT_ALLOWED = 405
HTTP_406_NOT_ACCEPTABLE = 406
HTTP_407_PROXY_AUTHENTICATION_REQUIRED = 407
HTTP_408_REQUEST_TIMEOUT = 408
HTTP_409_CONFLICT = 409
HTTP_410_GONE = 410
HTTP_411_LENGTH_REQUIRED = 411
HTTP_412_PRECONDITION_FAILED = 412
HTTP_413_REQUEST_ENTITY_TOO_LARGE = 413
HTTP_414_REQUEST_URI_TOO_LONG = 414
HTTP_415_UNSUPPORTED_MEDIA_TYPE = 415
HTTP_416_REQUESTED_RANGE_NOT_SATISFIABLE = 416
HTTP_417_EXPECTATION_FAILED = 417
HTTP_418_IM_A_TEAPOT = 418
HTTP_421_MISDIRECTED_REQUEST = 421
HTTP_422_UNPROCESSABLE_ENTITY = 422
HTTP_423_LOCKED = 423
HTTP_424_FAILED_DEPENDENCY = 424
HTTP_425_TOO_EARLY = 425
HTTP_426_UPGRADE_REQUIRED = 426
HTTP_428_PRECONDITION_REQUIRED = 428
HTTP_429_TOO_MANY_REQUESTS = 429
HTTP_431_REQUEST_HEADER_FIELDS_TOO_LARGE = 431
HTTP_451_UNAVAILABLE_FOR_LEGAL_REASONS = 451
HTTP_500_INTERNAL_SERVER_ERROR = 500
HTTP_501_NOT_IMPLEMENTED = 501
HTTP_502_BAD_GATEWAY = 502
HTTP_503_SERVICE_UNAVAILABLE = 503
HTTP_504_GATEWAY_TIMEOUT = 504
HTTP_505_HTTP_VERSION_NOT_SUPPORTED = 505
HTTP_506_VARIANT_ALSO_NEGOTIATES = 506
HTTP_507_INSUFFICIENT_STORAGE = 507
HTTP_508_LOOP_DETECTED = 508
HTTP_510_NOT_EXTENDED = 510
HTTP_511_NETWORK_AUTHENTICATION_REQUIRED = 511

briskapi/status.py
achaayb/BriskAPI
from briskapi.applications import BriskAPI
from briskapi.request import Request
from briskapi.response import JSONResponse
from briskapi.connection import WebSocketConnection
from briskapi.status import HTTP_202_ACCEPTED
app = BriskAPI()


@app.route(
    "/{hello}",
    methods=["POST"],
    request_schema={
        "type": "object",
        "properties": {"name": {"type": "string"}, "age": {"type": "integer"}},
        "required": ["name", "age"],
    }
)
def hello_world(request: Request):
    hello = request.slugs["hello"]
    response = JSONResponse(data={"name": request.headers["foo"]}, status=HTTP_202_ACCEPTED)
    response.set_header("Hello", "World")
    return response

@app.ws("/")
async def foo(connection: WebSocketConnection):
    await connection.accept()
    while True:
        message = await connection.receive_message()
        if message == "close":
            await connection.close()
            return
        await connection.send_message(str(message))
test_app.py
achaayb/BriskAPI