quart.app 模块#

class quart.app.Quart(import_name: str, static_url_path: str | None = None, static_folder: str | None = 'static', static_host: str | None = None, host_matching: bool = False, subdomain_matching: bool = False, template_folder: str | None = 'templates', instance_path: str | None = None, instance_relative_config: bool = False, root_path: str | None = None)#

基础类：App

Web 框架类，处理请求并返回响应。

从服务角度来看，主要方法是 handle_request()，从应用程序角度来看，所有其他方法都很重要。

这可以通过多种方式扩展，大多数方法的设计都考虑了这一点。此外，任何列为属性的类都可以被替换。

aborter_class#: 用于通过 abort 辅助函数引发 HTTP 错误的类。

app_ctx_globals_class#: 用于 g 对象的类

asgi_http_class#

用于处理 ASGI HTTP 协议的类。

类型:: type[ASGIHTTPProtocol]

asgi_lifespan_class#

用于处理 ASGI 生命周期协议的类。

类型:: type[ASGILifespanProtocol]

asgi_websocket_class#

用于处理 ASGI websocket 协议的类。

类型:: type[ASGIWebsocketProtocol]

config_class#: 用于配置的类。

env#: 应用程序运行的环境名称。

event_class#: 用于以异步方式发出事件的类。

debug#: 对配置 DEBUG 值的包装，在许多情况下，如果为 True，将产生更多输出。如果未设置，如果 environ 设置为“development”，将激活调试模式。

jinja_environment#: 用于 jinja 环境的类。

jinja_options#: 在创建 jinja 环境时设置的默认选项。

permanent_session_lifetime#: 对配置 PERMANENT_SESSION_LIFETIME 值的包装。指定会话数据应存活多长时间。

request_class#: 用于请求的类。

response_class#: 用于响应的类。

secret_key#: 对配置 SECRET_KEY 值的包装。用于签名会话的应用程序密钥。

session_interface#: 用作会话接口的类。

shutdown_event#

此事件在应用程序开始关闭时设置，允许等待的任务知道何时停止。

类型:: 事件

url_map_class#: 将规则映射到端点的类。

url_rule_class#: 用于 URL 规则的类。

websocket_class#: 用于 websockets 的类。

aborter_class#: 别名 Aborter

add_background_task(func: Callable, *args: Any, **kwargs: Any) → None#

向应用程序添加 websocket url 规则。

这旨在直接在应用程序上使用。示例用法：

def websocket_route():
    ...

app.add_websocket('/', websocket_route)

参数:

rule – 要路由的路径，应以 / 开头。
endpoint – 可选的端点名称，如果不存在，则使用函数名称。
view_func – 返回响应的可调用对象。
defaults –
要自动提供的变量字典，用于为路由提供更简单的默认路径，例如，允许使用 /book 而不是 /book/0，
```
@app.websocket('/book', defaults={'page': 0})
@app.websocket('/book/<int:page>')
def book(page):
    ...
```
host – 此路由的完整主机名（应包含子域名，如果需要） - 不能与子域名一起使用。
subdomain – 此特定路由的子域名。
strict_slashes – 严格匹配路径中存在的尾部斜杠。将重定向叶子（无斜杠）到分支（有斜杠）。

after_serving(func: T_after_serving) → T_after_serving#

添加服务后函数。

这将允许在任何内容被服务后（最后一个字节发送后）调用提供的函数。

这旨在用作装饰器，如果用于装饰同步函数，则该函数将在 run_sync() 中被包装并在线程执行器中运行（并返回包装后的函数）。示例用法：

@app.after_serving
async def func():
    ...

参数:: func – 函数本身。

after_websocket(func: T_after_websocket) → T_after_websocket#

添加一个 websocket 后处理函数。

这旨在用作装饰器，如果用于装饰同步函数，则该函数将在 run_sync() 中被包装并在线程执行器中运行（并返回包装后的函数）。示例用法：

