Updating objects from pydantic model

[waldner]

If I have a piccolo object, I want to update only some fields that are in an incoming pydantic model. So if this is my table:

class Pieces(Table, tablename="pieces"):

    id = Serial(primary_key=True, null=False)
    type = Text(null=False, required=True)
    description = Text(null=False, required=True)
    start_date = Date(null=False, required=True)
    unique_id = Text(null=False, required=True)

I have two pydantic models to create and update a piece respectively:

class PieceCreate(BaseModel):
    type: str                   # cannot be modified afterwards
    description: str
    start_date: datetime.date
    unique_id: str           # cannot be modified afterwards

    model_config = ConfigDict(extra='forbid')

class PieceUpdate(BaseModel):
    description: str
    start_date: datetime.date

    model_config = ConfigDict(extra='forbid')

Creating a new row is relatively easy, as in eg

# create a piece
p1 = PieceCreate(
      type="type1",
      description="piece1",
      start_date=datetime.date(2023, 1, 1),
      unique_id="uniqueidforpiece1",
)
p1_dict = p1.model_dump()
p1_obj = PiccoloPieces(**p1_dict)
await p1_obj.save()

but now I'm getting a PieceUpdate to apply, and the only way to update all fields that need to be updated is something like

piece_update = PieceUpdate(
    description = "new description",
    start_date = datetime.date(2025, 1, 1)
)

# here I have to manually map all fields from pydantic object to piccolo object
p1_obj.description = piece_update.description
p1_obj.start_date = piece_update.start_date

await p1_obj.save()

# the alternative isn't much better:

await p1_obj.update_self({
    Pieces.description: piece_update.description,
    Pieces.start_date: piece_update.start_date,
})

So this looks kind of cumbersome if there are many updatable fields, and also prone to errors if at some point fields are added or removed. Am I missing a simpler method here?

When using dictionary, the update is straightforward:

updated_piece = await Pieces().update(**piece_update.model_dump()).where(....)

[pavdwest]

You can just pass a dict into update_self, e.g. this is my generic update item by id function:

async def update_one(cls, id: int, dto: UpdateDTOClassType) -> ReadDTOClassType:
        item = await cls.objects().where(cls.id == id).first().run()
        if item is None:
            raise NotFoundException.from_id(id, cls)
        await item.update_self(dto.dict_without_nones()).run()
        return cls.ReadDTOClass(**item.to_dict())

[waldner]

Thanks, that works. Since the only example in the docs uses a special dictionary (Pieces.description: etc) I didn't suppose that you can also use a "normal" dictionary. Thanks