Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

Make every fields as optional with Pydantic

I'm making an API with FastAPI and Pydantic.

I would like to have some PATCH endpoints, where 1 or N fields of a record could be edited at once. Moreover, I would like the client to only pass the necessary fields in the payload.

Example:

class Item(BaseModel):
    name: str
    description: str
    price: float
    tax: float


@app.post("/items", response_model=Item)
async def post_item(item: Item):
    ...

@app.patch("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    ...

In this example, for the POST request, I want every field to be required. However, in the PATCH endpoint, I don't mind if the payload only contains, for example, the description field. That's why I wish to have all fields as optional.

Naive approach:

class UpdateItem(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None
    tax: Optional[float]

But that would be terrible in terms of code repetition.

Any better option?

like image 208
nolw38 Avatar asked May 26 '21 06:05

nolw38


People also ask

What is __ root __ In Pydantic?

Pydantic models can be defined with a custom root type by declaring the __root__ field. The root type can be any type supported by pydantic, and is specified by the type hint on the __root__ field.

What is Pydantic Constr?

constr is a specific type that give validation rules regarding this specific type. You have equivalent for all classic python types.

What is Pydantic field?

Pydantic is a useful library for data parsing and validation. It coerces input types to the declared type (using type hints), accumulates all the errors using ValidationError & it's also well documented making it easily discoverable.

What is Pydantic in python?

pydantic allows custom data types to be defined or you can extend validation with methods on a model decorated with the validator decorator. dataclasses integration. As well as BaseModel , pydantic provides a dataclass decorator which creates (almost) vanilla python dataclasses with input data parsing and validation.


Video Answer


4 Answers

Solution with metaclasses

I've just come up with the following:


class AllOptional(pydantic.main.ModelMetaclass):
    def __new__(self, name, bases, namespaces, **kwargs):
        annotations = namespaces.get('__annotations__', {})
        for base in bases:
            annotations.update(base.__annotations__)
        for field in annotations:
            if not field.startswith('__'):
                annotations[field] = Optional[annotations[field]]
        namespaces['__annotations__'] = annotations
        return super().__new__(self, name, bases, namespaces, **kwargs)

Use it as:

class UpdatedItem(Item, metaclass=AllOptional):
    pass

So basically it replace all non optional fields with Optional

Any edits are welcome!

With your example:

from typing import Optional

from fastapi import FastAPI
from pydantic import BaseModel
import pydantic

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str
    price: float
    tax: float


class AllOptional(pydantic.main.ModelMetaclass):
    def __new__(self, name, bases, namespaces, **kwargs):
        annotations = namespaces.get('__annotations__', {})
        for base in bases:
            annotations.update(base.__annotations__)
        for field in annotations:
            if not field.startswith('__'):
                annotations[field] = Optional[annotations[field]]
        namespaces['__annotations__'] = annotations
        return super().__new__(self, name, bases, namespaces, **kwargs)

class UpdatedItem(Item, metaclass=AllOptional):
    pass

# This continues to work correctly
@app.get("/items/{item_id}", response_model=Item)
async def get_item(item_id: int):
    return {
        'name': 'Uzbek Palov',
        'description': 'Palov is my traditional meal',
        'price': 15.0,
        'tax': 0.5,
    }

@app.patch("/items/{item_id}") # does using response_model=UpdatedItem makes mypy sad? idk, i did not check
async def update_item(item_id: str, item: UpdatedItem):
    return item
like image 189
Drdilyor Avatar answered Oct 20 '22 13:10

Drdilyor


The problem is once FastAPI sees item: Item in your route definition, it will try to initialize an Item type from the request body, and you can't declare your model's fields to be optional sometimes depending on some conditional, such as depending on which route it is used.

I have 3 solutions:

Solution #1: Separate Models

I would say that having separate models for the POST and PATCH payloads seems to be the more logical and readable approach. It might lead to duplicated code, yes, but I think clearly defining which route has an all-required or an all-optional model balances out the maintainability cost.

The FastAPI docs has a section for partially updating models with PUT or PATCH that uses Optional fields, and there's a note at the end that says something similar:

Notice that the input model is still validated.

So, if you want to receive partial updates that can omit all the attributes, you need to have a model with all the attributes marked as optional (with default values or None).

So...

class NewItem(BaseModel):
    name: str
    description: str
    price: float
    tax: float

class UpdateItem(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None
    tax: Optional[float] = None

@app.post('/items', response_model=NewItem)
async def post_item(item: NewItem):
    return item

@app.patch('/items/{item_id}',
           response_model=UpdateItem,
           response_model_exclude_none=True)
async def update_item(item_id: str, item: UpdateItem):
    return item

Solution #2: Declare as All-Required, but Manually Validate for PATCH

You can define your model to have all-required fields, then define your payload as a regular Body parameter on the PATCH route, and then initialize the actual Item object "manually" depending on what's available in the payload.

from fastapi import Body
from typing import Dict

class Item(BaseModel):
    name: str
    description: str
    price: float
    tax: float

@app.post('/items', response_model=Item)
async def post_item(item: Item):
    return item

@app.patch('/items/{item_id}', response_model=Item)
async def update_item(item_id: str, payload: Dict = Body(...)):
    item = Item(
        name=payload.get('name', ''),
        description=payload.get('description', ''),
        price=payload.get('price', 0.0),
        tax=payload.get('tax', 0.0),
    )
    return item

Here, the Item object is initialized with whatever is in the payload, or some default if there isn't one. You'll have to manually validate if none of the expected fields are passed, ex.:

from fastapi import HTTPException

@app.patch('/items/{item_id}', response_model=Item)
async def update_item(item_id: str, payload: Dict = Body(...)):
    # Get intersection of keys/fields
    # Must have at least 1 common
    if not (set(payload.keys()) & set(Item.__fields__)):
        raise HTTPException(status_code=400, detail='No common fields')
    ...
$ cat test2.json
{
    "asda": "1923"
}
$ curl -i -H'Content-Type: application/json' --data @test2.json --request PATCH localhost:8000/items/1
HTTP/1.1 400 Bad Request
content-type: application/json

{"detail":"No common fields"}

The behavior for the POST route is as expected: all the fields must be passed.

Solution #3: Declare as All-Optional But Manually Validate for POST

Pydantic's BaseModel's dict method has exclude_defaults and exclude_none options for:

  • exclude_defaults: whether fields which are equal to their default values (whether set or otherwise) should be excluded from the returned dictionary; default False

  • exclude_none: whether fields which are equal to None should be excluded from the returned dictionary; default False

This means, for both POST and PATCH routes, you can use the same Item model, but now with all Optional[T] = None fields. The same item: Item parameter can also be used.

class Item(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None
    tax: Optional[float] = None

On the POST route, if not all the fields were set, then exclude_defaults and exclude_none will return an incomplete dict, so you can raise an error. Else, you can use the item as your new Item.

@app.post('/items', response_model=Item)
async def post_item(item: Item):
    new_item_values = item.dict(exclude_defaults=True, exclude_none=True)

    # Check if exactly same set of keys/fields
    if set(new_item_values.keys()) != set(Item.__fields__):
        raise HTTPException(status_code=400, detail='Missing some fields..')

    # Use `item` or `new_item_values`
    return item
$ cat test_empty.json
{
}
$ curl -i -H'Content-Type: application/json' --data @test_empty.json --request POST localhost:8000/items
HTTP/1.1 400 Bad Request
content-type: application/json

{"detail":"Missing some fields.."}

$ cat test_incomplete.json 
{
    "name": "test-name",
    "tax": 0.44
}
$ curl -i -H'Content-Type: application/json' --data @test_incomplete.json --request POST localhost:8000/items
HTTP/1.1 400 Bad Request
content-type: application/json

{"detail":"Missing some fields.."}

$ cat test_ok.json
{
    "name": "test-name",
    "description": "test-description",
    "price": 123.456,
    "tax": 0.44
}
$ curl -i -H'Content-Type: application/json' --data @test_ok.json --request POST localhost:8000/items
HTTP/1.1 200 OK
content-type: application/json

{"name":"test-name","description":"test-description","price":123.456,"tax":0.44}

On the PATCH route, if at least 1 value is not default/None, then that will be your update data. Use the same validation from Solution 2 to fail if none of the expected fields were passed in.

@app.patch('/items/{item_id}', response_model=Item)
async def update_item(item_id: str, item: Item):
    update_item_values = item.dict(exclude_defaults=True, exclude_none=True)

    # Get intersection of keys/fields
    # Must have at least 1 common
    if not (set(update_item_values.keys()) & set(Item.__fields__)):
        raise HTTPException(status_code=400, detail='No common fields')

    update_item = Item(**update_item_values)

    return update_item
$ cat test2.json
{
    "asda": "1923"
}
$ curl -i -s -H'Content-Type: application/json' --data @test2.json --request PATCH localhost:8000/items/1
HTTP/1.1 400 Bad Request
content-type: application/json

{"detail":"No common fields"}

$ cat test2.json
{
    "description": "test-description"
}
$ curl -i -s -H'Content-Type: application/json' --data @test2.json --request PATCH localhost:8000/items/1
HTTP/1.1 200 OK
content-type: application/json

{"name":null,"description":"test-description","price":null,"tax":null}
like image 39
Gino Mempin Avatar answered Oct 20 '22 14:10

Gino Mempin


Modified @Drdilyor solution. Added checking for nesting of models.

from pydantic.main import ModelMetaclass, BaseModel
from typing import Any, Dict, Optional, Tuple

class _AllOptionalMeta(ModelMetaclass):
    def __new__(self, name: str, bases: Tuple[type], namespaces: Dict[str, Any], **kwargs):
        annotations: dict = namespaces.get('__annotations__', {})

        for base in bases:
            for base_ in base.__mro__:
                if base_ is BaseModel:
                    break

                annotations.update(base_.__annotations__)

        for field in annotations:
            if not field.startswith('__'):
                annotations[field] = Optional[annotations[field]]

        namespaces['__annotations__'] = annotations

        return super().__new__(mcs, name, bases, namespaces, **kwargs)
like image 22
Maxim Avatar answered Oct 20 '22 14:10

Maxim


For my case creating a new class was the only solution that worked, but packed into a function it is quite convenient:

from pydantic import BaseModel, create_model
from typing import Optional

def make_optional(baseclass):
    # Extracts the fields and validators from the baseclass and make fields optional
    fields = baseclass.__fields__
    validators = {'__validators__': baseclass.__validators__}
    optional_fields = {key: (Optional[item.type_], None)
                       for key, item in fields.items()}
    return create_model(f'{baseclass.__name__}Optional', **optional_fields,
                        __validators__=validators)

class Item(BaseModel):
    name: str
    description: str
    price: float
    tax: float

ItemOptional = make_optional(Item)

Comparing after and before:

> Item.__fields__

{'name': ModelField(name='name', type=str, required=True),
 'description': ModelField(name='description', type=str, required=True),
 'price': ModelField(name='price', type=float, required=True),
 'tax': ModelField(name='tax', type=float, required=True)}

> ItemOptional.__fields__

{'name': ModelField(name='name', type=Optional[str], required=False, default=None),
 'description': ModelField(name='description', type=Optional[str], required=False, default=None),
 'price': ModelField(name='price', type=Optional[float], required=False, default=None),
 'tax': ModelField(name='tax', type=Optional[float], required=False, default=None)}

It does work, and also it allows you to filter out some fields in the dict_comprehension if it is required.

Moreover in fastapi this approach allows you to do something like this:

@app.post("/items", response_model=Item)
async def post_item(item: Item = Depends()):
    ...

@app.patch("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: make_optional(Item) = Depends()):
    ...

Which reduces a lot the boilerplate, using the same approach you can also make a function that makes optional the fields and also exclude a field in case your Item has an ID field,the id would will be repeated in your PATCH call. That can be solved like this:

def make_optional_no_id(baseclass):
    ... # same as make optional
    optional_fields = {key: (Optional[item.type_], None) 
                       for key, item in fields.items() if key != 'ID'} # take out here ID
    ... # you can also take out also validators of ID

@app.patch("/items/{item_id}", response_model=Item)
async def update_item(item: make_optional_no_id(Item) = Depends()):
like image 1
Ziur Olpa Avatar answered Oct 20 '22 13:10

Ziur Olpa