@app.after_websocket
async def func(response):
    return response

参数:: func – 实际的 websocket 后处理函数。

app_context() → AppContext#

创建并返回一个应用上下文。

这最好在上下文内使用，例如：

async with app.app_context():
    ...

app_ctx_globals_class#: _AppCtxGlobals 的别名

这处理 ASGI 调用，它可以被封装在中间件中。

当与 Quart 使用中间件时，最好包装此方法而不是应用本身。这是为了确保该应用是该类的实例 - 这使得 quart cli 能够正常工作。要使用此功能，只需执行以下操作：

app.asgi_app = middleware(app.asgi_app)

asgi_http_class#: ASGIHTTPConnection 的别名

asgi_lifespan_class#: ASGILifespan 的别名

asgi_websocket_class#: ASGIWebsocketConnection 的别名

before_serving(func: T_before_serving) → T_before_serving#

添加一个服务前处理函数。

这将允许在任何内容被服务之前（在任何字节被接收之前）调用提供的函数。

这旨在用作装饰器，如果用于装饰同步函数，则该函数将在 run_sync() 中被包装并在线程执行器中运行（并返回包装后的函数）。示例用法：

@app.before_serving
async def func():
    ...

参数:: func – 函数本身。

before_websocket(func: T_before_websocket) → T_before_websocket#

添加一个 websocket 前处理函数。

这旨在用作装饰器，如果用于装饰同步函数，则该函数将在 run_sync() 中被包装并在线程执行器中运行（并返回包装后的函数）。示例用法：

@app.before_websocket
async def func():
    ...

参数:: func – 实际的 websocket 前处理函数。

config_class#: Config 的别名

create_jinja_environment() → Environment#

创建并返回 Jinja 环境。

这将根据 jinja_options 和配置设置创建环境。环境默认将包含 Quart 全局变量。

create_url_adapter(request: BaseRequestWebsocket | None) → MapAdapter | None#

创建并返回一个 URL 适配器。

这将根据请求创建适配器，如果请求不存在，则使用应用程序配置。

default_config: dict[str, t.Any] = {'APPLICATION_ROOT': '/', 'BACKGROUND_TASK_SHUTDOWN_TIMEOUT': 5, 'BODY_TIMEOUT': 60, 'DEBUG': None, 'ENV': None, 'EXPLAIN_TEMPLATE_LOADING': False, 'MAX_CONTENT_LENGTH': 16777216, 'MAX_COOKIE_SIZE': 4093, 'PERMANENT_SESSION_LIFETIME': datetime.timedelta(days=31), 'PREFER_SECURE_URLS': False, 'PRESERVE_CONTEXT_ON_EXCEPTION': None, 'PROPAGATE_EXCEPTIONS': None, 'RESPONSE_TIMEOUT': 60, 'SECRET_KEY': None, 'SEND_FILE_MAX_AGE_DEFAULT': datetime.timedelta(seconds=43200), 'SERVER_NAME': None, 'SESSION_COOKIE_DOMAIN': None, 'SESSION_COOKIE_HTTPONLY': True, 'SESSION_COOKIE_NAME': 'session', 'SESSION_COOKIE_PATH': None, 'SESSION_COOKIE_SAMESITE': None, 'SESSION_COOKIE_SECURE': False, 'SESSION_REFRESH_EACH_REQUEST': True, 'TEMPLATES_AUTO_RELOAD': None, 'TESTING': False, 'TRAP_BAD_REQUEST_ERRORS': None, 'TRAP_HTTP_EXCEPTIONS': False}#

将请求分派到视图函数。

参数:: request_context – 请求上下文，可选，因为 Flask 省略了此参数。

将 websocket 分发到视图函数。

参数:: websocket_context – websocket 上下文，可选，与 Flask 规范匹配。

async do_teardown_appcontext(exc: BaseException | None) → None#: 拆卸应用程序（上下文），调用拆卸函数。

