quart_trio.app module

class quart_trio.app.QuartTrio(import_name: str, static_url_path: Optional[str] = None, static_folder: Optional[str] = 'static', static_host: Optional[str] = None, host_matching: bool = False, subdomain_matching: bool = False, template_folder: Optional[str] = 'templates', instance_path: Optional[str] = None, instance_relative_config: bool = False, root_path: Optional[str] = None)

Bases: quart.app.Quart

after_request_funcs: Dict[AppOrBlueprintKey, List[AfterRequestCallable]]
after_serving_funcs: List[Callable[], Awaitable[None]]]
after_websocket_funcs: Dict[AppOrBlueprintKey, List[AfterWebsocketCallable]]
asgi_http_class

alias of quart_trio.asgi.TrioASGIHTTPConnection

asgi_lifespan_class

alias of quart_trio.asgi.TrioASGILifespan

asgi_websocket_class

alias of quart_trio.asgi.TrioASGIWebsocketConnection

before_first_request_funcs: List[BeforeRequestCallable]
before_request_funcs: Dict[AppOrBlueprintKey, List[BeforeRequestCallable]]
before_serving_funcs: List[Callable[], Awaitable[None]]]
before_websocket_funcs: Dict[AppOrBlueprintKey, List[BeforeWebsocketCallable]]
blueprints: Dict[str, Blueprint]
error_handler_spec: Dict[AppOrBlueprintKey, Dict[Optional[int], Dict[Type[Exception], ErrorHandlerCallable]]]
extensions: Dict[str, Any]
async full_dispatch_request(request_context: Optional[quart.ctx.RequestContext] = None) → Union[quart.wrappers.response.Response, werkzeug.wrappers.response.Response]

Adds pre and post processing to the request dispatching.

Parameters

request_context – The request context, optional as Flask omits this argument.

async full_dispatch_websocket(websocket_context: Optional[quart.ctx.WebsocketContext] = None) → Optional[Union[quart.wrappers.response.Response, werkzeug.wrappers.response.Response]]

Adds pre and post processing to the websocket dispatching.

Parameters

websocket_context – The websocket context, optional to match the Flask convention.

async handle_request(request: quart.wrappers.request.Request) → Union[quart.wrappers.response.Response, werkzeug.wrappers.response.Response]
async handle_user_exception(error: Union[Exception, trio.MultiError]) → Union[werkzeug.exceptions.HTTPException, Response, WerkzeugResponse, AnyStr, Dict[str, Any], AsyncGenerator[AnyStr, None], Generator[AnyStr, None, None], Tuple[Union[Response, WerkzeugResponse, AnyStr, Dict[str, Any], AsyncGenerator[AnyStr, None], Generator[AnyStr, None, None]], Union[Headers, Dict[str, Union[str, List[str], Tuple[str, ...]]], List[Tuple[str, Union[str, List[str], Tuple[str, ...]]]]]], Tuple[Union[Response, WerkzeugResponse, AnyStr, Dict[str, Any], AsyncGenerator[AnyStr, None], Generator[AnyStr, None, None]], int], Tuple[Union[Response, WerkzeugResponse, AnyStr, Dict[str, Any], AsyncGenerator[AnyStr, None], Generator[AnyStr, None, None]], int, Union[Headers, Dict[str, Union[str, List[str], Tuple[str, ...]]], List[Tuple[str, Union[str, List[str], Tuple[str, ...]]]]]]]

Handle an exception that has been raised.

This should forward HTTPException to handle_http_exception(), then attempt to handle the error. If it cannot it should reraise the error.

async handle_websocket(websocket: quart.wrappers.websocket.Websocket) → Optional[Union[quart.wrappers.response.Response, werkzeug.wrappers.response.Response]]
lock_class

alias of trio.Lock

nursery: trio.Nursery
request_class

alias of quart_trio.wrappers.request.TrioRequest

response_class

alias of quart_trio.wrappers.response.TrioResponse

run(host: str = '127.0.0.1', port: int = 5000, debug: Optional[bool] = None, use_reloader: bool = True, ca_certs: Optional[str] = None, certfile: Optional[str] = None, keyfile: Optional[str] = None, **kwargs: Any) → None

Run this application.

This is best used for development only, see using Gunicorn for production servers.

Parameters

  • host – Hostname to listen on. By default this is loopback only, use 0.0.0.0 to have the server listen externally.

  • port – Port number to listen on.

  • debug – If set enable (or disable) debug mode and debug output.

  • use_reloader – Automatically reload on code changes.

  • ca_certs – Path to the SSL CA certificate file.

  • certfile – Path to the SSL certificate file.

  • ciphers – Ciphers to use for the SSL setup.

  • keyfile – Path to the SSL key file.

run_task(host: str = '127.0.0.1', port: int = 5000, debug: Optional[bool] = None, use_reloader: bool = True, ca_certs: Optional[str] = None, certfile: Optional[str] = None, keyfile: Optional[str] = None, shutdown_trigger: Optional[Callable[[...], Awaitable[None]]] = None) → Coroutine[None, None, None]

Return a task that when awaited runs this application.

This is best used for development only, see Hypercorn for production servers.

Parameters

  • host – Hostname to listen on. By default this is loopback only, use 0.0.0.0 to have the server listen externally.

  • port – Port number to listen on.

  • debug – If set enable (or disable) debug mode and debug output.

  • use_reloader – Automatically reload on code changes.

  • loop – Asyncio loop to create the server in, if None, take default one. If specified it is the caller’s responsibility to close and cleanup the loop.

  • ca_certs – Path to the SSL CA certificate file.

  • certfile – Path to the SSL certificate file.

  • keyfile – Path to the SSL key file.

shell_context_processors: List[Callable[], Dict[str, Any]]]
sync_to_async(func: Callable[[...], Any]) → Callable[[...], Awaitable[Any]]

Return a async function that will run the synchronous function func.

This can be used as so,:

result = await app.sync_to_async(func)(*args, **kwargs)

Override this method to change how the app converts sync code to be asynchronously callable.

teardown_appcontext_funcs: List[TeardownCallable]
teardown_request_funcs: Dict[AppOrBlueprintKey, List[TeardownCallable]]
teardown_websocket_funcs: Dict[AppOrBlueprintKey, List[TeardownCallable]]
template_context_processors: Dict[AppOrBlueprintKey, List[TemplateContextProcessorCallable]]
test_app_class

alias of quart_trio.testing.app.TrioTestApp

test_client_class

alias of quart_trio.testing.client.TrioClient

url_build_error_handlers: List[Callable[[Exception, str, dict], str]]
url_default_functions: Dict[AppOrBlueprintKey, List[URLDefaultCallable]]
url_value_preprocessors: Dict[AppOrBlueprintKey, List[URLValuePreprocessorCallable]]
view_functions: Dict[str, Callable]
websocket_class

alias of quart_trio.wrappers.websocket.TrioWebsocket

while_serving_gens: List[AsyncGenerator[None, None]]
quart_trio.wrappers.websocket module quart_trio.asgi module