Python中的Pydantic序列化详解

from datetime import datetime, timedelta, timezone
from pydantic import BaseModel, ConfigDict, field_serializer
from pydantic_core.core_schema import FieldSerializationInfo, SerializerFunctionWrapHandler
class WithCustomEncoders(BaseModel):
    model_config = ConfigDict(ser_json_timedelta='iso8601')
    dt: datetime
    diff: timedelta
    diff2: timedelta
    @field_serializer('dt')
    def serialize_dt(self, dt: datetime, _info: FieldSerializationInfo):
        print(_info)
        return dt.timestamp()
    # 下面的装饰器先执行
    @field_serializer('diff')
    def ssse(self, diff: timedelta, info: FieldSerializationInfo):
        print(info)
        return diff.total_seconds()
    @field_serializer('diff2', mode='wrap')
    @staticmethod
    def diff2_ser(diff2: timedelta, nxt: SerializerFunctionWrapHandler, info: FieldSerializationInfo):
        value = nxt(diff2)
        return value + 'postprocess'
m = WithCustomEncoders(
    dt=datetime(2032, 6, 1, tzinfo=timezone.utc), diff=timedelta(minutes=2),
    diff2=timedelta(minutes=1)
)
print(m.model_dump_json())
# {"dt":1969660800.0,"diff":120.0,"diff2":"PT60Spostprocess"}

from typing import Dict, Any
from pydantic import BaseModel, model_serializer
from pydantic_core.core_schema import SerializerFunctionWrapHandler, SerializationInfo
class Model(BaseModel):
    x: str
    @model_serializer
    def ser_model(self, info: SerializationInfo):
        print(info)
        return {'x': f'xxxxxx {self.x}'}
    @model_serializer(mode='wrap')
    def ser_model_wrap(self, nxt: SerializerFunctionWrapHandler, info: SerializationInfo) -> Dict[str, Any]:
        print(info)
        return {'x': f'serialized {nxt(self)}'}
print(Model(x='test value').model_dump_json())
# {"x":"serialized {'x': 'test value'}"}

from typing import Any
from typing_extensions import Annotated
from pydantic import BaseModel, SerializerFunctionWrapHandler
from pydantic.functional_serializers import WrapSerializer, PlainSerializer
def ser_wrap(v: Any, nxt: SerializerFunctionWrapHandler) -> str:
    return f'{nxt(v + 1):,}'
FancyInt = Annotated[int, WrapSerializer(ser_wrap, when_used='json')]
DoubleInt = Annotated[int, PlainSerializer(lambda x: x * 2)]
class MyModel(BaseModel):
    x: FancyInt
    y: DoubleInt
print(MyModel(x=1234, y=2).model_dump())
# {'x': 1234, 'y': 4}
print(MyModel(x=1234, y=2).model_dump(mode='json'))
# {'x': '1,235', 'y': 4}

def serialize_datetime(value: datetime.datetime, __: SerializerFunctionWrapHandler, _: SerializationInfo):
    return value.strftime('%Y-%m-%d %H:%M:%S')
LocalDateTime = Annotated[datetime.datetime, WrapSerializer(serialize_datetime, when_used='json')]

from pydantic import BaseModel, SerializeAsAny
class User(BaseModel):
    name: str
class UserLogin(User):
    password: str
class OuterModel(BaseModel):
    # 声明为User类型，按照User类序列化，只有name字段
    user: User
    user1: SerializeAsAny[User] = UserLogin(name='serialize as any', password='hunter')
# 实际类型为UserLogin
user = UserLogin(name='pydantic', password='hunter2')
m = OuterModel(user=user)
print(m)
# user=UserLogin(name='pydantic', password='hunter2') user1=UserLogin(name='serialize as any', password='hunter')
print(m.model_dump())
# {'user': {'name': 'pydantic'}, 'user1': {'name': 'serialize as any', 'password': 'hunter'}}

# TODO need to get pickling to work
import pickle
from pydantic import BaseModel
class FooBarModel(BaseModel):
    a: str
    b: int
m = FooBarModel(a='hello', b=123)
print(m)
#> a='hello' b=123
data = pickle.dumps(m)
print(data[:20])
#> b'\x80\x04\x95\x95\x00\x00\x00\x00\x00\x00\x00\x8c\x08__main_'
m2 = pickle.loads(data)
print(m2)
#> a='hello' b=123

from pydantic import BaseModel, SecretStr
class User(BaseModel):
    id: int
    username: str
    password: SecretStr
class Transaction(BaseModel):
    id: str
    user: User
    value: int
t = Transaction(
    id='1234567890',
    user=User(id=42, username='JohnDoe', password='hashedpassword'),
    value=9876543210,
)
# using a set:
print(t.model_dump(exclude={'user', 'value'}))
#> {'id': '1234567890'}
# using a dict:
print(t.model_dump(exclude={'user': {'username', 'password'}, 'value': True}))
#> {'id': '1234567890', 'user': {'id': 42}}
print(t.model_dump(include={'id': True, 'user': {'id'}}))
#> {'id': '1234567890', 'user': {'id': 42}}