ASGI (Asynchronous Server Gateway Interface)¶

HTTPConnection¶

class baize.asgi.HTTPConnection(scope: ~typing.MutableMapping[str, ~typing.Any], receive: ~typing.Callable[[], ~typing.Awaitable[~typing.MutableMapping[str, ~typing.Any]]] = <function empty_receive>, send: ~typing.Callable[[~typing.MutableMapping[str, ~typing.Any]], ~typing.Awaitable[None]] = <function empty_send>)¶

A base class for incoming HTTP connections.

It is a valid Mapping type that allows you to directly access the values in any ASGI scope dictionary.

property client: Address¶

Client’s IP and Port.

Note that this depends on the “client” value given by the ASGI Server, and is not necessarily accurate.

property url: URL¶: The full URL of this request.

property path_params: Dict[str, Any]¶: The path parameters parsed by the framework.

property query_params: QueryParams¶: Query parameter. It is a multi-value mapping.

property headers: Headers¶

A read-only case-independent mapping.

Note that in its internal storage, all keys are in lower case.

property accepted_types: List[MediaType]¶: Request’s accepted types

accepts(media_type: str) → bool¶: e.g. request.accepts("application/json")

property content_length: Optional[int]¶: Request’s content-length

property content_type: ContentType¶: Request’s content-type

property cookies: Dict[str, str]¶

Returns cookies in as a dict.

NOTE: Modifications to this dictionary will not affect the response value. In fact, this value should not be modified.

property date: Optional[datetime]¶

The sending time of the request.

NOTE: The datetime object is timezone-aware.

get(k[, d]) → D[k] if k in D, else d. d defaults to None.¶

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

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

property referrer: Optional[URL]¶: The Referer HTTP request header contains an absolute or partial address of the page making the request.

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

Request¶

class baize.asgi.Request(scope: ~typing.MutableMapping[str, ~typing.Any], receive: ~typing.Callable[[], ~typing.Awaitable[~typing.MutableMapping[str, ~typing.Any]]] = <function empty_receive>, send: ~typing.Callable[[~typing.MutableMapping[str, ~typing.Any]], ~typing.Awaitable[None]] = <function empty_send>)¶

property method: str¶: HTTP method. Uppercase string.

async stream() → AsyncIterator[bytes]¶

Streaming read request body. e.g. async for chunk in request.stream(): ...

If you access .stream() then the byte chunks are provided without storing the entire body to memory. Any subsequent calls to .body, .form, or .json will raise an error.

property body: bytes¶: Read all the contents of the request body into the memory and return it.

property json: Any¶

Call await self.body and use json.loads parse it.

If content_type is not equal to application/json, an HTTPExcption exception will be thrown.

property form: FormData¶

Parse the data in the form format and return it as a multi-value mapping.

If content_type is equal to multipart/form-data, it will directly perform streaming analysis, and subsequent calls to self.body or self.json will raise errors.

If content_type is not equal to multipart/form-data or application/x-www-form-urlencoded, an HTTPExcption exception will be thrown.

async close() → None¶

Close all temporary files in the self.form.

This can always be called, regardless of whether you use form or not.

async is_disconnected() → bool¶: The method used to determine whether the connection is interrupted.

Response¶

class baize.asgi.Response(status_code: int = 200, headers: Optional[Mapping[str, str]] = None)¶: The parent class of all responses, whose objects can be used directly as ASGI application.

SmallResponse¶

class baize.asgi.SmallResponse(content: _ContentType, status_code: int = 200, headers: Optional[Mapping[str, str]] = None, media_type: Optional[str] = None, charset: Optional[str] = None)¶

Bases: Response, ABC, Generic[_ContentType]

Abstract base class for small response objects.

PlainTextResponse¶

class baize.asgi.PlainTextResponse(content: _ContentType, status_code: int = 200, headers: Optional[Mapping[str, str]] = None, media_type: Optional[str] = None, charset: Optional[str] = None)¶: Bases: SmallResponse[Union[bytes, str]]

HTMLResponse¶

class baize.asgi.HTMLResponse(content: _ContentType, status_code: int = 200, headers: Optional[Mapping[str, str]] = None, media_type: Optional[str] = None, charset: Optional[str] = None)¶: Bases: PlainTextResponse

JSONResponse¶

class baize.asgi.JSONResponse(content: Any, status_code: int = 200, headers: Optional[Mapping[str, str]] = None, **kwargs: Any)¶

Bases: SmallResponse[Any]

**kwargs is used to accept all the parameters that json.loads can accept.

RedirectResponse¶

class baize.asgi.RedirectResponse(url: Union[str, URL], status_code: int = 307, headers: Optional[Mapping[str, str]] = None)¶: Bases: Response

StreamResponse¶

class baize.asgi.StreamResponse(iterable: AsyncIterable[bytes], status_code: int = 200, headers: Optional[Mapping[str, str]] = None, content_type: str = 'application/octet-stream')¶: Bases: StreamingResponse[bytes]

FileResponse¶

class baize.asgi.FileResponse(filepath: str, headers: Optional[Mapping[str, str]] = None, content_type: Optional[str] = None, download_name: Optional[str] = None, stat_result: Optional[stat_result] = None, chunk_size: int = 262144)¶

Bases: Response, FileResponseMixin

File response.

It will automatically determine whether to send only headers and the range of files that need to be sent.

SendEventResponse¶

class baize.asgi.SendEventResponse(iterable: AsyncIterable[ServerSentEvent], status_code: int = 200, headers: Optional[Mapping[str, str]] = None, *, ping_interval: float = 3, charset: str = 'utf-8')¶

Bases: StreamingResponse[ServerSentEvent]

Server-sent events response.

When the cilent closes the connection, the generator will be closed. Use try-finally to clean up resources.

async def generator():
    try:
        while True:
            yield ServerSentEvent()
    finally:
        print("generator closed")

response = SendEventResponse(generator())

WebSocket¶

class baize.asgi.WebSocket(scope: MutableMapping[str, Any], receive: Callable[[], Awaitable[MutableMapping[str, Any]]], send: Callable[[MutableMapping[str, Any]], Awaitable[None]])¶

async receive() → MutableMapping[str, Any]¶: Receive ASGI websocket messages, ensuring valid state transitions.

async send(message: MutableMapping[str, Any]) → None¶: Send ASGI websocket messages, ensuring valid state transitions.

async accept(subprotocol: Optional[str] = None) → None¶: Accept websocket connection.

async receive_text() → str¶: Receive a WebSocket text frame and return.

async receive_bytes() → bytes¶: Receive a WebSocket binary frame and return.

async iter_text() → AsyncIterator[str]¶: Keep receiving text frames until the WebSocket connection is disconnected.

async iter_bytes() → AsyncIterator[bytes]¶: Keep receiving binary frames until the WebSocket connection is disconnected.

async send_text(data: str) → None¶: Send a WebSocket text frame.

async send_bytes(data: bytes) → None¶: Send a WebSocket binary frame.

async close(code: int = 1000, reason: Optional[str] = None) → None¶: Close WebSocket connection. It can be called multiple times.

WebsocketDenialResponse¶

class baize.asgi.WebsocketDenialResponse(response: Response)¶

A response that will deny a WebSocket connection.

https://asgi.readthedocs.io/en/latest/extensions.html#websocket-denial-response

Router¶

class baize.asgi.Router(*routes: Tuple[str, Interface])¶

A router to assign different paths to different ASGI applications.

applications = Router(
    ("/static/{filepath:any}", static_files),
    ("/api/{_:any}", api_app),
    ("/about/{name}", about_page),
    ("/", homepage),
)

Use {} to mark path parameters, the format is {name[:type]}. If type is not explicitly specified, it defaults to str.

The built-in types are str, int, decimal, uuid, date, any. Among them, str can match all strings except /, and any can match all strings.

