仅当作为 FastAPI 调用的一部分返回时才排除 pydantic 模型字段

2024-01-08

Context

我有一个非常复杂的 pydantic 模型,其中有很多嵌套的 pydantic 模型。我想确保某些字段永远不会作为 API 调用的一部分返回,但我希望这些字段用于内部逻辑。

我尝试过的

我首先尝试使用 pydanticField函数来指定exclude在我不想返回的字段上标记。这有效,但是我的内部逻辑中的函数必须在调用时覆盖它.dict()通过致电.dict(exclude=None).

相反,我指定了一个自定义标志return_in_api on the Field,目标是仅在调用 FastAPI 时应用排除.dict()。我尝试编写一个中间件来调用.dict()并通过我自己的exclude基于包含的嵌套字段的属性return_in_api=False。然而,FastAPI 的中间件为我提供了一个响应流,我不想过早地解析它。

相反,我写了一个装饰器,叫做.dict()在我的路由处理程序的返回值上具有适当的exclude value.

Problem

一个挑战是,每当添加新端点时,添加它们的人都必须记住包含此装饰器,否则字段会泄漏。

理想情况下,我想将此装饰器应用于每个路由,但通过中间件执行此操作似乎会破坏响应流。


系统地排除所有路线的字段

我发现最好使用一个具体但非常简单的示例。假设您有以下模型:

from pydantic import BaseModel, Field


class Demo(BaseModel):
    foo: str
    bar: str = Field(return_in_api=False)

我们希望确保bar永远不会在响应中返回,无论是在response_model明确作为参数提供给路由装饰器and当它只是设置为路由处理函数的返回注释时。 (假设我们不想使用内置的exclude无论出于何种原因,我们的字段的参数。)

我发现最可靠的方法是子类化fastapi.routing.APIRoute https://github.com/tiangolo/fastapi/blob/0.95.1/fastapi/routing.py#L322并钩入其__init__方法。通过复制父类的一小部分代码,我们可以确保我们始终获得正确的响应模型。一旦我们有了这个,只需设置路线的问题response_model_exclude调用父构造函数之前的参数。

这是我的建议:

from collections.abc import Callable
from typing import Any

from fastapi.responses import Response
from fastapi.dependencies.utils import get_typed_return_annotation, lenient_issubclass
from fastapi.routing import APIRoute, Default, DefaultPlaceholder


class CustomAPIRoute(APIRoute):
    def __init__(
        self,
        path: str,
        endpoint: Callable[..., Any],
        *,
        response_model: Any = Default(None),
        **kwargs: Any,
    ) -> None:
        # We need this part to ensure we get the response model,
        # even if it is just set as an annotation on the handler function.
        if isinstance(response_model, DefaultPlaceholder):
            return_annotation = get_typed_return_annotation(endpoint)
            if lenient_issubclass(return_annotation, Response):
                response_model = None
            else:
                response_model = return_annotation
        # Find the fields to exclude:
        if response_model is not None:
            kwargs["response_model_exclude"] = {
                name
                for name, field in response_model.__fields__.items()
                if field.field_info.extra.get("return_in_api") is False
            }
        super().__init__(path, endpoint, response_model=response_model, **kwargs)

我们现在可以在路由器上设置自定义路由类(文档 https://fastapi.tiangolo.com/advanced/custom-request-and-route/)。这样它将用于all其路线:

from fastapi import FastAPI
# ... import CustomAPIRoute
# ... import Demo

api = FastAPI()
api.router.route_class = CustomAPIRoute


@api.get("/demo1")
async def demo1() -> Demo:
    return Demo(foo="a", bar="b")


@api.get("/demo2", response_model=Demo)
async def demo2() -> dict[str, Any]:
    return {"foo": "x", "bar": "y"}

尝试这个简单的 API 示例uvicorn and GET设置端点/demo1 and /demo2产生响应{"foo":"a"} and {"foo":"x"}分别。


确保模式一致性

然而值得一提的是(除非我们采取额外的步骤)bar场将仍然是架构的一部分。这意味着例如为这两个端点自动生成的 OpenAPI 文档将会呈现 bar作为预期响应的顶级属性。

这不是您问题的一部分,因此我认为您已经意识到这一点并正在采取措施确保一致性。如果没有,对于阅读本文的其他人,您可以定义一个静态schema_extra https://docs.pydantic.dev/usage/schema/#schema-customization方法上的Config您的基本模型的删除那些在返回模式之前永远不会“向外部”显示的字段:

from typing import Any
from pydantic import BaseModel, Field


class CustomBaseModel(BaseModel):
    class Config:
        @staticmethod
        def schema_extra(schema: dict[str, Any]) -> None:
            properties = schema.get("properties", {})
            to_delete = set()
            for name, prop in properties.items():
                if prop.get("return_in_api") is False:
                    to_delete.add(name)
            for name in to_delete:
                del properties[name]


class Demo(CustomBaseModel):
    foo: str
    bar: str = Field(return_in_api=False)
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)

