Excluding pydantic model fields only when returned as part of a FastAPI call.

huangapple go评论52阅读模式
英文:

Excluding pydantic model fields only when returned as part of a FastAPI call

问题

Context

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

What I tried

我首先尝试使用pydantic的Field函数来指定不希望返回的字段上的exclude标志。这确实有效,但是在内部逻辑中调用.dict()时,必须通过调用.dict(exclude=None)来覆盖这个设置。

相反,我在Field上指定了一个自定义标志return_in_api,目标是只在FastAPI调用.dict()时应用排除。我尝试编写一个中间件,在调用.dict()并根据哪些嵌套字段包含return_in_api=False的属性来传递自己的exclude属性时调用它。然而,FastAPI的中间件为响应提供了一个流,我不想提前解析它。

相反,我编写了一个装饰器,在路由处理程序的返回值上调用.dict(),并使用适当的exclude值。

Problem

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

理想情况下,我希望将此装饰器应用于每个路由,但通过中间件来实现似乎会破坏响应流。

英文:

Context

I have a very complex pydantic model with a lot of nested pydantic models. I would like to ensure certain fields are never returned as part of API calls, but I would like those fields present for internal logic.

What I tried

I first tried using pydantic's Field function to specify the exclude flag on the fields I didn't want returned. This worked, however functions in my internal logic had to override this whenever they called .dict() by calling .dict(exclude=None).

Instead, I specified a custom flag return_in_api on the Field, with the goal being to only apply exclusions when FastAPI called .dict(). I tried writing a middleware to call .dict() and pass through my own exclude property based on which nested fields contained return_in_api=False. However FastAPI's middleware was giving me a stream for the response which I didn't want to prematurely resolve.

Instead, I wrote a decorator that called .dict() on the return values of my route handlers with the appropriate exclude value.

Problem

One challenge is that whenever new endpoints get added, the person who added them has to remember to include this decorator, otherwise fields leak.

Ideally I would like to apply this decorator to every route, but doing it through middleware seems to break response streaming.

答案1

得分: 5

以下是翻译好的内容:

对所有路由系统地排除字段

我发现最好的方法是使用一个具体但超级简单的示例。假设您有以下模型:

from pydantic import BaseModel, Field


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

我们希望确保在响应中永远不返回bar,无论response_model是作为路由装饰器的参数明确提供还是作为路由处理程序函数的返回注释设置。(假设出于某种原因我们不想使用内置的exclude参数来排除字段。)

我发现的最可靠的方法是子类化fastapi.routing.APIRoute并钩入其__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:
        # 为了确保获取响应模型,我们需要这部分,即使它只是在处理程序函数的注释上设置。
        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
        # 找到要排除的字段:
        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)

现在,我们可以在我们的路由器上设置自定义路由类(文档)。这样,它将用于_所有_其路由:

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"}

使用uvicorn尝试此简单的API示例,并访问端点/demo1/demo2,将得到响应{"foo":"a"}{"foo":"x"}


确保模式一致性

然而,值得一提的是(除非我们采取额外措施),bar字段仍然将作为模式的一部分。这意味着例如自动生成的用于这两个端点的OpenAPI文档将显示bar作为要期望的响应的顶级属性。

这不是您的问题的一部分,所以我假设您已经意识到了这一点并采取了措施来确保一致性。如果没有,对于其他人来说,您可以在基础模型的Config上定义一个静态schema_extra方法,以在返回模式之前删除那些永远不会显示“向外部”的字段:

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)
英文:

Excluding fields systematically for all routes

I find it best to work with a concrete albeit super simple example. Let's assume you have the following model:

from pydantic import BaseModel, Field


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

We want to ensure that bar is never returned in a response, both when the response_model is explicitly provided as an argument to the route decorator and when it is just set as the return annotation for the route handler function. (Assume we do not want to use the built-in exclude parameter for our fields for whatever reason.)

The most reliable way that I found is to subclass fastapi.routing.APIRoute and hook into its __init__ method. By copying a tiny bit of the parent class' code, we can ensure that we will always get the correct response model. Once we have that, it is just a matter of setting up the route's response_model_exclude argument before calling the parent constructor.

Here is what I would suggest:

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)

We can now set that custom route class on our router (documentation). That way it will be used for all its routes:

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"}

Trying this simple API example out with uvicorn and GETting the endpoints /demo1 and /demo2 yields the responses {"foo":"a"} and {"foo":"x"} respectively.


Ensuring schema consistency

It is however worth mentioning that (unless we take additional steps) the bar field will still be part of the schema. That means for example the auto-generated OpenAPI documentation for both those endpoints will show bar as a top-level property of the response to expect.

This was not part of your question so I assume you are aware of this and are taking measures to ensure consistency. If not, and for others reading this, you can define a static schema_extra method on the Config of your base model to delete those fields that will never be shown "to the outside" before the schema is returned:

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)

huangapple
  • 本文由 发表于 2023年4月17日 04:15:06
  • 转载请务必保留本文链接:https://go.coder-hub.com/76030107.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定