🔧 Advanced JSON Serialization — Custom Encoders and Validation

Professional applications need robust serialization with validation. Learn to build custom encoders, validate schemas, and transform complex data structures reliably.

🎯 Custom JSON Encoders

import json
from datetime import datetime, date
from decimal import Decimal
from uuid import UUID
from enum import Enum

class AdvancedEncoder(json.JSONEncoder):
    """Custom encoder for complex types"""

    def default(self, obj):
        # Datetime handling
        if isinstance(obj, (datetime, date)):
            return obj.isoformat()

        # UUID handling
        if isinstance(obj, UUID):
            return str(obj)

        # Decimal handling
        if isinstance(obj, Decimal):
            return float(obj)

        # Enum handling
        if isinstance(obj, Enum):
            return obj.value

        # Sets to lists
        if isinstance(obj, set):
            return list(obj)

        # Bytes to string
        if isinstance(obj, bytes):
            return obj.decode('utf-8')

        return super().default(obj)

# Usage
from decimal import Decimal
from uuid import uuid4

data = {
    'id': uuid4(),
    'created': datetime.now(),
    'balance': Decimal('99.99'),
    'tags': {'python', 'api', 'json'},
    'status': 'active'
}

json_string = json.dumps(data, cls=AdvancedEncoder, indent=2)
print(json_string)

✅ Schema Validation with Pydantic

from pydantic import BaseModel, EmailStr, Field, validator
from typing import Optional, List
from datetime import datetime

# Define data model
class User(BaseModel):
    id: int
    name: str = Field(..., min_length=1, max_length=100)
    email: EmailStr
    age: Optional[int] = Field(None, ge=0, le=150)
    tags: List[str] = []
    created_at: datetime = Field(default_factory=datetime.now)

    @validator('name')
    def name_not_empty(cls, v):
        if not v.strip():
            raise ValueError('Name must not be empty')
        return v.title()

    @validator('tags')
    def tags_not_empty(cls, v):
        return [tag.lower() for tag in v]

# Validation on instantiation
try:
    user = User(
        id=1,
        name='alice johnson',
        email='alice@example.com',
        age=28,
        tags=['Python', 'API']
    )
    print(user.dict())
    # id=1, name='Alice Johnson', tags=['python', 'api']

except ValueError as e:
    print(f"Validation error: {e}")

# From JSON string
user_json = '{"id": 2, "name": "Bob", "email": "bob@example.com"}'
user = User.parse_raw(user_json)

# To JSON
json_str = user.json(indent=2)

📋 Marshmallow Schemas

from marshmallow import Schema, fields, validate, post_load, pre_dump
from datetime import datetime

class PostSchema(Schema):
    """Marshmallow schema for post data"""

    id = fields.Int(dump_only=True)
    title = fields.Str(required=True, validate=validate.Length(min=1, max=200))
    content = fields.Str(required=True)
    author_id = fields.Int(required=True)
    tags = fields.List(fields.Str(), dump_default=[])
    created_at = fields.DateTime(dump_only=True)
    updated_at = fields.DateTime(dump_only=True)
    status = fields.Str(validate=validate.OneOf(['draft', 'published', 'archived']))

    @post_load
    def make_post(self, data, **kwargs):
        # Transform data after validation
        return Post(**data)

    @pre_dump
    def prepare_for_dump(self, data, **kwargs):
        # Prepare data before serialization
        if hasattr(data, '__dict__'):
            return data.__dict__
        return data

class Post:
    def __init__(self, title, content, author_id, tags=None, status='draft'):
        self.title = title
        self.content = content
        self.author_id = author_id
        self.tags = tags or []
        self.status = status
        self.created_at = datetime.now()
        self.updated_at = datetime.now()

# Usage
schema = PostSchema()

# Load and validate
post_data = {
    'title': 'My First Post',
    'content': 'Great content...',
    'author_id': 1,
    'tags': ['python', 'api']
}

post = schema.load(post_data)

# Dump to JSON
json_output = schema.dump(post)

🔄 Complex Nested Validation

from pydantic import BaseModel, validator
from typing import List, Optional

class Address(BaseModel):
    street: str
    city: str
    state: str
    zip_code: str

    @validator('zip_code')
    def validate_zip(cls, v):
        if not v.isdigit() or len(v) != 5:
            raise ValueError('Invalid ZIP code')
        return v

class ContactInfo(BaseModel):
    email: str
    phone: Optional[str] = None
    address: Address

class UserProfile(BaseModel):
    name: str
    contact: ContactInfo
    preferences: dict = {}

# Nested validation
data = {
    'name': 'Alice',
    'contact': {
        'email': 'alice@example.com',
        'phone': '555-1234',
        'address': {
            'street': '123 Main St',
            'city': 'Springfield',
            'state': 'IL',
            'zip_code': '62701'
        }
    }
}