async do_teardown_request(exc: BaseException | None, request_context: RequestContext | None = None) → None#

拆卸请求，调用拆卸函数。

参数:

exc – 任何导致请求拆卸的未处理异常。
request_context – 请求上下文，可选，因为 Flask 省略了此参数。

async do_teardown_websocket(exc: BaseException | None, websocket_context: WebsocketContext | None = None) → None#

拆卸 websocket，调用拆卸函数。

参数:

exc – 任何导致 websocket 拆卸的未处理异常。
websocket_context – websocket 上下文，可选，因为 Flask 省略了此参数。

ensure_async(func: Callable[[P], Awaitable[T]]) → Callable[[P], Awaitable[T]]#

ensure_async(func: Callable[[P], T]) → Callable[[P], Awaitable[T]]

确保返回的 func 是异步的，并且调用 func。

在版本 0.11 中添加。

如果您希望更改同步函数的运行方式，请覆盖它。在 Quart 0.11 之前，这不是在执行器中运行同步代码。

event_class#: 是 Event 的别名

将视图响应返回值转换为响应。

参数:

result – 要最终化为响应的请求结果。
request_context – 请求上下文，可选，因为 Flask 省略了此参数。

将视图响应返回值转换为响应。

参数:

result – 将 WebSocket 结果转换为响应。
websocket_context – websocket 上下文，可选，因为 Flask 省略了此参数。

async full_dispatch_request(request_context: RequestContext | None = None) → Response | Response#

为请求分发添加前处理和后处理。

参数:: request_context – 请求上下文，可选，因为 Flask 省略了此参数。

async full_dispatch_websocket(websocket_context: WebsocketContext | None = None) → Response | Response | None#

为 WebSocket 分发添加前处理和后处理。

参数:: websocket_context – websocket 上下文，可选，与 Flask 规范匹配。

get_send_file_max_age(filename: str | None) → int | None#

由 send_file() 用于确定给定文件路径的 max_age 缓存值（如果未传递）。

默认情况下，这将从 current_app 的配置中返回 SEND_FILE_MAX_AGE_DEFAULT。这默认为 None，这告诉浏览器使用条件请求而不是定时缓存，这通常是比较好的选择。

请注意，这与 Quart 类中的相同方法重复。

async handle_background_exception(error: Exception) → None#

async handle_exception(error: Exception) → Response | Response#

处理未捕获的异常。

默认情况下，这会将错误响应切换为 500 内部服务器错误。

处理 HTTPException 子类错误。

这将尝试为错误找到一个处理程序，如果失败，将回退到错误响应。

async handle_request(request: Request) → Response | Response#

处理已引发的异常。

这应该将 HTTPException 转发到 handle_http_exception()，然后尝试处理错误。如果无法处理，则应重新抛出错误。

async handle_websocket(websocket: Websocket) → Response | Response | None#

async handle_websocket_exception(error: Exception) → Response | Response | None#

处理未捕获的异常。

默认情况下，这将记录异常并重新抛出异常。

jinja_environment#: Environment 的别名

lock_class#: Lock 的别名

log_exception(exception_info: tuple[type, BaseException, TracebackType] | tuple[None, None, None]) → None#

将异常记录到 logger。

默认情况下，这仅针对未处理的异常调用。

async make_default_options_response() → Response#: 这是 OPTIONS 请求的默认路由函数。

从路由处理程序的结果创建 Response。

结果本身可以是：

一个 Response 对象（或子类）。
一个包含 ResponseValue 和头字典的元组。
一个包含 ResponseValue、状态码和头字典的元组。

ResponseValue 是 Response 对象（或子类）或 str。

make_shell_context() → dict#

为交互式 shell 使用创建上下文。

shell_context_processors 可用于添加其他上下文。

async open_instance_resource(path: bytes | str | PathLike, mode: str = 'rb') → AiofilesContextManager[None, None, AsyncBufferedReader]#

