处理特定的个性化异常/条件（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

我的问题是： -> 如何使用上面的代码实现这种异常或条件处理？

Answer 1

您可以通过解码提交的 JSON 并自行处理列表来完成此操作。当预期和提交的数据类型不匹配时，您可以捕获 Pydantic 的 ValidationError 加注。

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 端点会生成（在本例中）处理值列表或错误：

['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)'
]

处理特定的个性化异常/条件（FastAPI、Pydantic 模型、预测模型部署）

Handling specific personalised exceptions / conditions ( FastAPI , Pydantic Models , Prediction Model deployment )

python

validation

exception

pydantic

fastapi