FastApi 模型的条件调用

Question

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


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

因此，我实现了两个模型 DatasetFR 和 DatasetEN，每个模型都在每个语言中为 category 和 tags 引用了特定的外部模型（枚举）。

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

# same for DatasetEN chnaging the lang tag to EN

在routes定义中我强制language参数声明对应的Model并得到对应的validation


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

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

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

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

This would incredibly ease building my API routes and the call of my models and mathematically divide by two the writing...

Answer 1

答案分为两部分（API调用和数据结构）

对于 API 调用，您可以将它们分成 2 条路线，如 /api/v1/fr/... 和 /api/v1/en/...（分隔资源表示！）并使用 fastapi.APIRouter 来声明同一条路线两次，但每条路线的验证模式都更改为您要使用的路线。

您可以首先将通用 BaseModel 声明为 ABC 和 ABCEnum。

from abc import ABC
from pydantic import BaseModel

class MyModelABC(ABC, BaseModel):
    attribute1: MyEnumABC

class MyModelFr(MyModelABC):
    attribute1: MyEnumFR

class MyModelEn(MyModelABC):
    attribute1: MyEnumEn

然后您可以 select 通过 class 工厂的路线的准确模型：

my_class_factory: dict[str, MyModelABC] = {
    "fr": MyModelFr,
    "en": MyModelEn, 
}

最后您可以通过路线工厂创建您的路线：

def generate_language_specific_router(language: str, ...) -> APIRouter:
    router = APIRouter(prefix=language)
    MySelectedModel: MyModelABC = my_class_factory[language]

    @router.post("/")
    def post_something(my_model_data: MySelectedModel):
        # My internal logic
    return router

关于第二部分（内部计算和数据存储），国际化通常通过hashmaps来完成。

可以研究标准 python 库 gettext

否则，可以将原始语言明确用作 key/hash，然后将翻译映射到它（如果您想在通话中保持一致，也包括原始语言）。

它看起来像：

dictionnary_of_babel = {
    "word1": {
        "en": "word1",
        "fr": "mot1",
    },
    "word2": {
        "en": "word2",
    },
    "Drinking Water Composition": {
        "en": "Drinking Water Composition",
        "fr": "Composition de l'eau de boisson",
    },
}

my_arbitrary_object = {
    "attribute1": "word1",
    "attribute2": "word2",
    "attribute3": "Drinking Water Composition",
}

my_translated_object = {}
for attribute, english_sentence in my_arbitrary_object.items():
    if "fr" in dictionnary_of_babel[english_sentence].keys():
        my_translated_object[attribute] = dictionnary_of_babel[english_sentence]["fr"]
    else:
        my_translated_object[attribute] = dictionnary_of_babel[english_sentence]["en"]  # ou sans "en"

expected_translated_object = {
    "attribute1": "mot1",
    "attribute2": "word2",
    "attribute3": "Composition de l'eau de boisson",
}

assert expected_translated_object == my_translated_object

此代码应运行原样

关于 mongoDB 表示的提议，如果我们不希望有一个单独的 table 用于翻译，可以是 data structure 例如：

# normal:
my_attribute: "sentence"

# internationalized
my_attribute_internationalized: {
    sentence: {
        original_lang: "sentence"
        lang1: "sentence_lang1",
        lang2: "sentence_lang2",
    }
}

推广字符串翻译的一个简单策略是定义一个匿名函数 _() 嵌入如下翻译：

CURRENT_MODULE_LANG = "fr"

def _(original_string: str) -> str:
    """Switch from original_string to translation"""
    return dictionnary_of_babel[original_string][CURRENT_MODULE_LANG]

然后在需要翻译的地方调用它：

>>> print(_("word 1"))
"mot 1"

您可以在有关 internationalization-in-python-code 的 django 文档中找到对这种做法的参考。

对于静态翻译（例如网站或文档），您可以使用 .po 文件和 poedit 等编辑器（请参阅 french translation of python docs 了解实际用例）！

Answer 2

解决方案如下。将参数应匹配的 lang 定义为 Query paramter and add a regular expression。在您的情况下，这将是 ^(fr|en)$，这意味着只有 fr 或 en 是有效输入。因此，如果未找到匹配项，请求将停在那里，客户端将收到 "string does not match regex..." 错误。

接下来，将body参数定义为generic type of dict and declare it as Body字段；因此，指示 FastAPI 期望一个 JSON 主体。

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

的错误

如果您还希望 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(model.dict()), status_code=status.HTTP_201_CREATED)

更新 1

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

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]]

此外，将逻辑从路由内部移动到 dependecy function and have it return the model, if it passes the validation; otherwise, raise an HTTPException, as also demonstrated by @tiangolo. You can use jsonable_encoder, which is internally used by FastAPI 以编码验证 errors() 可能是更好的方法（当 [=110 时也可以使用相同的函数=]正在 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(model.dict()), status_code=status.HTTP_201_CREATED)

更新 2

另一种方法是使用单个 Pydantic 模型（假设 Dataset）并自定义 validators for category and tags fields. You can also define lang as part of Dataset, thus, no need to have it as query parameter. You can use a set, as described , to keep the values of each Enum class, so that you can efficiently check if a value exists in the Enum; and have dictionaries to quickly look up for a set using the lang attribute. In the case of tags, to verify that every element in the list is valid, use set.issubset, as described here. If an attribute is not valid, you can raise ValueError, as shown in the documentation, "which will be caught and used to populate ValidationError" (see "Note" section here）。同样，如果您需要以大写形式编写的 lang 代码作为有效输入，请调整 regex 模式，如前所述。

P.S。在这种方法中，您甚至不需要使用 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(model.dict()), status_code=status.HTTP_201_CREATED)

更新 3

另一种方法是使用 Discriminated Unions, as described in 。

根据文档：

When Union is used with multiple submodels, you sometimes know exactly which submodel needs to be checked and validated and want to enforce this. To do that you can set the same field - let's call it my_discriminator - in each of the submodels with a discriminated value, which is one (or many) Literal value(s). For your Union, you can set the discriminator in its value: Field(discriminator='my_discriminator').

Setting a discriminated union has many benefits:

validation is faster since it is only attempted against one model

only one explicit error is raised in case of failure

the generated JSON schema implements the associated OpenAPI specification

FastApi 模型的条件调用

Conditional call of a FastApi Model

python

mongodb

pydantic

fastapi

更新 1

更新 2

更新 3