I want to use Pydantic 2 to export data to a Kafka Sub Queue and I need to export to JSON in a different model.
For example I have following object:
class obj(BaseModel):
resource: str
resourceid: int
resourceTime: datetime
name: str
option1: str
option2: int
{
"resource": "TEST",
"resourceid": "1234",
"resourceTime": "2024-09-03 00:00:00",
"option1": "test",
"option2": 23421,
"name": "test"
}
But I need to dump in this way:
{
"resource": "TEST",
"name": "test,
"resourceid": "1234",
"resourceTime": "2024-09-03 00:00:00",
"attributes": [
{
"name": "option1",
"value": "test",
},
{
"name": "option2",
"value": 23421
},
{
"name": "name",
"value": "test"
}
]
}
There is a possibility to use some functionality of Pydantic to achieve this?
Thanks.
You can use model_serializer
:
from datetime import datetime
from pydantic import BaseModel, Field, model_serializer
class obj(BaseModel):
resource: str
resourceid: int
resourceTime: datetime
name: str
option1: str = Field(exclude=True)
option2: int = Field(exclude=True)
@model_serializer(mode="wrap")
def _serialize(self, handler):
d = handler(self)
d["resourceid"] = str(d["resourceid"])
d["attributes"] = [
{
"name": "option1",
"value": self.option1
},
{
"name": "option2",
"value": self.option2
},
{
"name": "name",
"value": self.name
},
]
return d
o = obj.model_validate(
{
"resource": "TEST",
"resourceid": "1234",
"resourceTime": "2024-09-03 00:00:00",
"option1": "test",
"option2": 23421,
"name": "test"
}
)
print(o.model_dump())
UPD:
To achieve dynamic behavior you can do something like this:
class obj(BaseModel):
resource: str
resourceid: int
resourceTime: datetime
name: str
option1: str = Field(exclude=True)
option2: int = Field(exclude=True)
@model_serializer(mode="wrap")
def _serialize(self, handler):
d = handler(self)
d["resourceid"] = str(d["resourceid"])
attributes = [
{
"name": "name",
"value": self.name
},
]
for field_name in self.model_fields.keys():
if d.get(field_name) is None:
attributes.append(
{"name": field_name, "value": getattr(self, field_name)}
)
d["attributes"] = attributes
return d
This implementation will add to the attributes
all fields marked as exclude=True
plus the name
field
1
Thanks.
from typing import Literal
from datetime import datetime
from pydantic import BaseModel, model_serializer, Field
class A(BaseModel):
resource: str = Field(serialization_alias="RESOURCE")
resourceid: int
resourceTime: datetime
name: str
option1: str = Field(exclude=True)
option2: int = Field(exclude=True)
@model_serializer(mode="wrap")
def test(self, handler):
d = handler(self)
d["attributes"] = [{ "name": k, "value": v } for k,v in self.__dict__.items()]
return d
o = A(
**{
"resource": "TEST",
"resourceid": "1234",
"resourceTime": "2024-09-03 00:00:00",
"option1": "test",
"option2": 23421,
"name": "test"
}
)
print (o.model_dump_json(indent=2, by_alias=True))
And I have:
{
"RESOURCE": "TEST",
"resourceid": 1234,
"resourceTime": "2024-09-03T00:00:00",
"name": "test",
"attributes": [
{
"name": "resource",
"value": "TEST"
},
{
"name": "resourceid",
"value": 1234
},
{
"name": "resourceTime",
"value": "2024-09-03T00:00:00"
},
{
"name": "name",
"value": "test"
},
{
"name": "option1",
"value": "test"
},
{
"name": "option2",
"value": 23421
}
]
}
So in this way I could only fields for validation.
The only drawback that I lose some Pydantic features on values that are in loop attribute (ex. by_alias=True). That is acceptable for my case.
1