FastAPI 模型的条件调用

Question

我有一个连接到 MongoDB 的多语言 FastAPI。我在 MongoDB 中的文档以两种可用语言复制，并以这种方式构建（简化示例）：

\n

\n{\n  "_id": xxxxxxx,\n  "en": { \n          "title": "Drinking Water Composition",\n          "description": "Drinking water composition expressed in... with pesticides.",\n          "category": "Water", \n          "tags": ["water","pesticides"] \n         },\n  "fr": { \n          "title": "Composition de l\'eau de boisson",\n          "description": "Composition de l\'eau de boisson exprim\xc3\xa9e en... pr\xc3\xa9sence de pesticides....",\n          "category": "Eau", \n          "tags": ["eau","pesticides"] \n         },  \n}\n\n

\n

因此，我实现了两个模型DatasetFR和，每个模型都对每种语言中的特定外部模型 ( )DatasetEN进行引用。Enumcategorytags

\n

class DatasetFR(BaseModel):\n    title:str\n    description: str\n    category: CategoryFR\n    tags: Optional[List[TagsFR]]\n\n# same for DatasetEN chnaging the lang tag to EN \n

\n

在路由定义中，我强制语言参数声明相应的模型并获得相应的验证。

\n

\n@router.post("?lang=fr", response_description="Add a dataset")\nasync def create_dataset(request:Request, dataset: DatasetFR = Body(...), lang:str="fr"):\n    ...\n    return JSONResponse(status_code=status.HTTP_201_CREATED, content=created_dataset)\n\n@router.post("?lang=en", response_description="Add a dataset")\nasync def create_dataset(request:Request, dataset: DatasetEN = Body(...), lang:str="en"):\n    ...\n    return JSONResponse(status_code=status.HTTP_201_CREATED, content=created_dataset)\n\n

\n

但这似乎与DRY原则相矛盾。所以，我想知道是否有人知道一个优雅的解决方案： - 给定参数 lang，动态调用相应的 model。

\n

或者，如果我们可以创建一个采用 lang 参数的父模型数据集并检索子模型数据集。

\n

\n

这将非常容易地构建我的 API 路由和模型的调用，并在数学上除以二......

\n

\n

Answer 1

选项1

解决方案如下：定义lang为Query参数并添加参数应匹配的正则表达式。在您的情况下，这将是^(fr|en)$，这意味着只有fr或en才是有效的输入。因此，如果没有找到匹配项，请求将在那里停止，客户端将收到“字符串与正则表达式不匹配...”错误。

接下来，将body参数定义为泛型类型并将dict其声明为Body字段；因此，指示 FastAPI 期待一个JSON主体。

接下来，创建一个字典models，您可以使用该字典来查找使用该lang属性的模型。一旦找到对应的model，try就可以使用 using 来解析JSON正文models[lang].parse_obj(body)（相当于 using models[lang](**body)）。如果没有ValidationError引发，您就知道生成的model实例是有效的。否则，返回一个HTTP_422_UNPROCESSABLE_ENTITY错误，包括错误，您可以根据需要进行处理。

如果您也希望FR成为EN有效值lang，请调整正则表达式以忽略大小写^(?i)(fr|en)$，并确保lang在查找模型时转换为小写（即models[lang.lower()].parse_obj(body)）。

import  pydantic 
from fastapi import FastAPI, Response, status, Body, Query
from fastapi.responses import JSONResponse
from fastapi.encoders import jsonable_encoder

models = {"fr": DatasetFR, "en": DatasetEN}

@router.post("/", response_description="Add a dataset")
async def create_dataset(body: dict = Body(...), lang: str = Query(..., regex="^(fr|en)$")):
    try:
        model = models[lang].parse_obj(body)
    except pydantic.ValidationError as e:
        return Response(content=e.json(), status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, media_type="application/json")

    return JSONResponse(content=jsonable_encoder(dict(model)), status_code=status.HTTP_201_CREATED)

更新

由于这两个模型具有相同的属性（即title和），因此您可以使用这两个属性description定义父模型（例如），并让和模型继承这些属性。DatasetDatasetFRDatasetEN

class Dataset(BaseModel):
    title:str
    description: str
    
class DatasetFR(Dataset):
    category: CategoryFR
    tags: Optional[List[TagsFR]]
    
