def ensure_sync(self, func: t.Callable[..., t.Any]) -> t.Callable[..., t.Any]: if iscoroutinefunction(func): return self.async_to_sync(func)
return func
def async_to_sync( self, func: t.Callable[..., t.Coroutine[t.Any, t.Any, t.Any]] ) -> t.Callable[..., t.Any]: try: from asgiref.sync import async_to_sync as asgiref_async_to_sync except ImportError: raise RuntimeError( "Install Flask with the 'async' extra in order to use async views." ) from None
return asgiref_async_to_sync(func)
两个方法ensure_sync 和 async_to_sync主要用于处理Python的异步函数（async functions）和同步函数（sync functions）之间的转换，使得无论输入的函数是异步的还是同步的，都能以同步的方式调用它，这对于在同步环境中处理可能包含异步函数的代码库特别有用。在Python中，异步函数是一种特殊的函数，它使用async def定义，并且可以通过await关键字调用其他异步函数。这种函数主要用于异步I/O操作，如网络请求或数据库查询，以提高程序的并发性能。 1.ensure_sync 函数： 这个函数接受一个函数func作为输入，检查它是否是一个异步函数。如果是，它会调用self.async_to_sync(func)将这个异步函数转换为同步函数；如果不是，它会直接返回这个函数。这里t.Callable[..., t.Any]表示接受任意数量和类型的参数，并返回任意类型的函数。t.Coroutine[t.Any, t.Any, t.Any]表示一个异步函数，它接受任意类型的参数，可能在执行中产生任意类型的中间结果，并最终返回任意类型的值。
2.async_to_sync 函数： 这个函数接受一个异步函数func作为输入，并从asgiref.sync模块导入async_to_sync函数。如果导入失败，则抛出一个运行时错误，提示用户安装带有'async'额外项的Flask。如果导入成功，则使用asgiref_async_to_sync函数将输入的异步函数func转换为同步函数，并返回转换后的函数。 asgiref.sync.async_to_sync是一个常见的工具，用于在同步代码中调用异步函数。它将异步函数的执行封装在一个线程或进程中，以便在同步代码中同步地等待异步函数的完成。
def url_for( self, /, endpoint: str, *, _anchor: str | None = None, _method: str | None = None, _scheme: str | None = None, _external: bool | None = None, **values: t.Any, ) -> str: req_ctx = _cv_request.get(None)
if req_ctx is not None: url_adapter = req_ctx.url_adapter blueprint_name = req_ctx.request.blueprint
if endpoint[:1] == ".": if blueprint_name is not None: endpoint = f"{blueprint_name}{endpoint}" else: endpoint = endpoint[1:]
if _external is None: _external = _scheme is not None else: app_ctx = _cv_app.get(None)
if app_ctx is not None: url_adapter = app_ctx.url_adapter else: url_adapter = self.create_url_adapter(None)
if url_adapter is None: raise RuntimeError( "Unable to build URLs outside an active request" " without 'SERVER_NAME' configured. Also configure" " 'APPLICATION_ROOT' and 'PREFERRED_URL_SCHEME' as" " needed." )
if _external is None: _external = True
if _scheme is not None and not _external: raise ValueError("When specifying '_scheme', '_external' must be True.")
self.inject_url_defaults(endpoint, values)
try: rv = url_adapter.build( # type: ignore[union-attr] endpoint, values, method=_method, url_scheme=_scheme, force_external=_external, ) except BuildError as error: values.update( _anchor=_anchor, _method=_method, _scheme=_scheme, _external=_external ) return self.handle_url_build_error(error, endpoint, values)
if _anchor is not None: _anchor = _url_quote(_anchor, safe="%!#$&'()*+,/:;=?@") rv = f"{rv}#{_anchor}"
return rv
url_for 是 Flask 框架中用于生成 URL 的一个非常有用的函数。它接收一个或多个参数，包括端点（endpoint）名称和可选的关键字参数，确保了无论是在请求上下文中还是在应用上下文中，都可以正确地生成 URL。同时，它也提供了灵活的配置选项，比如生成外部 URL 或包含锚点。具体实现如下：
确定上下文：
首先，它尝试从当前请求的上下文中获取 url_adapter 和 blueprint_name（蓝图名称）。如果请求上下文不存在，则尝试从应用上下文中获取 url_adapter。如果在应用上下文中也找不到 url_adapter，则尝试创建一个新的 url_adapter。如果无法创建 url_adapter（因为没有配置 SERVER_NAME），则抛出一个运行时错误。处理端点：
如果提供的端点以 . 开头，则它表示相对于当前蓝图的端点。因此，代码会检查 blueprint_name，如果存在，则将其添加到端点前面；否则，去掉 .。处理外部 URL：
如果 _external 参数没有显式指定，代码会根据 _scheme 参数是否被设置来设置 _external 的值。如果 _scheme 被设置了，则 _external 会被设置为 True，意味着需要生成一个绝对 URL（包含协议和域名）。构建 URL：
使用 url_adapter 的 build 方法来生成 URL。这里可以指定多种参数，比如方法（_method）、URL 方案（_scheme）和是否强制生成外部 URL（_external）。处理锚点：
如果 _anchor 参数被提供了，它会在生成的 URL 后面添加一个锚点（以 # 开头的部分）。错误处理：
如果在构建 URL 时发生错误（比如端点不存在），则会调用 handle_url_build_error 方法来处理这个错误。
def make_response(self, rv: ft.ResponseReturnValue) -> Response: status = headers = None
if isinstance(rv, tuple): len_rv = len(rv)
if len_rv == 3: rv, status, headers = rv # type: ignore[misc] elif len_rv == 2: if isinstance(rv[1], (Headers, dict, tuple, list)): rv, headers = rv else: rv, status = rv # type: ignore[assignment,misc] else: raise TypeError( "The view function did not return a valid response tuple." " The tuple must have the form (body, status, headers)," " (body, status), or (body, headers)." )
if rv is None: raise TypeError( f"The view function for {request.endpoint!r} did not" " return a valid response. The function either returned" " None or ended without a return statement." )
if not isinstance(rv, self.response_class): if isinstance(rv, (str, bytes, bytearray)) or isinstance(rv, cabc.Iterator): rv = self.response_class( rv, status=status, headers=headers, # type: ignore[arg-type] ) status = headers = None elif isinstance(rv, (dict, list)): rv = self.json.response(rv) elif isinstance(rv, BaseResponse) or callable(rv): try: rv = self.response_class.force_type( rv, # type: ignore[arg-type] request.environ, ) except TypeError as e: raise TypeError( f"{e}\nThe view function did not return a valid" " response. The return type must be a string," " dict, list, tuple with headers or status," " Response instance, or WSGI callable, but it" f" was a {type(rv).__name__}." ).with_traceback(sys.exc_info()[2]) from None else: raise TypeError( "The view function did not return a valid" " response. The return type must be a string," " dict, list, tuple with headers or status," " Response instance, or WSGI callable, but it was a" f" {type(rv).__name__}." )
rv = t.cast(Response, rv) if status is not None: if isinstance(status, (str, bytes, bytearray)): rv.status = status else: rv.status_code = status
if headers: rv.headers.update(headers) # type: ignore[arg-type]
return rv
make_response 方法的主要作用是确保视图函数返回了一个有效的响应，并将其转换为一个 Response 对象，以便 Flask 可以将其发送回客户端。这有助于简化 Flask 应用中的视图函数编写，并确保返回给客户端的 HTTP 响应始终是有效和一致的。 1.初始化变量: 初始化两个变量 status 和 headers，分别用于存储 HTTP 状态码和 HTTP 头部信息。 2. 处理返回值为元组的情况: 如果视图函数返回的是一个元组，代码会检查元组的长度和内容,元组内容应包含响应体（`body`）、状态码（`status`）和头部信息（`headers`）。 如果元组的第二个元素是 `Headers`、`dict`、`tuple` 或 `list` 类型，那么它被视为头部信息；否则，它被视为状态码。如果元组的长度既不是2也不是3，则抛出一个 `TypeError` 异常。 3. 处理返回值为 None 的情况: 如果视图函数返回 None，则抛出一个 TypeError 异常。这是因为 Flask 需要一个有效的响应体来构建 HTTP 响应。 4.检查返回值rv类型: 如果视图函数的返回值rv不是 Response类的实例，则进行进一步的处理。 5. 处理字符串、字节、字节数组和迭代器: 如果 rv 是字符串、字节、字节数组或迭代器，则将其封装为 Response 对象，并清除 status 和 headers 变量，因为它们已经被包含在 Response 对象中了。 6. 处理字典和列表: 如果 rv 是字典或列表，通常表示 JSON 响应。这里使用 self.json.response(rv) 将其转换为 JSON 格式的 Response 对象。 7. 处理 BaseResponse 实例或可调用对象: 如果 rv 是 BaseResponse 的实例或可调用对象，尝试将其转换为 Response 对象。如果转换失败，则抛出异常。 8. 处理其他类型的返回值: 如果 rv 是其他类型的值，则抛出一个 TypeError 异常，说明视图函数返回了无效的响应类型。 9. 类型断言: 这里使用类型断言将 rv 断言为 Response 类型。这有助于在后续代码中确保 rv 是 Response 对象。 10. 设置状态码: 如果 status 不为 None，则将其设置为 Response 对象的状态码。注意这里区分了字符串和整数类型的状态码。 11. 设置头部信息: 如果 headers 不为空，则将其更新到 Response 对象的头部信息中。 12. 最后，返回处理后的 Response 对象，以便 Flask 可以将其发送给客户端。
def preprocess_request(self) -> ft.ResponseReturnValue | None: names = (None, *reversed(request.blueprints))
for name in names: if name in self.url_value_preprocessors: for url_func in self.url_value_preprocessors[name]: url_func(request.endpoint, request.view_args)
for name in names: if name in self.before_request_funcs: for before_func in self.before_request_funcs[name]: rv = self.ensure_sync(before_func)()
if rv is not None: return rv # type: ignore[no-any-return]
return None
def process_response(self, response: Response) -> Response: ctx = request_ctx._get_current_object() # type: ignore[attr-defined]
for func in ctx._after_request_functions: response = self.ensure_sync(func)(response)
for name in chain(request.blueprints, (None,)): if name in self.after_request_funcs: for func in reversed(self.after_request_funcs[name]): response = self.ensure_sync(func)(response)
if not self.session_interface.is_null_session(ctx.session): self.session_interface.save_session(self, ctx.session, response)
return response
preprocess_request 方法在请求处理之前执行，主要用于运行预处理器函数。
1.获取蓝图名称： 这里创建了一个名称元组，首先是None，然后是请求中涉及的蓝图（blueprints）的逆序。蓝图在 Flask 中用于组织大型应用的视图、模板和其他资源。 2. 运行 URL 值预处理器： 对于每个蓝图名称（包括 None），如果它存在于url_value_preprocessors 字典中，就运行与该名称关联的所有预处理器函数。这些函数通常用于修改请求中的视图参数。 3. 运行请求前函数： 与URL值预处理器类似，但这次运行的是before_request_funcs中的函数。这些函数在请求处理之前运行，并可能返回响应对象，如果返回了非None的值，则直接返回该值并停止进一步处理。
process_response 方法在请求处理完成后执行，主要用于运行后处理器函数并保存会话。
1.运行上下文后处理器： 对于当前请求上下文中的每个后处理器函数，使用它来修改响应对象。 2. 运行蓝图后处理器： 对于每个蓝图名称（包括 None），如果它存在于after_request_funcs字典中，就运行与该名称关联的所有后处理器函数。这些函数也用于修改响应对象。 3. 保存会话： 如果当前会话不是空会话，则保存会话。这通常涉及将会话数据写入客户端的 cookie 或服务器端的存储。最后，返回处理后的响应对象。
推荐文章
您阅读本篇文章共花了：
发表评论