打开一个文件进行读取。

使用：

async with await app.open_instance_resource(path) as file_:
    await file_.read()

async open_resource(path: bytes | str | PathLike, mode: str = 'rb') → AiofilesContextManager[None, None, AsyncBufferedReader]#

打开一个文件进行读取。

使用：

async with await app.open_resource(path) as file_:
    await file_.read()

async postprocess_websocket(response: Response | Response | None, websocket_context: WebsocketContext | None = None) → Response | Response#

在 websocket 结束运行后，对响应进行后处理。

参数:

response – websocket 结束运行后的响应。
websocket_context – websocket 上下文，可选，因为 Flask 省略了此参数。

预处理请求，即调用 before_request 函数。

参数:: request_context – 请求上下文，可选，因为 Flask 省略了此参数。

预处理 websocket，即在 before_websocket 函数调用之前执行。

参数:: websocket_context – websocket 上下文，可选，因为 Flask 省略了此参数。

async process_response(response: Response | Response, request_context: RequestContext | None = None) → Response | Response#

对请求进行后处理，作用于响应。

参数:

response – 请求完成后的响应。
request_context – 请求上下文，可选，因为 Flask 省略了此参数。

raise_routing_exception(request: BaseRequestWebsocket) → NoReturn#

request_class#: Request 的别名。

request_context(request: Request) → RequestContext#

创建并返回一个请求上下文。

在测试时使用 test_request_context()。这最好在上下文中使用，例如：

async with app.request_context(request):
    ...

参数:: request – 要构建上下文的请求。

response_class#: Response 的别名。

运行此应用程序。

此方法最适合开发环境，生产环境建议使用 Hypercorn 服务器。

参数:

host – 要监听的 hostname。默认情况下仅监听回环地址，使用 0.0.0.0 使服务器监听外部地址。
port – 要监听的端口号。
debug – 设置是否启用（或禁用）调试模式和调试输出。
use_reloader – 代码更改时自动重新加载。
loop – 用于创建服务器的 Asyncio loop，如果为 None，则使用默认 loop。如果指定了 loop，则调用者有责任关闭和清理 loop。
ca_certs – SSL CA 证书文件的路径。
certfile – SSL 证书文件的路径。
keyfile – SSL 密钥文件的路径。

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

返回一个任务，当 await 时运行此应用程序。

此方法最适合开发环境，生产环境建议使用 Hypercorn 服务器。

参数:

host – 要监听的 hostname。默认情况下仅监听回环地址，使用 0.0.0.0 使服务器监听外部地址。
port – 要监听的端口号。
debug – 设置是否启用（或禁用）调试模式和调试输出。
ca_certs – SSL CA 证书文件的路径。
certfile – SSL 证书文件的路径。
keyfile – SSL 密钥文件的路径。

async send_static_file(filename: str) → Response#

session_interface = <quart.sessions.SecureCookieSessionInterface object>#

async shutdown() → None#

shutdown_event: Event#

async startup() → None#

sync_to_async(func: Callable[[P], T]) → Callable[[P], Awaitable[T]]#

返回一个异步函数，该函数将运行同步函数 func。

此方法可按以下方式使用：

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

覆盖此方法以更改应用程序将同步代码转换为异步可调用方式的方法。

teardown_websocket(func: T_teardown) → T_teardown#

添加一个 teardown websocket 函数。此方法旨在用作装饰器，如果用于装饰同步函数，则该函数将被包装在 run_sync() 中并在线程执行器中运行（并返回包装后的函数）。示例用法，.. code-block:: python

@app.teardown_websocket async def func()

…

参数:: func – teardown websocket 函数本身。

test_app() → TestAppProtocol#

test_app_class#: TestApp 的别名

test_cli_runner(**kwargs: Any) → QuartCliRunner#: 创建并返回一个 CLI 测试运行器。

test_cli_runner_class#: 的别名 QuartCliRunner

