Api¶

Ferramenta para criação de aplicações e frameworks ASGI

class asgikit.Cookies¶

Cookies que serão enviados na resposta

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

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

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

Arquivo enviado em um formulário multipart

async copy_to(dst: str | PathLike)¶: Copia o arquivo para o caminho fornecido

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

Cabeçalhos baseados em MultiDict

get(key: str, default: str | None = None) → str | None¶: Recupera o primeiro item na chave fornecida

getall(key: str, default: list[str] | None = None) → list[str] | None¶: Recupera todos os itens na chave fornecida

class asgikit.MultiDict(data: Iterable[tuple[T_KEY, T_VAL]])¶

dict que pode armazenar múltiplos valores em uma mesma chave

dict()¶: Retorna uma cópia do dict interno

get(key: T_KEY, default: T_VAL = None) → T_VAL | None¶: Recupera o primeiro item na chave fornecida

getall(key: T_KEY, default: list[T_VAL] | None = None) → list[T_VAL] | None¶: Recupera todos os itens na chave fornecida

keys() → a set-like object providing a view on D's keys¶

values() → an object providing a view on D's values¶

items() → a set-like object providing a view on D's items¶

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

Representa uma requisição

Encapsula um objeto Response que é utilizado para responder à requisição.

property body: Body¶

Corpo da requisição

Retorna:: O objeto da requisição Body, ou None sea requisição for websocket

property response: Response¶: O objeto da resposta

property websocket: WebSocket¶: O objeto do websocket, ou None se a requisição for http

property state: dict | None¶: Estado gerenciado pelo servidor ASGI

property http_version: str¶: Versão do HTTP

property server: tuple[str, int | None]¶: Dados do servidor

property client: tuple[str, int] | None¶: Dados do cliente

property scheme: str¶: Esquema da URL

property method: HTTPMethod | None¶: Método HTTP da requisição

property root_path: str¶: Caminho raiz

property path: str¶: Caminho da requisição

property raw_path: str | None¶: Caminho bruto da requisição

property headers: Headers¶: Cabeçalhos da requisição

property raw_query: str¶: Quer string bruto

property query: MultiDict[str, str]¶: Query string processado

property cookies: MultiDict[str, str]¶: Cookies da requisição

property session: Any¶

Recupera o atributo session do escopo asgi

Para compatibilidade com middleware starlette

property auth: Any¶

Recupera o atributo auto do escopo asgi

Para compatibilidade com middleware starlette

property user: Any¶

Recupera o atributo user do escopo asgi

Para compatibilidade com middleware starlette

property path_params: dict[str, Any]¶

Recupera o atributo path_params do escopo asgi

Para compatibilidade com middleware starlette

async respond_bytes(content: bytes, *, status=HTTPStatus.OK, media_type: str | None = None, headers: dict[str, str] | None = None, cookies: Cookies | None = None)¶: Responde com o conteúdo fornecido e finaliza a resposta

async respond_text(content: str, *, status=HTTPStatus.OK, media_type: str = 'text/plain', headers: dict[str, str] | None = None, cookies: Cookies | None = None, encoding: str = 'utf-8')¶: Responde com o conteúdo fornecido e finaliza a resposta

async respond_json(content: Any, *, status=HTTPStatus.OK, media_type: str = 'application/json', headers: dict[str, str] | None = None, cookies: Cookies | None = None, encoding: str = 'utf-8')¶: Responde com o conteúdo fornecido serializado como JSON

async respond_empty(status: HTTPStatus = HTTPStatus.NO_CONTENT, *, headers: dict[str, str] | None = None, cookies: Cookies | None = None)¶: Envia uma resposta vazia com o status fornecido

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

Responde com um redirecionamento

Parâmetros:

location – Localização para onde redirecionar
permanent – Se verdadeiro, envia um redirecionamento permanente (HTTP 308), caso contrário, envia um redirecionamento temporário (HTTP 307)

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

Responde com HTTP 303

Utilizado para enviar um redirecionamento para um endpoint GET após uma requisição POST, conhecido como post/redirect/get https://en.wikipedia.org/wiki/Post/Redirect/Get

Parâmetros:: location – Localização para onde redirecionar

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

Context manager para streaming para a resposta

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

Levanta:: ClientDisconnectError – Se o cliente desconectar enquanto envia dados

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

Responde com o stream de dados fornecido

