bareasgi

class Application(CoreApplication) ¶

Summary¶

Construct the application

Description¶

from bareasgi import (
    Application,
    Scope,
    HttpRequest,
    HttpResponse,
    text_reader,
    text_writer
)

async def http_request_callback(request: HttpRequest) -> HttpResponse:
    text = await text_reader(request.body)
    return HttpResponse(
        200,
        [(b'content-type', b'text/plain')],
        text_writer('This is not a test')
    )

import uvicorn

app = Application()
app.http_router.add({'GET', 'POST', 'PUT', 'DELETE'}, '/{path}', http_request_callback)

uvicorn.run(app, port=9009)

bareasgi.Application(

*,

middlewares: Optional[List[HttpMiddlewareCallback]],

http_router: Optional[HttpRouter],

web_socket_router: Optional[WebSocketRouter],

startup_handlers: Optional[List[LifespanHandler]],

shutdown_handlers: Optional[List[LifespanHandler]],

not_found_response: Optional[HttpResponse],

info: Optional[Dict[str, Any]]

) -> None

Parameters¶

middlewares: Optional[List[HttpMiddlewareCallback]] (optional)

Optional middleware callbacks. Defaults to None.

http_router: Optional[HttpRouter] (optional)

Optional router to for http routes. Defaults to None.

web_socket_router: Optional[WebSocketRouter] (optional)

Optional router for web routes. Defaults to None.

startup_handlers: Optional[List[LifespanHandler]] (optional)

Optional handlers to run at startup. Defaults to None.

shutdown_handlers: Optional[List[LifespanHandler]] (optional)

Optional handlers to run at shutdown. Defaults to None.

not_found_response: Optional[HttpResponse] (optional)

Optional not found (404) response. Defaults to DEFAULT_NOT_FOUND_RESPONSE.

info: Optional[Dict[str, Any]] (optional)

Optional dictionary for user data. Defaults to None.

method Application.on_http_request ¶

Summary¶

A decorator to add an http route handler to the application

Application.on_http_request(

methods: AbstractSet[str],

path: str

) -> Callable[[HttpRequestCallback], HttpRequestCallback]

Parameters¶

methods: AbstractSet[str]