test_client(use_cookies: bool = True, **kwargs: Any) → TestClientProtocol#: 创建并返回一个测试客户端。

test_client_class#: 的别名 QuartClient

test_request_context(path: str, *, method: str = 'GET', headers: dict | ~werkzeug.datastructures.headers.Headers | None = None, query_string: dict | None = None, scheme: str = 'http', send_push_promise: ~typing.Callable[[str, ~werkzeug.datastructures.headers.Headers], ~typing.Awaitable[None]] = <function no_op_push>, data: AnyStr | None = None, form: dict | None = None, json: ~typing.Any = <object object>, root_path: str = '', http_version: str = '1.1', scope_base: dict | None = None, auth: ~werkzeug.datastructures.auth.Authorization | tuple[str, str] | None = None, subdomain: str | None = None) → RequestContext#

为测试目的创建一个请求上下文。

这最适合在请求上下文中测试代码。它是 request_context() 的简化包装器。最好在 with 块中使用它，例如

async with app.test_request_context("/", method="GET"):
    ...

参数:

path – 请求路径。
method – HTTP 动词
headers – 要包含在请求中的标头。
query_string – 作为字典发送，或者可以从路径确定查询字符串。
scheme – 请求的方案，默认 http。

async update_template_context(context: dict) → None#

更新提供的模板上下文。

这将从各种模板上下文处理器添加额外的上下文。

参数:: context – 要更新（修改）的上下文。

url_for(endpoint: str, *, _anchor: str | None = None, _external: bool | None = None, _method: str | None = None, _scheme: str | None = None, **values: Any) → str#

返回特定端点的 URL。

这在模板和重定向中最有用，可以创建一个可在浏览器中使用的 URL。

参数:

endpoint – 要为其构建 URL 的端点，如果以 . 为前缀，则它会针对当前蓝图中的端点。
_anchor – 要附加的额外锚文本（例如 #text）。
_external – 为外部（应用程序）使用返回绝对 URL。
_method – 要与端点一起考虑的方法。
_scheme – 要使用的特定方案。
values – 要构建到 URL 中的值，如端点规则中所指定。

url_map_class#: 的别名 QuartMap

url_rule_class#: 的别名 QuartRule

websocket(rule: str, **options: Any) → Callable[[T_websocket], T_websocket]#

向应用程序添加一个 websocket。

这旨在用作装饰器，如果用于装饰同步函数，则该函数将在 run_sync() 中被包装并在线程执行器中运行（并返回包装后的函数）。示例用法：

@app.websocket('/')
async def websocket_route():
    ...

参数:

rule – 要路由的路径，应以 / 开头。
endpoint – 可选的端点名称，如果不存在，则使用函数名称。
defaults –
要自动提供的变量字典，用于为路由提供更简单的默认路径，例如，允许使用 /book 而不是 /book/0，
```
@app.websocket('/book', defaults={'page': 0})
@app.websocket('/book/<int:page>')
def book(page):
    ...
```
host – 此路由的完整主机名（应包含子域名，如果需要） - 不能与子域名一起使用。
subdomain – 此特定路由的子域名。
strict_slashes – 严格匹配路径中存在的尾部斜杠。将重定向叶子（无斜杠）到分支（有斜杠）。

websocket_class#: 的别名 Websocket

websocket_context(websocket: Websocket) → WebsocketContext#

创建并返回一个 websocket 上下文。

在测试时使用 test_websocket_context()。这最好在上下文中使用，例如

async with app.websocket_context(websocket):
    ...

参数:: websocket – 要在其周围构建上下文的 websocket。

while_serving(func: T_while_serving) → T_while_serving#

添加一个 while serving 生成器函数。

这将允许提供的生成器在启动时调用，然后在关闭时再次调用。

这被设计为用作装饰器。示例用法：

@app.while_serving
async def func():
    ...  # Startup
    yield
    ...  # Shutdown

参数:: func – 函数本身。