处理特定的个性化例外 /条件（FastApi，Pydantic模型，预测模型部署）

Question

我正在使用 Pydantic模型用于使用 fastapi 来部署预测的机器学习模型，所以我想处理以下例外/条件：

• 给出许多输入，如果其中一个不匹配功能要求（类型，长度...）给出了该特定无效输入的例外，但是显示了其他有效输入的输出

我想实现的目标

输入：

    [
      {
        "name":"John",
        "age": 20,
        "salary": 15000
      },
      {
        "name":"Emma",
        "age": 25,
        "salary": 28000
      },
      {
        "name":"David",
        "age": "test",
        "salary": 50000
      },
      {
        "name":"Liza",
        "age": 5000,
        "salary": 30000
      }
    ]
   

输出：

    [
      {
        "prediction":"Class1",
        "probability": 0.88
      },
      {
        "prediction":"Class0",
        "probability": 0.79
      },
      {
      "ËRROR: Expected type int but got str instead"
      },
      {
      "ËRROR: invalid age number"
      }
    ]

我对基本模型类别的内容：

from pydantic import BaseModel, validator
from typing import List

n_inputs = 3
n_outputs = 2


class Inputs(BaseModel):
    name: str
    age: int
    salary: float


class InputsList(BaseModel):
    inputs: List[Inputs]

    @validator("inputs", pre=True)
    def check_dimension(cls, v):
        for point in v:
            if len(point) != n_inputs:
                raise ValueError(f"Input data must have a length of {n_inputs} features")

        return v


class Outputs(BaseModel):
    prediction: str
    probability: float

class OutputsList(BaseModel):
    output: List[Outputs]

    @validator("output", pre=True)
    def check_dimension(cls, v):
        for point in v:
            if len(point) != n_outputs:
                raise ValueError(f"Output data must a length of {n_outputs}")

        return v

我的问题是： - ＆gt;如何通过上面的代码实现这种例外或条件处理？

Answer 1

您可以通过解码提交的JSON并自己处理清单来做到这一点。然后，当预期数据类型和已提交的数据类型之间存在不匹配时，您可以捕获verionationError pydantic升起。

from fastapi import FastAPI, Request
from pydantic import BaseModel, validator, ValidationError, conint
from typing import List

app = FastAPI()


class Inputs(BaseModel):
    name: str
    age: conint(lt=130)
    salary: float
    
   
@app.post("/foo")
async def create_item(request: Request):
    input_list = await request.json()
    outputs = []
    
    for element in input_list:
        try:
            read_input = Inputs(**element)
            outputs.append(f'{read_input.name}: {read_input.age * read_input.salary}')
        except ValidationError as e:
            outputs.append(f'Invalid input: {e}')
    
    return outputs

将您的列表提交到/foo endpoint生成（在这种情况下）已处理的值列表或错误：

['John: 300000.0', 
 'Emma: 700000.0', 
 'Invalid input: 1 validation error for Inputs\nage\n  value is not a valid integer (type=type_error.integer)', 
 'Invalid input: 1 validation error for Inputs\nage\n  ensure this value is less than 130 (type=value_error.number.not_lt; limit_value=130)'
]