Implementing Custom Serializers

Custom serializers in pydantic-core allow you to define exactly how your data should be transformed during serialization. Whether you need to convert a complex object to a string for JSON or modify the output of a standard serializer, you can achieve this using Plain and Wrap serializers.

In this tutorial, you will build a custom serializer for a Coordinate class that outputs a list in Python but a formatted string in JSON.

Prerequisites

To follow this tutorial, you need pydantic-core installed in your environment.

pip install pydantic-core

Step 1: Implementing a Plain Serializer

A Plain Serializer completely replaces the default serialization logic for a type. You define it using plain_serializer_function_ser_schema.

First, let's define a simple Coordinate class and a serialization function:

from typing import Any
from pydantic_core import SchemaSerializer, core_schema

class Coordinate:
    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

def serialize_coordinate_plain(value: Coordinate) -> str:
    return f"{value.x},{value.y}"

# Define the schema
schema = core_schema.any_schema(
    serialization=core_schema.plain_serializer_function_ser_schema(
        serialize_coordinate_plain
    )
)

# Initialize the serializer
serializer = SchemaSerializer(schema)

# Test serialization
coord = Coordinate(1.5, 2.5)
print(serializer.to_python(coord))
# Output: '1.5,2.5'

In this step, serialize_coordinate_plain takes the raw object and returns a string. The plain_serializer_function_ser_schema tells pydantic-core to ignore any default logic and use this function instead.

Step 2: Adding Context-Aware Logic

Often, you want different output formats depending on whether you are serializing to a Python dictionary or a JSON string. You can achieve this by enabling the info_arg and inspecting the SerializationInfo object.

def serialize_coordinate_with_info(value: Coordinate, info: core_schema.SerializationInfo) -> Any:
    if info.mode == 'json':
        return {"x": value.x, "y": value.y}
    return (value.x, value.y)

schema = core_schema.any_schema(
    serialization=core_schema.plain_serializer_function_ser_schema(
        serialize_coordinate_with_info,
        info_arg=True
    )
)

serializer = SchemaSerializer(schema)
coord = Coordinate(10.0, 20.0)

print(f"Python: {serializer.to_python(coord)}")
# Output: Python: (10.0, 20.0)

print(f"JSON: {serializer.to_json(coord)}")
# Output: JSON: b'{"x":10.0,"y":20.0}'

By setting info_arg=True, your function receives a SerializationInfo object. The mode property indicates if the target is 'python' or 'json'.

Step 3: Using Wrap Serializers to Extend Logic

A Wrap Serializer allows you to call the default serialization logic and then modify its result. This is useful when you want to "wrap" existing behavior rather than replace it entirely.

You use wrap_serializer_function_ser_schema and a SerializerFunctionWrapHandler.

def wrap_coordinate_serialization(
    value: Any, 
    handler: core_schema.SerializerFunctionWrapHandler, 
    info: core_schema.SerializationInfo
) -> Any:
    # Call the default serializer (which might be a dict serializer)
    result = handler(value)
    
    if info.mode == 'json' and isinstance(result, dict):
        # Add a metadata field only for JSON
        result['__type__'] = 'coordinate'
    
    return result

# We define a dict schema for the coordinate data
inner_schema = core_schema.dict_schema(
    keys_schema=core_schema.str_schema(),
    values_schema=core_schema.float_schema()
)

schema = core_schema.any_schema(
    serialization=core_schema.wrap_serializer_function_ser_schema(
        wrap_coordinate_serialization,
        schema=inner_schema,
        info_arg=True
    )
)

serializer = SchemaSerializer(schema)
data = {"x": 1.0, "y": 2.0}

print(serializer.to_json(data))
# Output: b'{"x":1.0,"y":2.0,"__type__":"coordinate"}'

The handler is a callable that executes the serialization logic defined in the schema argument (or the default logic if schema is omitted). This allows you to transform the input before calling the handler or modify the output after.

Step 4: Ensuring Type Safety with Return Schemas

To ensure your custom serializer returns the correct type, you can provide a return_schema. pydantic-core will validate the output of your function against this schema.

def bad_serializer(value: Any) -> int:
    return "not an integer"  # This will trigger a warning

schema = core_schema.any_schema(
    serialization=core_schema.plain_serializer_function_ser_schema(
        bad_serializer,
        return_schema=core_schema.int_schema()
    )
)

serializer = SchemaSerializer(schema)
# This will emit a UserWarning because the return value doesn't match int_schema
print(serializer.to_python("some input"))

Complete Working Result

Combining these concepts, here is a robust implementation for a custom type:

from typing import Any
from pydantic_core import SchemaSerializer, core_schema

class Point:
    def __init__(self, x: int, y: int):
        self.x = x
        self.y = y

def point_serializer(value: Point, info: core_schema.SerializationInfo) -> Any:
    if info.mode == 'json':
        return f"{value.x}:{value.y}"
    return [value.x, value.y]

# Define a schema that validates a Point instance and serializes it customly
schema = core_schema.is_instance_schema(
    Point,
    serialization=core_schema.plain_serializer_function_ser_schema(
        point_serializer,
        info_arg=True,
        return_schema=core_schema.union_schema([
            core_schema.str_schema(),
            core_schema.list_schema(core_schema.int_schema())
        ])
    )
)

serializer = SchemaSerializer(schema)
p = Point(5, 10)

assert serializer.to_python(p) == [5, 10]
assert serializer.to_json(p) == b'"5:10"'

print("Serialization successful!")

This implementation ensures that Point objects are handled correctly across different formats while maintaining type safety for the serialized output.