Handling specific personalised exceptions / conditions ( FastAPI , Pydantic Models , Prediction Mode-CodePudding

I'm using Pydantic models for data validation with FastAPI to deploy a Machine Learning model for predictions, so I want to handle the following exceptions / conditions :

Giving many inputs if one of them doesn't match the features requirements (types, length...) throw an exception for that specific invalid input but show outputs of the other valid inputs

What I want to achieve

Inputs :


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

Outputs :


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

What I have with my base model classes :

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

My question is : -> How can I achieve this kind of exception or condition handling with my code above ?

CodePudding user response：

You can do this by decoding the submitted JSON and handling the list yourself. You can then catch the ValidationError raise by Pydantic when there's a mismatch between expected and submitted data types.

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

Submitting your list to the /foo endpoint generates (in this case) a list of processed values or an error:

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