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 auth: Any

Get the auth attribute from the asgi scope

For compatibility with starlette middleware

property body: Body

Request body

Returns:

The request Body object, or None if request is websocket

property client: tuple[str, int] | None

Client information

property cookies: MultiDict[str]

Request cookies

property headers: Headers

Request headers

property http_version: str

HTTP version

property method: HTTPMethod | None

HTTP method of the request

property path: str

Request path

property path_params: dict[str, Any]

Get the path_params attribute from the asgi scope

For compatibility with starlette middleware

property query: MultiDict[str]

Parsed query string

property raw_path: str | None

Raw request path

property raw_query: str

Raw query string

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

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_empty(status: HTTPStatus = HTTPStatus.NO_CONTENT, *, headers: dict[str, str] = None, cookies: Cookies = None)

Send an empty response with the given status

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

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_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_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

property response: Response

The underlying response object

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

property root_path: str

Root path

property scheme: str

URL scheme

property server: tuple[str, int | None]

Server information

property session: Any

Get the session attribute from the asgi scope

For compatibility with starlette middleware

property state: dict | None

State managed by the ASGI server

property user: Any

Get the user attribute from the asgi scope

For compatibility with starlette middleware

class asgikit.Body(scope, receive)

Request body

Provides methods to read the request body

property charset: str | None

Charset of the request

property content_length: int | None

Content length of the request body

property content_type: str | None

Content type of the request body

property is_consumed: bool

Verifies whether the request body is consumed or not

async read_bytes() → bytes

Read the full request body

async read_form() → MultiDict[str | UploadedFile]

Read the full request body and parse it as form encoded

async read_json() → Any

Read the full request body and parse it as json

async read_text(encoding: str = None) → str

Read the full request body as str

class asgikit.Response(scope, receive, send)

Response object used to interact with the client

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

property is_finished: bool

Tells whether the response is started or not

property is_started: 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:
async write(body: bytes, *, more_body=False)

Write data to the response

Raises:

ResponseNotStartedError – If the response is not started

class asgikit.WebSocket(scope, receive, send)

Represents a WebSocket connection

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

Accepts the WebSocket connection

Raises:

WebSocketStateError – If the WebSocket has already been accepted

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

Close the WebSocket connection

Does nothing if the websocket is already closed

async iter() → AsyncIterator[str | bytes]

Iterate over data from the WebSocket connection

Data can be either str or bytes

async iter_json() → AsyncIterator[Any]

Iterate over data as json from the WebSocket connection

async read() → str | bytes

Read data from the WebSocket connection

Data can be either str or bytes

async read_json() → Any

Read data as json from the WebSocket connection

property state: WebSocketState

State of the websocket connection

property subprotocols: list[str]

Return a list of subprotocols of this WebSocket connection

async write(data: str | bytes)

Send data to the WebSocket connection

Raises:
async write_json(data: Any)

Send data as json to the WebSocket connection

Raises:

WebSocketClosedError – If the WebSocket is 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(key: str, default: list[str] = None) → list[str] | None

Get all items in the given key

get_first(key: str, default: str = None) → str | None

Get the first item in the given key

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

dict that can hold multiple values in a single key

get(key: str, default: list[T] = None) → list[T] | None

Get all items in the given key

get_first(key: str, default: T = None) → T | None

Get the first item in the given key

items() → Iterable[tuple[str, T]]

Iterate over all items

keys() → Iterable[str]

Iterate over all keys

values() → Iterable[list[T]]

Iterate over all values

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

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

Remove a cookie

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

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.MultipartNotEnabledError

Dependency supporting multipart is not installed

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.WebSocketClosedError

Tried to interact with a closed websocket

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

Websocket is in the wrong state for the interaction