class DatasetEN(Dataset):
    category: CategoryEN
    tags: Optional[List[TagsEN]]

此外，将逻辑从路由内部移动到依赖函数并让它返回model（如果它通过了验证）可能是更好的方法；否则，请提出HTTPException，正如@tiangolo所演示的那样。您可以使用FastAPI 内部使用的jsonable_encoder来对验证进行编码（返回 时也可以使用相同的函数）。errors()JSONResponse

from fastapi.exceptions import HTTPException
from fastapi import Depends

models = {"fr": DatasetFR, "en": DatasetEN}

async def checker(body: dict = Body(...), lang: str = Query(..., regex="^(fr|en)$")):
    try:
        model = models[lang].parse_obj(body)
    except pydantic.ValidationError as e:
        raise HTTPException(detail=jsonable_encoder(e.errors()), status_code=status.HTTP_422_UNPROCESSABLE_ENTITY)

    return model
        
@router.post("/", response_description="Add a dataset")
async def create_dataset(model: Dataset = Depends(checker)):    
    return JSONResponse(content=jsonable_encoder(dict(model)), status_code=status.HTTP_201_CREATED)

选项2

另一种方法是使用单个 Pydantic 模型（假设是Dataset）并自定义和字段的验证器。您还可以定义为 的一部分，因此不需要将其作为查询参数。您可以使用 a来保存每个类的值，如此处所述，以便您可以有效地检查 ; 中是否存在某个值。并且有字典可以快速查找使用属性。在 的情况下，要验证列表中的每个元素是否有效，请使用，如此处所述。如果属性无效，您可以如文档中所示“将捕获并用于填充” （请参阅​​此处的“注意”部分）。同样，如果您需要以大写形式编写的代码作为有效输入，请调整模式，如前所述。categorytagslangDatasetsetEnumEnumsetlangtagsset.issubsetraise ValueErrorValidationErrorlangregex

PS 你甚至不需要使用Enum这种方法。相反，请set使用允许的值填充下面的每个值。例如， categories_FR = {"Eau"} categories_EN = {"Water"} tags_FR = {"eau", "pesticides"} tags_EN = {"water", "pesticides"}。此外，如果您不想使用正则表达式，而是希望lang属性也有自定义验证错误，则可以将其添加到同一个validator装饰器中，并执行与其他两个字段类似（和之前）的验证。

from pydantic import validator

categories_FR = set(item.value for item in CategoryFR) 
categories_EN = set(item.value for item in CategoryEN) 
tags_FR = set(item.value for item in TagsFR) 
tags_EN = set(item.value for item in TagsEN) 
cats = {"fr": categories_FR, "en": categories_EN}
tags = {"fr": tags_FR, "en": tags_EN}

def raise_error(values):
    raise ValueError(f'value is not a valid enumeration member; permitted: {values}')

class Dataset(BaseModel):
    lang: str = Body(..., regex="^(fr|en)$")
    title: str
    description: str
    category: str
    tags: List[str]

    @validator("category", "tags")
    def validate_atts(cls, v, values, field):
        lang = values.get('lang')
        if lang:
            if field.name == "category":
                if v not in cats[lang]: raise_error(cats[lang])
            elif field.name == "tags":
                if not set(v).issubset(tags[lang]): raise_error(tags[lang])
        return v

        
@router.post("/", response_description="Add a dataset")
async def create_dataset(model: Dataset): 
    return JSONResponse(content=jsonable_encoder(dict(model)), status_code=status.HTTP_201_CREATED)

更新

请注意，在 Pydantic V2 中，@validator已被弃用并被替换为@field_validator. 请查看此答案以获取更多详细信息和示例。

选项3

另一种方法是使用受歧视的联合，如本答案中所述。

根据文档：

当Union与多个子模型一起使用时，您有时确切地知道需要检查和验证哪个子模型并希望强制执行此操作。为此，您可以 my_discriminator在每个子模型中设置相同的字段（我们称之为它）并具有可区分的值，该值是一个（或多个）Literal值。对于您的Union，您可以将鉴别器设置为其值： Field(discriminator='my_discriminator')。

建立歧视性工会有很多好处：

• 验证速度更快，因为仅针对一种模型进行尝试
• 如果失败，只会引发一个显式错误
• 生成的 JSON 模式实现关联的 OpenAPI 规范