Levanta:: ClientDisconnectError – Se o cliente desconectar enquanto envia dados

async respond_file(path: str | PurePath, *, status=HTTPStatus.OK, media_type: str | None = None, content_length: int | None = None, headers: dict[str, str] | None = None, cookies: Cookies | None = None, stat_result: stat_result | None = None)¶: Envia o arquivo fornecido para a resposta

class asgikit.Body(scope, receive)¶

Corpo da requisição

Provê métodos para ler o corpo da requisição

property content_type: str | None¶: Content type do corpo da requisição

property content_length: int | None¶: Content length do corpo da requisição

property charset: str | None¶: Charset da requisição

property is_consumed: bool¶: Verifica se o corpo da requisição foi consumido ou não

async read_bytes() → bytes¶: Lê o corpo inteiro da requisição

async read_text(encoding: str | None = None) → str¶: Lê o corpo inteiro da requisição como str

async read_json() → Any¶: Lê o corpo inteiro da requisição e processa como json

async read_form() → MultiDict[str | UploadedFile]¶: Lê o corpo inteiro da requisição e processa como form

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

Objeto da requisição utilizado para interagir com o cliente

property is_started: bool¶: Verifica se a resposta foi iniciada

property is_finished: bool¶: Verifica se a resposta foi iniciada

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

Inicia a resposta

Deve ser chamado antes de chamar write() ou end()

Levanta:

ResponseAlreadyStartedError – Se a resposta já estiver iniciada
ResponseAlreadyEndedError – Se a resposta estiver finilizada
WebSocketResponseNotSupportedError – Quando tenta enviar uma resposta websocket denial em um servidor que não suporta

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

Escreve dados para a resposta

Levanta:: ResponseNotStartedError – Se a resposta não estiver iniciada

async end()¶

Finaliza a resposta

Deve ser chamado quando não houver mais dados para serem escritos na resposta

Não faz nada se a resposta já estiver terminada

Levanta:: ResponseNotStartedError – Se a resposta não estiver iniciada

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

Representa uma conexão websocket

property state: WebSocketState¶: Estado da conexão websocket

property subprotocols: list[str]¶: Retorna uma lista de subprotocolos da conexão websocket

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

Aceita a conexão websocket

Levanta:: WebSocketStateError – Se o websocket já estiver aceito

async read() → str | bytes¶

Lê dados da conexão websocket

Dados podem ser str ou bytes

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

Itera sobre os dados da conexão websocket

Dados podem ser str ou bytes

async read_json() → Any¶: Lê dados da conexão websocket como JSON

async iter_json() → AsyncIterator[Any]¶: Itera sobre os dados da conexão websocket como JSON

async write(data: str | bytes)¶

Envia dados para a conexão websocket

Levanta:: WebSocketStateError – Se o websocket não estiver aceito

async write_json(data: Any)¶

Envia dados para a conexão websocket como JSON

Levanta:: WebSocketStateError – Se o websocket estiver fechado

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

Fecha a conexão websocket

Não faz nada se o websocket já estiver fechado

class asgikit.WebSocketState(value)¶

Estado da conexão websocket

NEW = 1¶: Criado, ainda não aceito

CONNECTED = 2¶: Evento connect recebido

ACCEPTED = 3¶: Websocket aceito

CLOSED = 4¶: Websocket fechado

exception asgikit.AsgiException¶: Exceção genérica ASGI

exception asgikit.ClientDisconnectError¶: Cliente desconectado

exception asgikit.HttpException¶: Exceção HTTP genérica

exception asgikit.MultipartBoundaryError¶: Falha ao procurar o boundary multipart

exception asgikit.RequestAlreadyConsumedError¶: Tentou consumir o corpo da requisição que já foi consumido

exception asgikit.ResponseAlreadyEndedError¶: Interagiu com uma resposta que já foi terminada

exception asgikit.ResponseAlreadyStartedError¶: Tentou iniciar uma resposta que já foi iniciada

exception asgikit.ResponseNotStartedError¶: Interafiu com uma resposta que ainda não foi iniciada

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

exception asgikit.WebSocketException¶: Exceção websocket genérica

exception asgikit.WebSocketResponseNotSupportedError¶: Servidor asgi não suport resposta websocket denail

exception asgikit.WebSocketStateError(actual, expected)¶: Websocket está no estado incorreto para a interação