Сделайте все поля необязательными с помощью Pydantic

Я делаю API с FastAPI и Pydantic.

Я хотел бы иметь несколько конечных точек PATCH, где можно было бы редактировать 1 или N полей записи одновременно. Более того, я бы хотел, чтобы клиент передавал только необходимые поля в полезной нагрузке.

Пример:

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):
    ...

В этом примере для запроса POST я хочу, чтобы каждое поле было обязательным. Однако в конечной точке PATCH я не возражаю, если полезная нагрузка содержит, например, только поле описания. Вот почему я хочу, чтобы все поля были необязательными.

Наивный подход:

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

Но это было бы ужасно с точки зрения повторения кода.

Есть лучший вариант?


person nolw38    schedule 26.05.2021    source источник


Ответы (2)


Решение с метаклассами

Я только что придумал следующее:


class AllOptional(pydantic.main.ModelMetaclass):
    def __new__(self, name, bases, namespaces, **kwargs):
        annotations = namespaces.get('__annotations__', {})
        for base in bases:
            annotations = {**annotations, **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

Таким образом, в основном он заменяет все необязательные поля на Optional

Любые правки приветствуются!

С вашим примером:

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__', {})
        print(bases)
        print(bases[0].__annotations__)
        for base in bases:
            annotations = {**annotations, **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}") # not using response_model=Item
async def update_item(item_id: str, item: UpdatedItem):
    return item
person Drdilyor    schedule 28.05.2021
comment
Представьте это как принятый ответ, поскольку это единственное решение, которое действительно решает проблему. Честно говоря, это могло быть улучшение Pydantic! - person nolw38; 28.05.2021
comment
я думаю, что это плохо работает с mypy, но я должен это проверить - person Drdilyor; 28.05.2021

Проблема в том, что как только FastAPI увидит item: Item в определении вашего маршрута, он попытается инициализировать Item тип из тела запроса, и вы не можете объявить поля вашей модели необязательными иногда в зависимости от некоторых условий, например, в зависимости от того, какой маршрут используется.

У меня есть 3 решения:

Решение №1: отдельные модели

Я бы сказал, что наличие отдельных моделей для полезных данных POST и PATCH кажется более логичным и читаемым подходом. Да, это может привести к дублированию кода, но я думаю, четкое определение того, какой маршрут имеет обязательную или полностью необязательную модель, уравновешивает затраты на ремонтопригодность.

В документации FastAPI есть раздел для частичного обновления моделей с помощью PUT или PATCH, который использует Optional поля, и в конце есть примечание, в котором говорится что-то похожее:

Обратите внимание, что модель ввода все еще проверена.

Итак, если вы хотите получать частичные обновления, которые могут опускать все атрибуты, вам нужна модель со всеми атрибутами, отмеченными как необязательные (со значениями по умолчанию или 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

Решение № 2: объявить как "все необходимое", но проверить исправность вручную.

Вы можете определить свою модель так, чтобы в ней были все обязательные поля, а затем определить свою полезную нагрузку как обычный _ 6_ в маршруте PATCH, а затем инициализировать фактический объект Item вручную в зависимости от того, что доступно в полезной нагрузке.

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

Здесь объект Item инициализируется тем, что есть в полезной нагрузке, или некоторым значением по умолчанию, если такового нет. Вам придется вручную проверить, не передано ли ни одно из ожидаемых полей, например:

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"}

Поведение для маршрута POST такое, как ожидалось: все поля должны быть переданы.

Решение № 3: объявить как полностью необязательный, но вручную проверить для POST

Метод dict Pydantic BaseModel имеет параметры exclude_defaults и exclude_none для:

  • exclude_defaults: следует ли исключать поля, которые равны их значениям по умолчанию (установленным или нет) из возвращаемого словаря; по умолчанию False

  • exclude_none: должны ли поля, которые равны None, исключаться из возвращаемого словаря; по умолчанию False

Это означает, что для маршрутов POST и PATCH вы можете использовать одну и ту же модель Item, но теперь со всеми полями Optional[T] = None. Можно также использовать тот же параметр item: Item.

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

На маршруте POST, если не все поля были установлены, то exclude_defaults и exclude_none вернут неполный dict, поэтому вы можете вызвать ошибку. В противном случае вы можете использовать item как свой новый 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}

На маршруте PATCH, если хотя бы 1 значение не является значением по умолчанию / None, то это будут ваши данные обновления. Используйте ту же проверку, что и в решении 2, если не было передано ни одно из ожидаемых полей.

@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}
person Gino Mempin    schedule 26.05.2021
comment
Спасибо ! Отличные объяснения. Итак, похоже, что решение 2 лучше, чем 3, поскольку ручная проверка PATCH должна выполняться в обоих, а проверка POST только в 3. Но я согласен, что решение 1 легче читать, когда вы не одиноки в проекте. . - person nolw38; 26.05.2021