Александр
Александр

Reputation: 193

Using different Pydantic models depending on the value of fields

I have 2 Pydantic models (var1 and var2). The input of the PostExample method can receive data either for the first model or the second. The use of Union helps in solving this issue, but during validation it throws errors for both the first and the second model.

How to make it so that in case of an error in filling in the fields, validator errors are returned only for a certain model, and not for both at once? (if it helps, the models can be distinguished by the length of the field A).

main.py

@app.post("/PostExample")
def postExample(request: Union[schemas.var1, schemas.var2]):
    
    result = post_registration_request.requsest_response()
    return result
  
  

schemas.py

class var1(BaseModel):
    A: str
    B: int
    C: str
    D: str
  
  
class var2(BaseModel):
    A: str
    E: int
    F: str

Upvotes: 19

Views: 26364

Answers (3)

Kai Lukowiak
Kai Lukowiak

Reputation: 308

Pydantic Updates

Pydantic has depricated parse_obj_as and replaced it with TypeAdapter. I modified @yaakov-bressler great answer. It's also now a lot faster which I presume is due to improvements from Pydantic.

# %%
import json
from typing import Annotated, Literal, Union

from pydantic import BaseModel, Field, TypeAdapter, parse_obj_as


# %%
class Model1(BaseModel):
    key: Literal["Model1", "Model1A"]
    value: int


class Model2(BaseModel):
    key: Literal["Model2", "Model2A"]
    value2: int
    name: str


# %%
ValidatorModel = Annotated[Union[Model1, Model2], Field(discriminator="key")]

# %%
adaptor = TypeAdapter(ValidatorModel)

# %% JSON Examples
model1 = {"key": "Model1", "value": 1}
model2 = {"key": "Model2", "value2": 2, "name": "name"}
model1a = {"key": "Model1A", "value": 23}

# %% Parse JSON New Way
%%timeit
for model in [model1, model2, model1a]:
    x = adaptor.validate_python(model)
# 2.06 µs ± 25.1 ns per loop (mean ± std. dev. of 7 runs, 100,000 loops each)

# %% Deprecated way
%%timeit
for model in [model1, model2, model1a]:
    x = parse_obj_as(ValidatorModel, model)
# 669 µs ± 43.1 µs per loop (mean ± std. dev. of 7 runs, 1,000 loops each)

Upvotes: 5

Chris
Chris

Reputation: 34045

You could use Discriminated Unions (credits to @larsks for mentioning that in the comments). Setting a discriminated union, "validation is faster since it is only attempted against one model", as well as "only one explicit error is raised in case of failure". Working example is given below.

Another approach would be to attempt parsing the models (based on a discriminator you pass as query/path param), as described in this answer (Option 1).

Working Example

app.py

import schemas
from fastapi import FastAPI, Body
from typing import Union

app = FastAPI()

@app.post("/")
def submit(item: Union[schemas.Model1, schemas.Model2] = Body(..., discriminator='model_type')):
    return item

schemas.py

from typing import Literal
from pydantic import BaseModel

class Model1(BaseModel):
    model_type: Literal['m1']
    A: str
    B: int
    C: str
    D: str
  
class Model2(BaseModel):
    model_type: Literal['m2']
    A: str
    E: int
    F: str

Test inputs - outputs

#1 Successful Response   #2 Validation error                   #3 Validation error
                                          
# Request body           # Request body                        # Request body
{                        {                                     {
  "model_type": "m1",      "model_type": "m1",                   "model_type": "m2",
  "A": "string",           "A": "string",                        "A": "string",
  "B": 0,                  "C": "string",                        "C": "string",
  "C": "string",           "D": "string"                         "D": "string"
  "D": "string"          }                                     }
}                                                              
                        
# Server response        # Server response                     # Server response
200                      {                                     {
                           "detail": [                           "detail": [
                             {                                     {
                               "loc": [                              "loc": [
                                 "body",                               "body",
                                 "Model1",                             "Model2",
                                 "B"                                   "E"
                               ],                                    ],
                               "msg": "field required",              "msg": "field required",
                               "type": "value_error.missing"         "type": "value_error.missing"
                             }                                     },
                           ]                                       {
                         }                                           "loc": [
                                                                       "body",
                                                                       "Model2",
                                                                       "F"
                                                                     ],
                                                                     "msg": "field required",
                                                                     "type": "value_error.missing"
                                                                   }
                                                                 ]
                                                               }

Upvotes: 20

Yaakov Bressler
Yaakov Bressler

Reputation: 12008

For those looking for a pure pydantic solution (without FastAPI):

You would need to:

This approach is demonstrated below:

Credit to @Chris for his previous answer, of which this solution is based on.

from typing import Literal, Union, Annotated
from pydantic import BaseModel, Field, parse_obj_as


class Model1(BaseModel):
    model_type: Literal['m1']
    A: str
    B: int
    C: str
    D: str


class Model2(BaseModel):
    model_type: Literal['m2']
    A: str
    E: int
    F: str


# Create a new model to represent the discriminated union
ValidModel = Annotated[Union[Model1, Model2], Field(discriminator='model_type')]

# Sample data
raw_data = {
    "model_type": "m1",
    "A": "foo",
    "B": 1,
    "C": "bar",
    "D": "zap"
}

# Parse as the correct model based on `model_type`
my_model = parse_obj_as(ValidModel, raw_data)
print(type(my_model))  # <class '__main__.Model1'>

Upvotes: 6

Related Questions