bareasgi

class Application(CoreApplication) ¶

Summary¶

Construct the application

Description¶

```python 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.Application(

*,

middlewares: HttpMiddlewares | None,

http_router: HttpRouter | None,

ws_middlewares: WebSocketMiddlewares | None,

ws_router: WebSocketRouter | None,

startup_handlers: Optional[List[LifespanHandler]],

shutdown_handlers: Optional[List[LifespanHandler]],

not_found_response: Optional[HttpResponse],

info: dict[str, Any] | None

) -> None

Parameters¶

middlewares: HttpMiddlewares | None

Optional middleware callbacks. Defaults to None.

http_router: HttpRouter | None

Optional router to for http routes. Defaults to None.

ws_middlewares: WebSocketMiddlewares | None (optional)

Optional middleware callbacks. Defaults to None.

ws_router: WebSocketRouter | None

Optional router for web routes. Defaults to None.

startup_handlers: Optional[List[LifespanHandler]]

Optional handlers to run at startup. Defaults to None.

shutdown_handlers: Optional[List[LifespanHandler]]

Optional handlers to run at shutdown. Defaults to None.

not_found_response: Optional[HttpResponse]

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

info: dict[str, Any] | None

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.request.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]

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.response.HttpResponse(

status: int,

headers: list[tuple[bytes, bytes]] | None,

body: AsyncIterable[bytes] | None,

pushes: Iterable[PushResponse] | None

) -> None

Parameters¶

status: int

The status code.

headers: list[tuple[bytes, bytes]] | None

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

body: AsyncIterable[bytes] | None

The body, if any. Defaults to None.

pushes: Iterable[PushResponse] | None

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: list[tuple[bytes, bytes]] | None,

chunk_size: int

) -> HttpResponse

Parameters¶

status: int (optional)

An optional status. Defaults to 200.

content_type: bytes

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

headers: list[tuple[bytes, bytes]] | None

Optional headers. Defaults to None.

chunk_size: int

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: list[tuple[bytes, bytes]] | None,

encode: Callable[[Any], str],

encode_bytes: Callable[[Any]], bytes] | None

) -> HttpResponse

Parameters¶

status: int (optional)

An optional status code. Defaults to 200.

content_type: bytes

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

headers: list[tuple[bytes, bytes]] | None

Optional headers. Defaults to None.

encode: Callable[[Any], str]

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

encode_bytes: Callable[[Any]], bytes] | None

An optional function to convert the data to JSON bytes. If specified this will be preferred to the encode argument. Defaults to None.

Returns¶

HttpResponse: A JSON http response.

class method HttpResponse.from_text ¶

Summary¶

Create an HTTP response from a text string.

HttpResponse.from_text(

*,

status: int,

content_type: bytes,

headers: list[tuple[bytes, bytes]] | None,

encoding: str,

chunk_size: int

) -> HttpResponse

Parameters¶

status: int (optional)

An optional status. Defaults to 200.

content_type: bytes

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

headers: list[tuple[bytes, bytes]] | None

Optional headers. Defaults to None.

encoding: str

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

chunk_size: int

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.request.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.WebSocket() -> None

property WebSocket.code ¶

Summary¶

The close code.

code -> int

property WebSocket.reason ¶

Summary¶

The reason for the closure.

reason -> str | None

property WebSocket.state ¶

Summary¶

The state of the WebSocket lifecycle.

state -> WebSocketState

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

) -> None

Parameters¶

subprotocol: Optional[str]

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

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

Optional headers to send. Defaults to None.

async method WebSocket.close ¶

Summary¶

Close the WebSocket.

async WebSocket.close(

code: int,

reason: str | None

) -> None

Parameters¶

code: int (optional)

The reason code. Defaults to 1000.

reason: str | None

The reason. Can be any string. Defaults to 1000.

async method WebSocket.receive ¶

Summary¶

Receive data from the WebSocket.

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

Returns¶

Optional[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: bytes | str

) -> None

Parameters¶

content: bytes | str

Either bytes or a string.

async method WebSocket.wait_closed ¶

Summary¶

Wait until the connection is closed.

async WebSocket.wait_closed() -> None

class LifespanRequest ¶

Summary¶

Initialise a class holding a lifespan request.

bareasgi.lifespan.request.LifespanRequest(

scope: LifespanScope,

info: ASGILifespanReceiveEvent

) -> None

Parameters¶

scope: LifespanScope

The ASGI lifespan scope.

info: ASGILifespanReceiveEvent

The global info dictionary.