Skip to content

v0.7.0 Migration Guide

This release introduces enhanced serialization capabilities with a global custom serializer registry and improved Pydantic v2 integration.

Key Changes

✨ New Features

  • Global Custom Serializer Registry: Register custom serializers for any type that apply throughout your application
  • Enhanced BaseModel: Intelligent fallback serialization for common types (datetime, complex, sets, etc.)
  • Improved Backward Compatibility: Most existing code continues to work without changes
  • Better Performance: Leverages Pydantic v2's native serialization with intelligent fallbacks

🔧 Breaking Changes

  • Custom types now require explicit handling: Types not natively serializable by Pydantic v2 need either:
  • arbitrary_types_allowed = True in model config, or
  • Custom serialization via global registry or @field_serializer

Migration Guide

For Most Users (No Changes Required)

If you're using standard Pydantic types (str, int, float, bool, dict, list, datetime, Enum), your code will continue to work without any changes.

For Users with Custom Types

# Before (v0.6.0 and earlier)
class MyModel(BaseModel):
    custom_field: MyCustomType

    def model_dump(self, **kwargs):
        data = super().model_dump(**kwargs)
        data['custom_field'] = self.custom_field.to_dict()
        return data

# After (v0.7.0)
from flujo.utils import register_custom_serializer

# Register once, use everywhere
register_custom_serializer(MyCustomType, lambda x: x.to_dict())

class MyModel(BaseModel):
    custom_field: MyCustomType  # No changes needed

Option 2: Use Field Serializer

# After (v0.7.0)
from pydantic import field_serializer
from flujo.models import BaseModel

class MyModel(BaseModel):
    custom_field: MyCustomType

    @field_serializer('custom_field', when_used='json')
    def serialize_custom_field(self, value: MyCustomType) -> dict:
        return value.to_dict()

Option 3: Enable Arbitrary Types

# After (v0.7.0)
from flujo.models import BaseModel

class MyModel(BaseModel):
    custom_field: MyCustomType
    model_config = {"arbitrary_types_allowed": True}

For State Backend Users

State backends now automatically handle custom types through the global registry:

# Before: Custom serialization in state backends
class CustomStateBackend(StateBackend):
    def save_state(self, run_id: str, state: Dict[str, Any]) -> None:
        # Manual serialization logic
        serialized = self._custom_serialize(state)
        # ... save logic

# After: Automatic handling
class CustomStateBackend(StateBackend):
    def save_state(self, run_id: str, state: Dict[str, Any]) -> None:
        # Custom types are automatically handled
        serialized = json.dumps(state)
        # ... save logic

For Cache Users

Cache keys now automatically handle custom types:

# Before: Manual cache key serialization
def create_cache_key(data):
    return hashlib.md5(str(data).encode()).hexdigest()

# After: Automatic handling
def create_cache_key(data):
    # Custom types are automatically serialized
    return hashlib.md5(json.dumps(data).encode()).hexdigest()

New API Reference

Global Registry Functions

from flujo.utils import (
    register_custom_serializer,
    lookup_custom_serializer,
    safe_serialize,
    serializable_field,
    create_serializer_for_type
)

# Register a custom serializer
register_custom_serializer(MyType, my_serializer_func)

# Look up a serializer (internal use)
serializer = lookup_custom_serializer(value)

# Safe serialization with fallbacks
result = safe_serialize(obj, default_serializer=str)

Enhanced BaseModel

The BaseModel now includes intelligent fallback serialization for:

  • datetime objects: ISO format strings
  • Enum values: The enum's value
  • Complex numbers: {"real": x, "imag": y} format
  • Sets and frozensets: Converted to lists
  • Bytes and memoryview: UTF-8 decoded strings
  • Callable objects: Module-qualified names
  • Custom objects: Dictionary representation or string representation

Examples

Database Connection Serialization

from flujo.utils import register_custom_serializer
from flujo.models import BaseModel

class DatabaseConnection:
    def __init__(self, host: str, port: int):
        self.host = host
        self.port = port

def serialize_db_connection(conn: DatabaseConnection) -> dict:
    return {
        "type": "database_connection",
        "host": conn.host,
        "port": conn.port
    }

register_custom_serializer(DatabaseConnection, serialize_db_connection)

class PipelineContext(BaseModel):
    db_connection: DatabaseConnection
    model_config = {"arbitrary_types_allowed": True}

# Now this works automatically
context = PipelineContext(db_connection=DatabaseConnection("localhost", 5432))
serialized = context.model_dump(mode="json")

Custom Enum Serialization

from enum import Enum
from flujo.utils import register_custom_serializer

class Priority(Enum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3

def serialize_priority(p: Priority) -> str:
    return p.name.lower()

register_custom_serializer(Priority, serialize_priority)

class Task(BaseModel):
    priority: Priority
    model_config = {"arbitrary_types_allowed": True}

task = Task(priority=Priority.HIGH)
serialized = task.model_dump(mode="json")
# Result: {"priority": "high"}

Performance Impact

  • Minimal overhead: Global registry uses simple dictionary lookups
  • Faster than custom serialization: Leverages Pydantic v2's optimized serialization
  • Backward compatible: Existing code performance is maintained or improved

Troubleshooting

Common Issues

"Unable to generate pydantic-core schema"

Add arbitrary_types_allowed = True to your model config:

class MyModel(BaseModel):
    custom_field: MyCustomType
    model_config = {"arbitrary_types_allowed": True}

Global registry not working

Make sure you're using model_dump(mode="json"):

# This uses the global registry
serialized = model.model_dump(mode="json")

# This does NOT use the global registry
serialized = model.model_dump()

Serialization errors with custom types

Register a custom serializer:

from flujo.utils import register_custom_serializer

def serialize_my_type(obj):
    return obj.to_dict()  # or appropriate serialization

register_custom_serializer(MyType, serialize_my_type)

Testing Your Migration

Run this test to verify your custom types work correctly:

from flujo.utils import register_custom_serializer
from flujo.models import BaseModel

class MyCustomType:
    def __init__(self, value):
        self.value = value

def serialize_my_type(obj):
    return {"value": obj.value}

register_custom_serializer(MyCustomType, serialize_my_type)

class TestModel(BaseModel):
    field: MyCustomType
    model_config = {"arbitrary_types_allowed": True}

# Test serialization
obj = TestModel(field=MyCustomType("test"))
serialized = obj.model_dump(mode="json")
assert serialized["field"]["value"] == "test"
print("✅ Migration successful!")

Backward Compatibility

  • ✅ All existing Pydantic models continue to work
  • ✅ All existing state backends continue to work
  • ✅ All existing cache functionality continues to work
  • ✅ All existing context functionality continues to work
  • ⚠️ Custom types may need explicit handling (see migration guide above)

What's Next

  • Enhanced documentation with more examples
  • Performance optimizations for high-throughput scenarios
  • Additional utility functions for complex serialization patterns
  • Integration with more third-party libraries

For questions or issues, please refer to the Advanced Serialization Guide or open an issue on GitHub.