The http methods, e.g. {{'POST', 'PUT'}

path: str

The path

Returns¶

Callable[[HttpRequestCallback], HttpRequestCallback]: The decorated request.

method Application.on_shutdown ¶

Summary¶

A decorator to add a startup handler to the application

Application.on_shutdown(

callback: LifespanRequestHandler

) -> LifespanRequestHandler

Parameters¶

callback: LifespanRequestHandler

The shutdown handler.

Returns¶

LifespanRequestHandler: The decorated handler.

method Application.on_startup ¶

Summary¶

A decorator to add a startup handler to the application

Application.on_startup(

callback: LifespanRequestHandler

) -> LifespanRequestHandler

Parameters¶

callback: LifespanRequestHandler

The startup handler.

Returns¶

LifespanRequestHandler: The decorated handler.

method Application.on_ws_request ¶

Summary¶

A decorator to add a websocket route handler to the application

Application.on_ws_request(

path: str

) -> Callable[[WebSocketRequestCallback], WebSocketRequestCallback]

Parameters¶

path: str

The path

Returns¶

Callable[[WebSocketRequestCallback], WebSocketRequestCallback]: The decorated handler

class HttpRequest ¶

Summary¶

An HTTP request.

bareasgi.http.HttpRequest(

scope: HTTPScope,

info: Dict[str, Any],

context: Dict[str, Any],

matches: Mapping[str, Any],

body: AsyncIterable[bytes]

) -> None

Parameters¶

scope: HTTPScope

The ASGI http scope.

info: Dict[str, Any]

Shared data from the application.

context: Dict[str, Any]

Private data for this request or chain of requests.

matches: Mapping[str, Any]

Matches made by the router.

body: AsyncIterable[bytes]

The body.

property HttpRequest.url ¶

Summary¶

Make the url from the scope.

url -> str

async method HttpRequest.content ¶

Summary¶

Return the contents of the request body as bytes.

Description¶

This function consumes the body. Calling it a second time will generate an error.

async HttpRequest.content() -> bytes

Returns¶

bytes: The body as bytes.

async method HttpRequest.json ¶

Summary¶

Return the contents of the request body as JSON.

Description¶

This function consumes the body. Calling it a second time will generate an error.

async HttpRequest.json(

decode: Callable[[Union[str, bytes]], Any]

) -> Any

Parameters¶

decode: Callable[[Union[str, bytes]], Any] (optional)

A function to decode the body to json. Defaults to json.loads.

Returns¶

Any: The body as JSON.

async method HttpRequest.text ¶

Summary¶

Return the request body as text.

Description¶

This function consumes the body. Calling it a second time will generate an error.

async HttpRequest.text(

encoding: str

) -> str

Parameters¶

encoding: str (optional)

The encoding. Defaults to 'utf-8'.

Returns¶

str: The body as text.

class HttpResponse ¶

Summary¶

The HTTP response.

bareasgi.http.HttpResponse(

status: int,

headers: Optional[List[Tuple[bytes, bytes]]],

body: Optional[AsyncIterable[bytes]],

pushes: Optional[Iterable[PushResponse]]

) -> None

Parameters¶

status: int

The status code.

headers: Optional[List[Tuple[bytes, bytes]]] (optional)

The headers if any. Mandatory headers will be added if missing. Defaults to None.

body: Optional[AsyncIterable[bytes]] (optional)

The body, if any. Defaults to None.

pushes: Optional[Iterable[PushResponse]] (optional)

Server pushes, if any. Defaults to None.

class method HttpResponse.from_bytes ¶

Summary¶

Create an HTTP response from bytes content.

HttpResponse.from_bytes(

*,

status: int,

content_type: bytes,

headers: Optional[List[Tuple[bytes, bytes]]],

chunk_size: int

) -> HttpResponse

Parameters¶

status: int (optional)

An optional status. Defaults to 200.

content_type: bytes (optional)

An optional content type. Defaults to b'text/plain'.

headers: Optional[List[Tuple[bytes, bytes]]] (optional)

Optional headers. Defaults to None.

chunk_size: int (optional)

The size of each chunk to send or -1 to send as a single chunk. Defaults to -1.

Returns¶

HttpResponse: The built HTTP response.

class method HttpResponse.from_json ¶

Summary¶

Create an HTTP response from data converted to JSON.

HttpResponse.from_json(

*,

status: int,

content_type: bytes,

headers: Optional[List[Tuple[bytes, bytes]]],

encode: Callable[[Any], str]

) -> HttpResponse

Parameters¶

status: int (optional)

An optional status code. Defaults to 200.

content_type: bytes (optional)

An optional content type. Defaults to b'application/json'.

headers: Optional[List[Tuple[bytes, bytes]]] (optional)

Optional headers. Defaults to None.

encode: Callable[[Any], str] (optional)

An optional function to convert the data to a JSON string. Defaults to json.dumps.

Returns¶

HttpResponse: description

class method HttpResponse.from_text ¶

Summary¶

Create an HTTP response from a text string.

HttpResponse.from_text(

*,

status: int,

content_type: bytes,

headers: Optional[List[Tuple[bytes, bytes]]],

encoding: str,

chunk_size: int

) -> HttpResponse

Parameters¶

status: int (optional)

An optional status. Defaults to 200.

content_type: bytes (optional)

An optional content type. Defaults to b'text/plain'.

headers: Optional[List[Tuple[bytes, bytes]]] (optional)

Optional headers. Defaults to None.

encoding: str (optional)

The encoding to apply when transforming the text into bytes. Defaults to 'utf-8'.

chunk_size: int (optional)

The size of each chunk to send or -1 to send as a single chunk. Defaults to -1.

Returns¶

HttpResponse: The built HTTP response.

class WebSocketRequest ¶

Summary¶

A WebSocket request.

bareasgi.websockets.WebSocketRequest(

scope: WebSocketScope,

info: Dict[str, Any],

context: Dict[str, Any],

matches: Mapping[str, Any],

web_socket: WebSocket

) -> None

Parameters¶

scope: WebSocketScope

The ASGI WebSocket scope.

info: Dict[str, Any]

The shared information from the Application.

context: Dict[str, Any]

The private information for this request or chain of requests.

matches: Mapping[str, Any]

Any path matches from the router.

web_socket: WebSocket

The WebSocket.

class WebSocket ¶

Summary¶

The interface for a server side WebSocket.

bareasgi.websockets.WebSocket() -> None

async method WebSocket.accept ¶

Summary¶

Accept the socket.

Description¶

This must be done before any other action is taken.

async WebSocket.accept(

subprotocol: Optional[str],

headers: Optional[List[Tuple[bytes, bytes]]]

) -> Any

Parameters¶

subprotocol: Optional[str] (optional)

An optional subprotocol sent by the client. Defaults to None.

headers: Optional[List[Tuple[bytes, bytes]]] (optional)

Optional headers to send. Defaults to None.

Returns¶

Any:

async method WebSocket.close ¶

Summary¶

Close the WebSocket.

async WebSocket.close(

code: int

) -> Any

Parameters¶

code: int (optional)

The reason code. Defaults to 1000.

Returns¶

Any:

async method WebSocket.receive ¶

Summary¶

Receive data from the WebSocket.

async WebSocket.receive() -> Optional[Union[bytes, str]]

Returns¶

Optional[Union[bytes, str]]: Either bytes of a string depending on the client.

async method WebSocket.send ¶

Summary¶

Send data to the client.

async WebSocket.send(

content: Union[bytes, str]

) -> Any

Parameters¶

content: Union[bytes, str]

Either bytes or a string.

Returns¶

Any:

class LifespanRequest ¶

Summary¶

Initialise a class holding a lifespan request.

bareasgi.lifespan.LifespanRequest(

scope: LifespanScope,

info: ASGILifespanReceiveEvent

) -> None

Parameters¶

scope: LifespanScope

The ASGI lifespan scope.

info: ASGILifespanReceiveEvent

The global info dictionary.