If the built-in types are not enough, then you only need to write a class that inherits baize.routing.Convertor and register it in baize.routing.CONVERTOR_TYPES.

Subpaths¶

class baize.asgi.Subpaths(*routes: Tuple[str, Interface])¶

A router allocates different prefix requests to different ASGI applications.

NOTE: This will change the values of scope["root_path"] and scope["path"].

applications = Subpaths(
    ("/static", static_files),
    ("/api", api_app),
    ("", default_app),
)

Hosts¶

class baize.asgi.Hosts(*hosts: Tuple[str, Interface])¶

A router that distributes requests to different ASGI applications based on Host.

applications = Hosts(
    (r"static\.example\.com", static_files),
    (r"api\.example\.com", api_app),
    (r"(www\.)?example\.com", default_app),
)

Shortcut functions¶

request_response¶

baize.asgi.request_response(view: Callable[[Request], Awaitable[Response]]) → ASGIApp¶

This can turn a callable object into a ASGI application.

@request_response
async def f(request: Request) -> Response:
    ...

decorator¶

baize.asgi.decorator(handler: Callable[[Request, Callable[[Request], Awaitable[Response]]], Awaitable[Response]]) → Callable[[Callable[[Request], Awaitable[Response]]], Callable[[Request], Awaitable[Response]]]¶

This can turn a callable object into a decorator for view.

@decorator
async def m(request: Request, next_call: Callable[[Request], Awaitable[Response]]) -> Response:
    ...
    response = await next_call(request)
    ...
    return response


@request_response
@m
async def v(request: Request) -> Response:
    ...

middleware¶

baize.asgi.middleware(handler: Callable[[NextRequest, Callable[[NextRequest], Awaitable[NextResponse]]], Awaitable[Response]]) → Callable[[ASGIApp], ASGIApp]¶

This can turn a callable object into a middleware for ASGI application.

@middleware
async def m(
    request: NextRequest, next_call: Callable[[NextRequest], Awaitable[NextResponse]]
) -> Response:
    ...
    response = await next_call(request)
    ...
    return response

@m
@request_response
async def v(request: Request) -> Response:
    ...

# OR

@m
async def asgi(scope: Scope, receive: Receive, send: Send) -> None:
    ...

websocket_session¶

baize.asgi.websocket_session(view: Callable[[WebSocket], Awaitable[None]]) → ASGIApp¶

This can turn a callable object into a ASGI application.

@websocket_session
async def f(websocket: WebSocket) -> None:
    ...

Files¶

class baize.asgi.Files(directory: Union[str, PathLike[str]], package: Optional[str] = None, *, handle_404: Optional[Interface] = None, cacheability: Literal['public', 'private', 'no-cache', 'no-store'] = 'public', max_age: int = 600)¶

Provide the ASGI application to download files in the specified path or the specified directory under the specified package.

Support request range and cache (304 status code).

Pages¶

class baize.asgi.Pages(directory: Union[str, PathLike[str]], package: Optional[str] = None, *, handle_404: Optional[Interface] = None, cacheability: Literal['public', 'private', 'no-cache', 'no-store'] = 'public', max_age: int = 600)¶

Provide the ASGI application to download files in the specified path or the specified directory under the specified package. Unlike Files, when you visit a directory, it will try to return the content of the file named index.html in that directory.

Support request range and cache (304 status code).

ASGI (Asynchronous Server Gateway Interface)¶

HTTPConnection¶

Request¶

Response¶

SmallResponse¶

PlainTextResponse¶

HTMLResponse¶

JSONResponse¶

RedirectResponse¶

StreamResponse¶

FileResponse¶

SendEventResponse¶

WebSocket¶

WebsocketDenialResponse¶

Router¶

Subpaths¶

Hosts¶

Shortcut functions¶

request_response¶

decorator¶

middleware¶

websocket_session¶

Files¶

Pages¶

BáiZé

Navigation

Related Topics