asgikit¶

Toolkit for creating ASGI application and frameworks

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

Bases: object

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 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)¶

Bases: object

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)¶

Bases: object

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)¶

Bases: object

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)¶

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
WebSocketClosedError – If the WebSocket is closed

async write_json(data: Any)¶

Send data as json to the WebSocket connection

Raises:: WebSocketClosedError – 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)¶

Bases: Enum

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)¶

Bases: MultiDict[str]

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

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.MultiDict(data: Iterable[tuple[str, T]] = None)¶

Bases: Generic[T]

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)¶

Bases: MultiDict[T]

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

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.Cookies¶

Bases: object

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)¶

Bases: object

File uploaded in a multipart form

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

exception asgikit.AsgiException¶

Bases: Exception

Generic ASGI exception

add_note()¶: Exception.add_note(note) – add a note to the exception

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception asgikit.ClientDisconnectError¶

Bases: HttpException

Client disconnected

add_note()¶: Exception.add_note(note) – add a note to the exception

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception asgikit.HttpException¶

Bases: AsgiException

Generic HTTP exception

add_note()¶: Exception.add_note(note) – add a note to the exception

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception asgikit.MultipartBoundaryError¶

Bases: Exception

Failed to find the multipart boundary

add_note()¶: Exception.add_note(note) – add a note to the exception

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception asgikit.RequestAlreadyConsumedError¶

Bases: HttpException

Tried to consume a request body that is already consumed

add_note()¶: Exception.add_note(note) – add a note to the exception

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception asgikit.ResponseAlreadyEndedError¶

Bases: HttpException

Interacted with a response that has already ended

add_note()¶: Exception.add_note(note) – add a note to the exception

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception asgikit.ResponseAlreadyStartedError¶

Bases: HttpException

Tried to start a response that has already started

add_note()¶: Exception.add_note(note) – add a note to the exception

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception asgikit.ResponseNotStartedError¶

Bases: HttpException

Interacted with a response that has not yet started

add_note()¶: Exception.add_note(note) – add a note to the exception

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception asgikit.WebSocketClosedError¶

Bases: WebSocketException

Tried to interact with a closed websocket

add_note()¶: Exception.add_note(note) – add a note to the exception

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

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

Bases: WebSocketException

Websocket client disconnected

add_note()¶: Exception.add_note(note) – add a note to the exception

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception asgikit.WebSocketException¶

Bases: AsgiException

Generic websocket exception

add_note()¶: Exception.add_note(note) – add a note to the exception

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception asgikit.WebSocketResponseNotSupportedError¶

Bases: WebSocketException

Asgi server does not support websocket denial response

add_note()¶: Exception.add_note(note) – add a note to the exception

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception asgikit.WebSocketStateError¶

Bases: WebSocketException

Websocket is in the wrong state for the interaction

add_note()¶: Exception.add_note(note) – add a note to the exception

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

asgikit¶

Zayt

Navigation

Related Topics