profile = UserProfile(**data)
print(profile.json(indent=2))

🔌 Custom Serialization Logic

import json
from typing import Any, Dict

class DynamicSerializer:
    """Serialize objects with custom logic"""

    def __init__(self):
        self.handlers = {}

    def register_handler(self, type_name: str, handler):
        """Register custom handler for type"""
        self.handlers[type_name] = handler

    def serialize(self, obj: Any) -> Dict:
        """Serialize object using handlers"""

        type_name = type(obj).__name__

        if type_name in self.handlers:
            return self.handlers[type_name](obj)

        # Default serialization
        if hasattr(obj, '__dict__'):
            return obj.__dict__

        return str(obj)

# Example: Custom handlers
serializer = DynamicSerializer()

# Handler for User objects
def serialize_user(user):
    return {
        'id': user.id,
        'name': user.name,
        'email': user.email,
        '_type': 'User'
    }

# Handler for datetime
def serialize_datetime(dt):
    return dt.isoformat()

serializer.register_handler('User', serialize_user)
serializer.register_handler('datetime', serialize_datetime)

# Usage
class User:
    def __init__(self, id, name, email):
        self.id = id
        self.name = name
        self.email = email

user = User(1, 'Alice', 'alice@example.com')
json_output = json.dumps(
    serializer.serialize(user),
    indent=2
)

🔐 Selective Field Serialization

from pydantic import BaseModel, Field
from typing import Optional

class User(BaseModel):
    id: int
    name: str
    email: str
    password_hash: str = Field(..., exclude=True)  # Never serialize
    ssn: Optional[str] = Field(None, exclude=True)
    api_key: Optional[str] = Field(None, exclude=True)

    class Config:
        # Fine-grained control
        fields = {
            'password_hash': {'exclude': True},
            'ssn': {'exclude': True}
        }

# Safe serialization
user = User(
    id=1,
    name='Alice',
    email='alice@example.com',
    password_hash='hashed_password',
    ssn='123-45-6789'
)

# Sensitive fields are excluded
public_json = user.json()
# {"id": 1, "name": "Alice", "email": "alice@example.com"}

# Selective fields
partial_json = user.json(include={'id', 'name'})
# {"id": 1, "name": "Alice"}

🔄 Versioned Serialization

from pydantic import BaseModel
from typing import Literal

class UserV1(BaseModel):
    id: int
    name: str
    email: str

class UserV2(BaseModel):
    id: int
    name: str
    email: str
    created_at: str
    updated_at: str
    status: str

def serialize_user(user_data: dict, version: int = 1):
    """Serialize user based on API version"""

    if version == 1:
        return UserV1(**{
            'id': user_data['id'],
            'name': user_data['name'],
            'email': user_data['email']
        }).dict()

    elif version == 2:
        return UserV2(**user_data).dict()

    else:
        raise ValueError(f'Unsupported version: {version}')

# Usage
user_data = {
    'id': 1,
    'name': 'Alice',
    'email': 'alice@example.com',
    'created_at': '2024-01-01T12:00:00',
    'updated_at': '2024-02-23T10:30:00',
    'status': 'active'
}

v1_response = serialize_user(user_data, version=1)
v2_response = serialize_user(user_data, version=2)

✅ Complete Validation Pipeline

from pydantic import BaseModel, validator, ValidationError
import json
from typing import List, Dict, Any

class BlogPost(BaseModel):
    title: str
    content: str
    author_id: int
    tags: List[str] = []
    draft: bool = False

    @validator('title')
    def title_length(cls, v):
        if len(v) < 10:
            raise ValueError('Title too short')
        return v

    @validator('content')
    def content_not_empty(cls, v):
        if not v.strip():
            raise ValueError('Content required')
        return v

class APIRequest(BaseModel):
    """Request with post data"""
    method: str
    posts: List[BlogPost]

def process_request(request_json: str) -> Dict[str, Any]:
    """Process and validate request"""

    try:
        data = json.loads(request_json)
        request = APIRequest(**data)

        # Process valid data
        return {
            'status': 'success',
            'posts_count': len(request.posts),
            'data': request.dict()
        }

    except json.JSONDecodeError as e:
        return {
            'status': 'error',
            'error': 'Invalid JSON',
            'details': str(e)
        }

    except ValidationError as e:
        return {
            'status': 'error',
            'error': 'Validation failed',
            'errors': e.errors()
        }

# Test
request_json = '''
{
    "method": "POST",
    "posts": [
        {
            "title": "My First Article",
            "content": "Great content here...",
            "author_id": 1,
            "tags": ["python", "api"]
        }
    ]
}
'''

result = process_request(request_json)
print(json.dumps(result, indent=2))

✅ Key Takeaways

🔗 What's Next?

Learn advanced error handling strategies for production APIs.

Next: Advanced Error Handling →

Ready for advanced challenges? Try advanced challenges