asgikit

Asgikit - ASGI Toolkit

Asgikit is a toolkit for building asgi applications and frameworks.

It is intended to be a minimal library and provide the building blocks for other libraries.

The examples directory contain usage examples of several use cases

Features:

• Request
  • Headers
  • Cookies
  • Body (bytes, str, json, form, stream)
  • Form
• Response
  • Plain text
  • Json
  • Streaming
  • File
• Websockets

Requests and Responses

Asgikit Request, like other libraries, have methods to read items from the incoming request. However, unlike other libraries, there is no response object. Instead, you use the methods in Request to respond to the request like respond_json and respond_stream. There is a response property in the Request object where you can set response attributes like status, headers and cookies.

The main methods to interact with the Request are the following:

class Request:
    # Read the request body as a byte stream
    async def read_stream(self) -> AsyncIterable[bytes]: ...
    # Read the request body as bytes
    async def read_bytes(self) -> bytes: ...
    # Read the request body as str
    async def read_text(self, encoding: str = None) -> str: ...
    # Read the request body and parse it as json
    async def read_json(self) -> Any: ...
    # Read the request body and parse it as form
    async def read_form(self) -> dict[str, str | list[str]]: ...

    # Respond with bytes
    async def respond_bytes(self, content: bytes): ...
    # Respond with str
    async def respond_text(self, content: str): ...
    # Respond with the given content encoded as json
    async def respond_json(self, content: Any): ...
    # Respond with empty response and given status
    async def respond_status(self, status: HTTPStatus): ...
    # Respond with redirect
    async def respond_redirect(self, location: str, permanent: bool = False): ...
    # Respond with a post/redirect/get
    # https://en.wikipedia.org/wiki/Post/Redirect/Get
    async def respond_redirect_post_get(self, location: str): ...
    # Context manager that provides a function to write to the response
    async def response_writer(self): ...
    # Respond with file
    async def respond_file(self, path: str | os.PathLike): ...

Example request and response

from asgikit.requests import Request


async def main(scope, receive, send):
    assert scope["type"] == "http"

    request = Request(scope, receive, send)

    # request method
    method = request.method

    # request path
    path = request.path

    # request headers
    headers = request.headers

    # read body as json
    body = await request.read_json()

    data = {
        "lang": "Python",
        "async": True,
        "platform": "asgi",
        "method": method,
        "path": path,
        "headers": dict(headers.items()),
        "body": body,
    }

    # send json response
    await request.respond_json(data)

Example websocket

from asgikit.requests import Request
from asgikit.websockets import WebSocketDisconnect


async def app(scope, receive, send):
    assert scope["type"] == "websocket"

    request = Request(scope, receive, send)
    ws = await request.websocket_accept()

    while True:
        try:
            message = await ws.read()
            await ws.write(message)
        except WebSocketDisconnect:
            print("Client disconnect")
            break