仅当作为 FastAPI 调用的一部分返回时才排除 pydantic 模型字段 的相关文章

  • 使用 python requests 模块时出现 HTTP 503 错误

    我正在尝试发出 HTTP 请求 但当前可以从 Firefox 浏览器访问的网站响应 503 错误 代码本身非常简单 在网上搜索一番后我添加了user Agent请求参数 但也没有帮助 有人能解释一下如何消除这个 503 错误吗 顺便说一句
  • 元组有什么用?

    我现在正在学习 Python 课程 我们刚刚介绍了元组作为数据类型之一 我阅读了它的维基百科页面 但是 我无法弄清楚这种数据类型在实践中会有什么用处 我可以提供一些需要一组不可变数字的示例吗 也许是在 Python 中 这与列表有何不同 每
  • 在 django ORM 中查询时如何将 char 转换为整数?

    最近开始使用 Django ORM 我想执行这个查询 select student id from students where student id like 97318 order by CAST student id as UNSIG
  • Python 中的舍入浮点问题

    我遇到了 np round np around 的问题 它没有正确舍入 我无法包含代码 因为当我手动设置值 而不是使用我的数据 时 返回有效 但这是输出 In 177 a Out 177 0 0099999998 In 178 np rou
  • 跟踪 pypi 依赖项 - 谁在使用我的包

    无论如何 是否可以通过 pip 或 PyPi 来识别哪些项目 在 Pypi 上发布 可能正在使用我的包 也在 PyPi 上发布 我想确定每个包的用户群以及可能尝试积极与他们互动 预先感谢您的任何答案 即使我想做的事情是不可能的 这实际上是不
  • 删除flask中的一对一关系

    我目前正在使用 Flask 开发一个应用程序 并且在删除一对一关系中的项目时遇到了一个大问题 我的模型中有以下结构 class User db Model tablename user user id db Column db String
  • 独立滚动矩阵的行

    我有一个矩阵 准确地说 是 2d numpy ndarray A np array 4 0 0 1 2 3 0 0 5 我想滚动每一行A根据另一个数组中的滚动值独立地 r np array 2 0 1 也就是说 我想这样做 print np
  • Pandas Merge (pd.merge) 如何设置索引和连接

    我有两个 pandas 数据框 dfLeft 和 dfRight 以日期作为索引 dfLeft cusip factorL date 2012 01 03 XXXX 4 5 2012 01 03 YYYY 6 2 2012 01 04 XX
  • Python 2:SMTPServerDisconnected:连接意外关闭

    我在用 Python 发送电子邮件时遇到一个小问题 me my email address you recipient s email address me email protected cdn cgi l email protectio
  • pyspark 将 twitter json 流式传输到 DF

    我正在从事集成工作spark streaming with twitter using pythonAPI 我看到的大多数示例或代码片段和博客是他们从Twitter JSON文件进行最终处理 但根据我的用例 我需要所有字段twitter J
  • Python3 在 DirectX 游戏中移动鼠标

    我正在尝试构建一个在 DirectX 游戏中执行一些操作的脚本 除了移动鼠标之外 我一切都正常 是否有任何可用的模块可以移动鼠标 适用于 Windows python 3 Thanks I used pynput https pypi or
  • 仅第一个加载的 Django 站点有效

    我最近向 stackoverflow 提交了一个问题 标题为使用mod wsgi在apache上多次请求后Django无限加载 https stackoverflow com questions 71705909 django infini
  • 为什么 Pickle 协议 4 中的 Pickle 文件是协议 3 中的两倍,而速度却没有任何提升?

    我正在测试 Python 3 4 我注意到 pickle 模块有一个新协议 因此 我对 2 个协议进行了基准测试 def test1 pickle3 open pickle3 wb for i in range 1000000 pickle
  • 如何在 pygtk 中创建新信号

    我创建了一个 python 对象 但我想在它上面发送信号 我让它继承自 gobject GObject 但似乎没有任何方法可以在我的对象上创建新信号 您还可以在类定义中定义信号 class MyGObjectClass gobject GO
  • 如何解决 PDFBox 没有 unicode 映射错误?

    我有一个现有的 PDF 文件 我想使用 python 脚本将其转换为 Excel 文件 目前正在使用PDFBox 但是存在多个类似以下错误 org apache pdfbox pdmodel font PDType0Font toUnico
  • 在本地网络上运行 Bokeh 服务器

    我有一个简单的 Bokeh 应用程序 名为app py如下 contents of app py from bokeh client import push session from bokeh embed import server do
  • python import inside函数隐藏现有变量

    我在我正在处理的多子模块项目中遇到了一个奇怪的 UnboundLocalError 分配之前引用的局部变量 问题 并将其精简为这个片段 使用标准库中的日志记录模块 import logging def foo logging info fo
  • 实现 XGboost 自定义目标函数

    我正在尝试使用 XGboost 实现自定义目标函数 在 R 中 但我也使用 python 所以有关 python 的任何反馈也很好 我创建了一个返回梯度和粗麻布的函数 它工作正常 但是当我尝试运行 xgb train 时它不起作用 然后 我
  • 在 JavaScript 函数的 Django 模板中转义字符串参数

    我有一个 JavaScript 函数 它返回一组对象 return Func id name 例如 我在传递包含引号的字符串时遇到问题 Dr Seuss ABC BOOk 是无效语法 I tried name safe 但无济于事 有什么解
  • 使用 z = f(x, y) 形式的 B 样条方法来拟合 z = f(x)

    作为一个潜在的解决方案这个问题 https stackoverflow com questions 76476327 how to avoid creating many binary switching variables in gekk

随机推荐