Api¶

Toolkit for creating ASGI application and frameworks

class asgikit.Request(scope, receive, send)¶

Represents an incoming request

It encapsulates a Response object that is used to respond to the request.

property body: Body¶

Request body

Returns:: The request Body object, or None if request is websocket

property response: Response¶: The underlying response object

property websocket: WebSocket¶: The underlying websocket object, or None if request is http

property state: dict | None¶: State managed by the ASGI server

property http_version: str¶: HTTP version

property server: tuple[str, int | None]¶: Server information

property client: tuple[str, int] | None¶: Client information

property scheme: str¶: URL scheme

property method: HTTPMethod | None¶: HTTP method of the request

property root_path: str¶: Root path

property path: str¶: Request path

property raw_path: str | None¶: Raw request path

property headers: Headers¶: Request headers

property raw_query: str¶: Raw query string

property query: MultiDict[str]¶: Parsed query string

property cookies: MultiDict[str]¶: Request cookies

property session: Any¶

Get the session attribute from the asgi scope

For compatibility with starlette middleware

property auth: Any¶

Get the auth attribute from the asgi scope

For compatibility with starlette middleware

property user: Any¶

Get the user attribute from the asgi scope

For compatibility with starlette middleware

property path_params: dict[str, Any]¶

Get the path_params attribute from the asgi scope

For compatibility with starlette middleware

async respond_bytes(content: bytes, *, status=HTTPStatus.OK, media_type: str = None, headers: dict[str, str] = None, cookies: Cookies = None)¶: Respond with the given content and finish the response

async respond_text(content: str, *, status=HTTPStatus.OK, media_type: str = 'text/plain', headers: dict[str, str] = None, cookies: Cookies = None, encoding: str = 'utf-8')¶: Respond with the given content and finish the response

async respond_json(content: Any, *, status=HTTPStatus.OK, media_type: str = 'application/json', headers: dict[str, str] = None, cookies: Cookies = None, encoding: str = 'utf-8')¶: Respond with the given content serialized as JSON

async respond_empty(status: HTTPStatus = HTTPStatus.NO_CONTENT, *, headers: dict[str, str] = None, cookies: Cookies = None)¶: Send an empty response with the given status

async redirect(location: str, *, permanent: bool = False, headers: dict[str, str] = None, cookies: Cookies = None)¶

Respond with a redirect

Parameters:

location – Location to redirect to
permanent – If true, send permanent redirect (HTTP 308), otherwise send a temporary redirect (HTTP 307).

async redirect_post_get(location: str, *, headers: dict[str, str] = None, cookies: Cookies = None)¶

Response with HTTP status 303

Used to send a redirect to a GET endpoint after a POST request, known as post/redirect/get https://en.wikipedia.org/wiki/Post/Redirect/Get

Parameters:: location – Location to redirect to

response_writer(status=HTTPStatus.OK, *, media_type: str = None, content_length: int = None, headers: dict[str, str] = None, cookies: Cookies = None, encoding='utf-8') → AsyncGenerator[Callable[[bytes], Awaitable], None]¶

Context manager for streaming data to the response

import json
response = Response(scope, receive, send)
async with response.response_writer(
    media_type="application/x-ndjson"
) as write:
    for i in range(10):
        await write(json.dump({"number": i}))

Raises:: ClientDisconnectError – If the client disconnects while sending data

async respond_stream(stream: AsyncIterable[bytes | str], *, status=HTTPStatus.OK, media_type: str = None, content_length: int = None, headers: dict[str, str] = None, cookies: Cookies = None, encoding: str = 'utf-8')¶

Respond with the given stream of data

Raises:: ClientDisconnectError – If the client disconnects while sending data

async respond_file(path: str | PathLike, *, status=HTTPStatus.OK, media_type: str = None, content_length: int = None, headers: dict[str, str] = None, cookies: Cookies = None, stat_result: stat_result = None)¶: Send the given file to the response

class asgikit.Body(scope, receive)¶

Request body

Provides methods to read the request body

property content_type: str | None¶: Content type of the request body

property content_length: int | None¶: Content length of the request body

property charset: str | None¶: Charset of the request

property is_consumed: bool¶: Verifies whether the request body is consumed or not

async read_bytes() → bytes¶: Read the full request body

async read_text(encoding: str = None) → str¶: Read the full request body as str

async read_json() → Any¶: Read the full request body and parse it as json

async read_form() → MultiDict[str | UploadedFile]¶: Read the full request body and parse it as form encoded

class asgikit.Response(scope, receive, send)¶

Response object used to interact with the client

property is_started: bool¶: Tells whether the response is started or not

property is_finished: bool¶: Tells whether the response is started or not

async start(status: HTTPStatus, *, media_type: str = None, content_length: int = None, headers: dict[str, str] = None, cookies: Cookies = None, encoding: str = 'utf-8')¶

Start the response

Must be called before calling write() or end()

Raises:

ResponseAlreadyStartedError – If the response is already started
ResponseAlreadyEndedError – If the response is finished
WebSocketResponseNotSupportedError – When sending websocket denial response on server that does not support it

async write(body: bytes, *, more_body=False)¶

Write data to the response

Raises:: ResponseNotStartedError – If the response is not started

async end()¶

Finish the response

Should be called when no more data will be written to the response

Does nothing if the response already ended

Raises:: ResponseNotStartedError – If the response is not started

class asgikit.WebSocket(scope, receive, send)¶

Represents a WebSocket connection

property state: WebSocketState¶: State of the websocket connection

property subprotocols: list[str]¶: Return a list of subprotocols of this WebSocket connection

async accept(subprotocol: str = None, headers: dict[str, str] = None) → Self¶

Accepts the WebSocket connection

Raises:: WebSocketStateError – If the WebSocket has already been accepted

async read() → str | bytes¶

Read data from the WebSocket connection

Data can be either str or bytes

async iter() → AsyncIterator[str | bytes]¶

Iterate over data from the WebSocket connection

Data can be either str or bytes

async read_json() → Any¶: Read data as JSON from the WebSocket connection

async iter_json() → AsyncIterator[Any]¶: Iterate over data as JSON from the WebSocket connection

async write(data: str | bytes)¶

Send data to the WebSocket connection

Raises:: WebSocketStateError – If the WebSocket is not yet accepted

async write_json(data: Any)¶

Send data as JSON to the WebSocket connection

Raises:: WebSocketStateError – If the WebSocket is closed

async close(code: int = 1000, reason: str = None)¶

Close the WebSocket connection

Does nothing if the websocket is already closed

class asgikit.WebSocketState(value)¶

State of the WebSocket connection

NEW = 1¶: Created, not yet accepted

CONNECTED = 2¶: Received the connect event

ACCEPTED = 3¶: Websocket accepted

CLOSED = 4¶: Websocket closed

class asgikit.Headers(data: Iterable[tuple[bytes, bytes]] = None)¶

Immutable headers backed by multivalue dict

get_first(key: str, default: str = None) → str | None¶: Get the first item in the given key

get(key: str, default: list[str] = None) → list[str] | None¶: Get all items in the given key

class asgikit.MultiDict(data: Iterable[tuple[str, T]] = None)¶

dict that can hold multiple values in a single key

get_first(key: str, default: T = None) → T | None¶: Get the first item in the given key

get(key: str, default: list[T] = None) → list[T] | None¶: Get all items in the given key

keys() → Iterable[str]¶: Iterate over all keys

values() → Iterable[list[T]]¶: Iterate over all values

items() → Iterable[tuple[str, T]]¶: Iterate over all items

class asgikit.MutableMultiDict(data: Iterable[tuple[str, T]] = None)¶

Mutable version of MultiValueDict

add(key: str, *args: T)¶: Add items in the given key. If the key already exists, the value is appended

set(key: str, *args: T)¶: Set items in the given key. Overwrite the existing value

class asgikit.Cookies¶

Cookies to be sent in the response

set(name: str, value: str, *, expires: int = None, domain: str = None, path: str = None, max_age: int = None, secure: bool = False, httponly: bool = True, samesite: Literal['strict', 'lax', 'none'] = 'lax', partitioned: bool = False)¶: Add a cookie

delete(name: str, *, domain: str = None, path: str = None, secure: bool = False, httponly: bool = True, samesite: Literal['strict', 'lax', 'none'] = 'lax')¶: Remove a cookie

class asgikit.UploadedFile(file: SpooledTemporaryFile, filename: str, media_type: str, size: int)¶

File uploaded in a multipart form

async copy_to(dst: str | PathLike)¶: Copy file into given path

exception asgikit.AsgiException¶: Generic ASGI exception

exception asgikit.ClientDisconnectError¶: Client disconnected

exception asgikit.HttpException¶: Generic HTTP exception

exception asgikit.MultipartBoundaryError¶: Failed to find the multipart boundary

exception asgikit.RequestAlreadyConsumedError¶: Tried to consume a request body that is already consumed

exception asgikit.ResponseAlreadyEndedError¶: Interacted with a response that has already ended

exception asgikit.ResponseAlreadyStartedError¶: Tried to start a response that has already started

exception asgikit.ResponseNotStartedError¶: Interacted with a response that has not yet started

exception asgikit.WebSocketDisconnect(code: int, reason: str | None)¶: Websocket client disconnected

exception asgikit.WebSocketException¶: Generic websocket exception

exception asgikit.WebSocketResponseNotSupportedError¶: Asgi server does not support websocket denial response

exception asgikit.WebSocketStateError(actual, expected)¶: Websocket is in the wrong state for the interaction