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 或服务器端的存储。最后,返回处理后的响应对象。

相关阅读

评论可见,请评论后查看内容,谢谢!!!评论后请刷新页面。

大家都在找:

flask:flask框架官网

python:python入门教程(非常详细)

后端:西安软件公司招聘后端工程师

大家都在看: