我正在编写一个脚本,将数据从内部系统发布到第三方 Web API。我们的内部系统使用 Pydantic 中定义的数据模型。我正在尝试使用
requests.post
将这些模型之一中的数据发送到 API,但它无法序列化某些类型,这让我发疯。
以此数据模型为例:
class Product(BaseModel):
id: int
group_id: int
sku: str
price: Decimal
special_offer_price: Decimal
last_update: datetime.date
如果我尝试使用此模型使用
requests.post
发送数据,则:
product_item = website_api.Product(**product.__dict__)
requests.post(url, headers=jwt_auth.auth_header, json=product_item)
str
和 int
字段很好,但是 Decimal
和 datetime.date
字段会抛出异常,因为它们不可序列化,给出的错误有点像:
TypeError: Object of type date is not JSON serializable
我尝试将字段的
@field_serializer
添加到 pydantic 的类定义中;
@field_serializer("price", "special_offer_price")
def serialize_decimal(self, value):
return str(value)
@field_serializer("last_update")
def serialize_date(self, value):
return str(value)
但是
requests
在序列化时似乎没有使用它。
这里:
product_item = website_api.Product(**product.__dict__)
如果您的
product_item
是您提到的 Product
模型的实例,那么您应该这样做:
requests.post(url, headers=jwt_auth.auth_header, json=product_item.model_dump())
您不应该需要
datetime.date
和 Decimal
的序列化器。当我在这个 Product
类上测试这个时,我能够使用 product_item.model_dump()
序列化它,没有任何问题。
就像这里:
import datetime
from decimal import Decimal
from pydantic import BaseModel
class Product(BaseModel):
id: int
group_id: int
sku: str
price: Decimal
special_offer_price: Decimal
last_update: datetime.date
product = Product(
id=1,
group_id=1,
sku="123456",
price=Decimal("100.00"),
special_offer_price=Decimal("90.00"),
last_update=datetime.date(2021, 1, 1),
)
print(product.model_dump())
产生:
{'id': 1, 'group_id': 1, 'sku': '123456', 'price': Decimal('100.00'), 'special_offer_price': Decimal('90.00'), 'last_update': datetime.date(2021, 1, 1)}
或者如果你喜欢它作为 json 字符串,那么这个:
print(product.model_dump_json())
产生:
{"id":1,"group_id":1,"sku":"123456","price":"100.00","special_offer_price":"90.00","last_update":"2021-01-01"}
Date 对象,如错误所示,默认情况下不可序列化。为此,您可以创建一个编码器:
import json
from decimal import Decimal
from datetime import date
from pydantic import BaseModel
class CustomJSONEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
return str(obj)
if isinstance(obj, date):
return obj.isoformat()
return super().default(obj)
class Product(BaseModel):
id: int
group_id: int
sku: str
price: Decimal
special_offer_price: Decimal
last_update: date
product_item = Product(
id=1,
group_id=10,
sku="ABC123",
price=Decimal('19.99'),
special_offer_price=Decimal('15.99'),
last_update=date.today()
)
product_dict = product_item.dict()
data = json.dumps(product_dict, cls=CustomJSONEncoder)
headers = {
'Content-Type': 'application/json',
**jwt_auth.auth_header
}
response = requests.post(url, headers=headers, data=data)
从这里开始,您将能够使用这个 JSON 序列化数据做任何您想做的事情。