What is Marshmallow?
Marshmallow is a Python library for object serialization and deserialization — converting complex data types like objects to and from native Python data types such as dictionaries, lists, strings, and numbers. It's commonly used for validating, transforming, and sanitizing data, especially when building APIs, handling user input, or working with JSON payloads.
At its core, Marshmallow provides:
- Schemas — blueprints that define how data should be structured
- Fields — individual data type descriptors with validation rules
- Serialization — converting Python objects into primitive types (dumping)
- Deserialization — converting raw input into clean, validated Python objects (loading)
- Validation — ensuring data conforms to expected formats and constraints
Why Build Marshmallow From Scratch?
🚀 Deploy your AI agent in 10 minutes
Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.
Try it free →Understanding the internals of a library like Marshmallow transforms you from a user who follows recipes into an engineer who can design data pipelines confidently. By building it from scratch, you'll grasp:
- How schema-based validation frameworks are architected
- The chain of responsibility pattern used in field validation
- How serialization and deserialization are mirror operations
- Why error collection matters for API usability
- How to design extensible, declarative APIs in Python
This knowledge applies directly to building form validators, API request parsers, configuration loaders, and any system that needs trustworthy data transformation.
Project Structure Overview
We'll build a minimal but complete Marshmallow clone with the following components:
- Field classes — String, Integer, Float, Boolean, List, Nested
- Validators — required, min/max length, range, custom callables
- Schema base class — metaclass-powered field collection
- Error collection — ValidationError with field-level messages
- Serialization engine — dump() and dumps() methods
- Deserialization engine — load() and loads() methods
Step 1: The Field Base Class
Every field in our Marshmallow clone inherits from a base Field class. This class holds the core logic: validators, default values, required flag, and the methods serialize and deserialize that each concrete field will override.
from typing import Any, Callable, List, Optional
from abc import ABC, abstractmethod
class Field(ABC):
"""Base class for all Marshmallow fields."""
def __init__(
self,
required: bool = False,
default: Any = None,
validators: Optional[List[Callable]] = None,
description: str = "",
):
self.required = required
self.default = default
self.validators = validators or []
self.description = description
self.name = None # Set by Schema metaclass
def _set_name(self, name: str):
"""Called by Schema metaclass to bind field name."""
self.name = name
@abstractmethod
def serialize(self, value: Any) -> Any:
"""Convert Python value to a serializable primitive."""
pass
@abstractmethod
def deserialize(self, value: Any) -> Any:
"""Convert raw input to a validated Python value."""
pass
def validate(self, value: Any):
"""Run all validators on the given value."""
errors = []
for validator in self.validators:
try:
validator(value)
except ValidationError as e:
errors.append(str(e))
except Exception as e:
errors.append(f"Validator raised: {e}")
if errors:
raise ValidationError({self.name: errors})Notice the _set_name method — this is how fields learn their attribute name once bound to a Schema class. The validate method runs all validators and collects errors. We'll define ValidationError shortly.
Step 2: Concrete Field Types
Now we implement the specific field types. Each overrides serialize and deserialize with type-specific coercion and validation logic.
class String(Field):
"""A field that serializes/deserializes string values."""
def serialize(self, value: Any) -> str:
if value is None:
return None
return str(value)
def deserialize(self, value: Any) -> str:
if value is None:
return None
if not isinstance(value, str):
raise ValidationError(
{self.name: [f"Expected string, got {type(value).__name__}"]}
)
return value.strip()
class Integer(Field):
"""A field that serializes/deserializes integer values."""
def serialize(self, value: Any) -> int:
if value is None:
return None
return int(value)
def deserialize(self, value: Any) -> int:
if value is None:
return None
try:
return int(value)
except (TypeError, ValueError):
raise ValidationError(
{self.name: [f"Expected integer, got {type(value).__name__}: {value}"]}
)
class Float(Field):
"""A field that serializes/deserializes float values."""
def serialize(self, value: Any) -> float:
if value is None:
return None
return float(value)
def deserialize(self, value: Any) -> float:
if value is None:
return None
try:
return float(value)
except (TypeError, ValueError):
raise ValidationError(
{self.name: [f"Expected float, got {type(value).__name__}: {value}"]}
)
class Boolean(Field):
"""A field that serializes/deserializes boolean values."""
def serialize(self, value: Any) -> bool:
if value is None:
return None
return bool(value)
def deserialize(self, value: Any) -> bool:
if value is None:
return None
if isinstance(value, bool):
return value
if isinstance(value, str):
lowered = value.lower()
if lowered in ("true", "1", "yes", "on"):
return True
if lowered in ("false", "0", "no", "off"):
return False
raise ValidationError(
{self.name: [f"Cannot coerce '{value}' to boolean"]}
)Each field handles None gracefully — if a value is None, it passes through. The actual required-check happens at the Schema level during loading. This separation keeps fields focused on type coercion.
Step 3: The List Field
A List field wraps another field and applies it to each element of a list. This demonstrates composition — a core Marshmallow pattern.
class List(Field):
"""A field that contains a list of values of a specific type."""
def __init__(self, inner_field: Field, **kwargs):
super().__init__(**kwargs)
self.inner_field = inner_field
def serialize(self, value: Any) -> list:
if value is None:
return None
if not isinstance(value, (list, tuple)):
raise ValidationError(
{self.name: [f"Expected list, got {type(value).__name__}"]}
)
return [self.inner_field.serialize(item) for item in value]
def deserialize(self, value: Any) -> list:
if value is None:
return None
if not isinstance(value, list):
raise ValidationError(
{self.name: [f"Expected list, got {type(value).__name__}"]}
)
result = []
errors = []
for i, item in enumerate(value):
try:
result.append(self.inner_field.deserialize(item))
except ValidationError as e:
errors.append({str(i): e.messages})
if errors:
raise ValidationError({self.name: errors})
return resultNotice how the List field delegates to its inner_field for each element, collecting per-index errors. This pattern scales to Nested fields as well.
Step 4: The Nested Field
A Nested field embeds one Schema inside another, enabling complex object graphs. This is where the real power of Marshmallow emerges.
class Nested(Field):
"""A field that contains a nested Schema."""
def __init__(self, nested_schema_class, **kwargs):
super().__init__(**kwargs)
self.nested_schema_class = nested_schema_class
def serialize(self, value: Any) -> dict:
if value is None:
return None
schema_instance = self.nested_schema_class()
return schema_instance.dump(value)
def deserialize(self, value: Any) -> dict:
if value is None:
return None
if not isinstance(value, dict):
raise ValidationError(
{self.name: [f"Expected dict, got {type(value).__name__}"]}
)
schema_instance = self.nested_schema_class()
result, errors = schema_instance.load(value)
if errors:
raise ValidationError({self.name: errors})
return resultThe Nested field creates a new Schema instance on each call, ensuring isolation between nested objects. This lazy instantiation is a deliberate design choice — it avoids circular import issues and keeps schemas stateless.
Step 5: ValidationError
Before we build the Schema, we need our error class. Marshmallow's ValidationError collects errors per-field and supports nesting for complex structures.
class ValidationError(Exception):
"""Exception raised when validation fails."""
def __init__(self, messages: dict):
self.messages = messages
super().__init__(self._format_messages())
def _format_messages(self) -> str:
flat = []
for field, errors in self.messages.items():
if isinstance(errors, list):
flat.append(f"{field}: {', '.join(str(e) for e in errors)}")
else:
flat.append(f"{field}: {errors}")
return " | ".join(flat)
def __repr__(self):
return f"ValidationError({self.messages})"This exception carries structured error data. When raised during deserialization, the Schema catches it and continues collecting errors from other fields before finally re-raising with the complete set.
Step 6: Built-in Validators
Validators are simple callables that raise ValidationError on failure. Here are the essential ones:
def required_validator(value):
"""Validator that ensures a value is present and not empty."""
if value is None:
raise ValidationError({"__field__": ["Value is required"]})
if isinstance(value, str) and value.strip() == "":
raise ValidationError({"__field__": ["Value cannot be empty"]})
return True
def length_validator(min_len: int = None, max_len: int = None):
"""Factory that creates a string length validator."""
def validate(value):
if not isinstance(value, str):
return True
if min_len is not None and len(value) < min_len:
raise ValidationError(
{"__field__": [f"Length must be at least {min_len}, got {len(value)}"]}
)
if max_len is not None and len(value) > max_len:
raise ValidationError(
{"__field__": [f"Length must be at most {max_len}, got {len(value)}"]}
)
return True
return validate
def range_validator(min_val: float = None, max_val: float = None):
"""Factory that creates a numeric range validator."""
def validate(value):
if value is None:
return True
if min_val is not None and value < min_val:
raise ValidationError(
{"__field__": [f"Value must be >= {min_val}, got {value}"]}
)
if max_val is not None and value > max_val:
raise ValidationError(
{"__field__": [f"Value must be <= {max_val}, got {value}"]}
)
return True
return validate
def one_of_validator(choices: list):
"""Factory that creates an enum-style validator."""
def validate(value):
if value not in choices:
raise ValidationError(
{"__field__": [f"Must be one of {choices}, got '{value}'"]}
)
return True
return validate
def email_validator(value):
"""Simple email format validator."""
if not isinstance(value, str):
return True
if "@" not in value or "." not in value.split("@")[-1]:
raise ValidationError(
{"__field__": [f"'{value}' is not a valid email address"]}
)
return TrueThese validators use the factory pattern — length_validator, range_validator, and one_of_validator return configured validator functions. This is how Marshmallow's validate.Length and validate.Range work under the hood.
Step 7: The Schema Metaclass
This is the magic that makes Marshmallow feel declarative. When you define a class like class UserSchema(Schema): name = String(), the metaclass collects all Field instances and registers them.
class SchemaMeta(type):
"""Metaclass that collects Field instances from class attributes."""
def __new__(mcs, name, bases, namespace):
fields = {}
# Collect fields defined in the class body
for key, value in list(namespace.items()):
if isinstance(value, Field):
fields[key] = value
value._set_name(key)
# Inherit fields from parent Schema classes
for base in bases:
if hasattr(base, "_declared_fields"):
inherited = {
k: v
for k, v in base._declared_fields.items()
if k not in fields
}
fields = {**inherited, **fields}
# Remove Field objects from namespace so they don't become class attributes
for key in fields:
if key in namespace:
del namespace[key]
cls = super().__new__(mcs, name, bases, namespace)
cls._declared_fields = fields
return clsThe metaclass does three critical things:
- Scans the class body for
Fieldinstances and extracts them - Sets each field's
nameattribute to its attribute name - Removes Field objects from the final class so they don't shadow instance attributes
- Merges fields from parent classes, with child fields taking precedence
Step 8: The Schema Base Class
Now we build the Schema class itself, which uses SchemaMeta and provides dump, load, and the public API.
from typing import Any, Dict, Tuple
class Schema(metaclass=SchemaMeta):
"""Base class for all Marshmallow schemas."""
def __init__(self):
# Instantiate fields for this schema instance
self.fields: Dict[str, Field] = {}
for name, field_cls in self._declared_fields.items():
# Create a fresh copy of each field
field_instance = field_cls.__class__(
required=field_cls.required,
default=field_cls.default,
validators=field_cls.validators,
description=field_cls.description,
)
field_instance._set_name(name)
self.fields[name] = field_instance
def dump(self, obj: Any) -> Dict[str, Any]:
"""Serialize an object into a dictionary of primitives."""
result = {}
for name, field in self.fields.items():
value = getattr(obj, name, None)
try:
result[name] = field.serialize(value)
except ValidationError as e:
# Serialization errors are rare but possible
raise ValidationError({name: e.messages})
return result
def dumps(self, obj: Any) -> str:
"""Serialize an object to a JSON string."""
import json
return json.dumps(self.dump(obj))
def load(self, data: Dict[str, Any]) -> Tuple[Dict[str, Any], Dict[str, Any]]:
"""Deserialize a dictionary, returning (valid_data, errors)."""
result = {}
errors = {}
for name, field in self.fields.items():
raw_value = data.get(name)
# Handle missing required fields
if raw_value is None and field.required:
errors[name] = ["Field is required"]
continue
# Use default if value is missing
if raw_value is None and field.default is not None:
result[name] = field.default
continue
# Skip if value is missing and not required
if raw_value is None:
result[name] = None
continue
try:
deserialized = field.deserialize(raw_value)
field.validate(deserialized)
result[name] = deserialized
except ValidationError as e:
errors.update(e.messages)
if errors:
return result, errors
return result, {}
def loads(self, json_string: str) -> Tuple[Dict[str, Any], Dict[str, Any]]:
"""Deserialize a JSON string, returning (valid_data, errors)."""
import json
data = json.loads(json_string)
return self.load(data)Key design decisions in this implementation:
- Fields are instantiated fresh in
__init__to avoid mutable state sharing across Schema instances load()returns a tuple of(data, errors)rather than raising immediately — this allows collecting all errors at once- The required check happens at the Schema level, not inside fields, keeping field logic focused on type coercion
- Default values are applied when raw input is missing, not when it's invalid
Step 9: Putting It All Together — A Complete Example
Let's define a realistic schema and exercise both serialization and deserialization:
# Define our data model (could be a dataclass, ORM model, or plain class)
class Address:
def __init__(self, street, city, zip_code):
self.street = street
self.city = city
self.zip_code = zip_code
class User:
def __init__(self, name, age, email, address):
self.name = name
self.age = age
self.email = email
self.address = address
# Define schemas
class AddressSchema(Schema):
street = String(required=True, validators=[length_validator(min_len=3)])
city = String(required=True)
zip_code = String(
required=True,
validators=[length_validator(min_len=5, max_len=10)]
)
class UserSchema(Schema):
name = String(
required=True,
validators=[length_validator(min_len=2, max_len=50)]
)
age = Integer(
required=True,
validators=[range_validator(min_val=0, max_val=150)]
)
email = String(
required=True,
validators=[email_validator]
)
address = Nested(AddressSchema)
# === SERIALIZATION (dump) ===
user = User(
name="Alice Smith",
age=30,
email="alice@example.com",
address=Address(
street="123 Main St",
city="Springfield",
zip_code="12345"
)
)
schema = UserSchema()
serialized = schema.dump(user)
print("Serialized:", serialized)
# Output: {'name': 'Alice Smith', 'age': 30, 'email': 'alice@example.com',
# 'address': {'street': '123 Main St', 'city': 'Springfield', 'zip_code': '12345'}}
json_output = schema.dumps(user)
print("JSON:", json_output)Now let's test deserialization with various scenarios:
# === DESERIALIZATION (load) — Happy Path ===
raw_data = {
"name": "Bob Jones",
"age": "25", # String that should be coerced to int
"email": "bob@test.org",
"address": {
"street": "456 Oak Ave",
"city": "Metropolis",
"zip_code": "98765"
}
}
clean_data, errors = UserSchema().load(raw_data)
print("Clean data:", clean_data)
print("Errors:", errors)
# age is now int 25, all fields validated
# === DESERIALIZATION — With Errors ===
bad_data = {
"name": "X", # Too short
"age": "-5", # Out of range
"email": "not-an-email", # Invalid format
"address": {
"street": "A", # Too short
"city": "Gotham",
"zip_code": "bad-zip-format-very-long" # Too long
}
}
clean_data, errors = UserSchema().load(bad_data)
print("Clean data:", clean_data)
print("Errors:")
import json
print(json.dumps(errors, indent=2))The error output would look something like:
{
"name": ["Length must be at least 2, got 1"],
"age": ["Value must be >= 0, got -5"],
"email": ["'not-an-email' is not a valid email address"],
"address": {
"street": ["Length must be at least 3, got 1"],
"zip_code": ["Length must be at most 10, got 21"]
}
}This demonstrates the power of nested error collection — errors propagate up from nested schemas while preserving their structural path.
Step 10: Adding Custom Field-Level Validation
Sometimes you need validation that spans multiple fields or requires database access. Marshmallow supports this via schema-level validate hooks. Let's implement one:
class Schema(metaclass=SchemaMeta):
# ... (previous code unchanged)
def validate_schema(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""Override this method to add schema-level validation.
Return a dict of field_name: [error_messages] or empty dict.
"""
return {}
def load(self, data: Dict[str, Any]) -> Tuple[Dict[str, Any], Dict[str, Any]]:
result = {}
errors = {}
for name, field in self.fields.items():
raw_value = data.get(name)
if raw_value is None and field.required:
errors[name] = ["Field is required"]
continue
if raw_value is None and field.default is not None:
result[name] = field.default
continue
if raw_value is None:
result[name] = None
continue
try:
deserialized = field.deserialize(raw_value)
field.validate(deserialized)
result[name] = deserialized
except ValidationError as e:
errors.update(e.messages)
# Run schema-level validation
schema_errors = self.validate_schema(result)
errors.update(schema_errors)
if errors:
return result, errors
return result, {}
class RegistrationSchema(Schema):
password = String(required=True, validators=[length_validator(min_len=8)])
password_confirm = String(required=True)
def validate_schema(self, data):
errors = {}
if data.get("password") != data.get("password_confirm"):
errors["password_confirm"] = ["Passwords do not match"]
return errorsThis pattern allows cross-field validation without polluting individual field logic.
Step 11: The @post_load Hook
Marshmallow provides decorators like @post_load to transform deserialized data. Here's how to implement it:
from functools import wraps
class Schema(metaclass=SchemaMeta):
# ... (previous code)
def __init__(self):
self.fields = {}
for name, field_cls in self._declared_fields.items():
field_instance = field_cls.__class__(
required=field_cls.required,
default=field_cls.default,
validators=field_cls.validators,
description=field_cls.description,
)
field_instance._set_name(name)
self.fields[name] = field_instance
# Collect post_load hooks
self._post_load_hooks = []
for attr_name in dir(self):
attr = getattr(self, attr_name)
if callable(attr) and getattr(attr, '_is_post_load', False):
self._post_load_hooks.append(attr)
@staticmethod
def post_load(func):
"""Decorator to mark a method as a post-load hook."""
@wraps(func)
def wrapper(self, data, **kwargs):
return func(self, data, **kwargs)
wrapper._is_post_load = True
return wrapper
def load(self, data: Dict[str, Any]) -> Tuple[Dict[str, Any], Dict[str, Any]]:
result = {}
errors = {}
for name, field in self.fields.items():
raw_value = data.get(name)
if raw_value is None and field.required:
errors[name] = ["Field is required"]
continue
if raw_value is None and field.default is not None:
result[name] = field.default
continue
if raw_value is None:
result[name] = None
continue
try:
deserialized = field.deserialize(raw_value)
field.validate(deserialized)
result[name] = deserialized
except ValidationError as e:
errors.update(e.messages)
schema_errors = self.validate_schema(result)
errors.update(schema_errors)
if errors:
return result, errors
# Run post_load hooks
for hook in self._post_load_hooks:
result = hook(self, result)
return result, {}
class UserSchema(Schema):
name = String(required=True)
age = Integer(required=True)
@Schema.post_load
def add_full_name_flag(self, data):
"""Post-load hook that adds a computed field."""
if " " in data.get("name", ""):
data["has_full_name"] = True
return dataThe @post_load decorator marks methods that should run after successful deserialization. These hooks receive the cleaned data dict and can transform, enrich, or convert it before returning to the caller.
Best Practices When Using Marshmallow
Keep Schemas Thin and Focused
A Schema should represent one coherent data structure. If you find yourself with a schema handling 20+ fields across multiple contexts, split it into smaller, composable schemas using Nested fields. This improves testability and reduces coupling.
Use Validators Over Custom Field Logic
Whenever possible, express constraints as composable validators rather than overriding field methods. Validators are reusable, testable in isolation, and can be combined freely. A field with validators=[length_validator(min_len=3), one_of_validator(["admin", "user"])] is clearer than a custom field subclass with hardcoded checks.
Handle Errors Gracefully
Never let a single validation failure prevent collecting all errors. The tuple-return pattern (data, errors) we implemented ensures users see all problems at once, dramatically improving developer experience and reducing the "fix one error, resubmit, discover another" loop.
Version Your Schemas
When building APIs, treat schemas as part of your contract. Version them explicitly — UserSchemaV1, UserSchemaV2 — rather than modifying fields in place. This prevents breaking existing consumers when you add or change validation rules.
Test Both Serialization and Deserialization
Many teams only test deserialization (loading input). But serialization bugs — where your API returns malformed data — are equally damaging. Write tests that round-trip data through dump and load to ensure consistency.
Document Fields With Descriptions
The description parameter we included on every field isn't just for show — it can power auto-generated API documentation, form labels, and help text. Use it consistently.
Conclusion
Building Marshmallow from scratch reveals that its elegance comes from a small set of well-composed patterns: a metaclass for declarative field collection, polymorphic field types with clear serialization/deserialization contracts, composable validators, and structured error collection. The entire system — the real Marshmallow library included — rests on roughly a dozen core abstractions.
You now understand not just how to use Marshmallow, but how to think about data validation frameworks in general. The patterns here — declarative field definitions, two-phase serialization/deserialization, error accumulation, and post-processing hooks — apply to form libraries, API gateways, ETL pipelines, and any system that must transform untrusted input into trusted data. Build on this foundation, and you'll design better data interfaces in every project.