Schema Handling¶

Summary¶

This chapter covers the schema handling layer that abstracts over different data format schemas through a protocol-based registry. You will learn about the SchemaHandler protocol and its methods (can_handle, get_fields, get_source_name, get_cardinality, get_choice_groups), the SchemaHandlerRegistry with its default registry and handler registration mechanism, and the To FieldSpec Conversion that bridges schema information into the TreeSpec type system.

Concepts Covered¶

SchemaHandler Protocol
SchemaHandlerRegistry
Default Registry
Handler Registration
Can Handle Check
Get Fields Method
Get Source Name Method
Get Cardinality Method
Get Choice Groups
To FieldSpec Conversion

Prerequisites¶

Chapter 1: Foundation Concepts (Python Protocols, Decorator Pattern)
Chapter 4: Type System (SchemaFieldInfo Protocol, UniversalType Enum, FieldSpec Bridge)

Learning Objectives¶

After completing this chapter, you will be able to:

Implement the SchemaHandler protocol for a new data format
Register handlers with the SchemaHandlerRegistry
Describe how can_handle dispatches to the correct handler
Extract field information, source names, cardinality, and choice groups
Convert schema field information to FieldSpec objects

Introduction¶

The schema handling layer is the abstraction that makes mountainash-treespec format-agnostic. Without it, the flatten and build engines would need separate code paths for every data format: one for Python dataclasses, another for xsdata XML classes, another for Protocol Buffer messages, another for Pydantic models, and so on. The schema handling layer replaces this proliferation with a single protocol that all format handlers implement, and a registry that dispatches to the correct handler based on the input class.

This design follows the Strategy pattern: the engines depend on the SchemaHandler protocol interface, and concrete handlers are plugged in at runtime. Adding support for a new format requires implementing a handler class and registering it -- no changes to the core engines.

The chapter proceeds from the protocol definition (what handlers must do) through the registry (how handlers are discovered and selected) to the individual protocol methods (what information each method provides).

SchemaHandler Protocol¶

The SchemaHandler protocol defines the complete interface that a format handler must satisfy. It is decorated with @runtime_checkable, meaning you can use isinstance(handler, SchemaHandler) to verify conformance at runtime, not just at type-check time.

The protocol declares seven core methods and three JSON interop methods:

@runtime_checkable
class SchemaHandler(Protocol):
    def can_handle(self, cls: type) -> bool: ...
    def get_fields(self, cls: type) -> list[SchemaFieldInfo]: ...
    def get_source_name(self, cls: type) -> str: ...
    def get_cardinality(self, cls: type, field_name: str) -> tuple[int, int | None]: ...
    def get_choice_groups(self, cls: type) -> list[list[str]]: ...
    def to_fieldspec(self, field_info: SchemaFieldInfo) -> FieldSpec: ...
    def construct(self, cls: type, kwargs: dict[str, Any]) -> Any: ...

    # JSON interop
    def to_json_dict(self, tree: Any) -> dict[str, Any]: ...
    def from_json_dict(self, data: dict, target_class: type) -> Any: ...
    def from_json_str(self, raw: str, target_class: type) -> Any: ...

The core methods serve the flatten and build engines. The JSON interop methods enable round-trip serialization to JSON for testing, debugging, and API integration. A handler that does not need JSON support can implement these methods as simple pass-throughs.

Each method has a well-defined responsibility:

can_handle -- Does this handler know how to introspect cls?
get_fields -- What scalar and nested fields does cls have?
get_source_name -- What is the canonical name of cls in its source format?
get_cardinality -- What are the min/max occurrence bounds for a specific field?
get_choice_groups -- Which fields are mutually exclusive (XML choice groups)?
to_fieldspec -- Convert handler-specific field info to the universal FieldSpec format
construct -- Build an instance of cls from a keyword argument dictionary

Can Handle Check¶

The can_handle method is the dispatch discriminator. When the registry receives a class to introspect, it iterates through registered handlers and calls can_handle on each until one returns True. That handler is then used for all subsequent operations on that class type.

Each format handler implements can_handle with a type check specific to its format:

Handler	can_handle implementation
DataclassSchemaHandler	`isinstance(cls, type) and dataclasses.is_dataclass(cls)`
XsdataSchemaHandler	`is_dataclass(cls) and hasattr(cls, "Meta") and has xsdata field metadata`
PydanticSchemaHandler	`isinstance(cls, type) and issubclass(cls, BaseModel)`
ProtobufSchemaHandler	`isinstance(cls, type) and issubclass(cls, pb_message.Message)`

The ordering matters. The XsdataSchemaHandler checks for xsdata-specific metadata (Meta class and "type" key in field metadata) to distinguish xsdata dataclasses from plain dataclasses. Since handlers are tried in reverse registration order (last registered wins), more specific handlers take priority over more general ones.

Diagram: Handler Dispatch Chain¶

Handler Dispatch Chain

Type: workflow sim-id: handler-dispatch-chain
Library: p5.js
Status: Specified

Purpose: Visualize the handler selection process -- a class enters the registry, each handler's can_handle is tried in order, and the first match wins. Shows four different input class types being dispatched to their respective handlers.

Components:

Input class type entering from the left
Chain of handler boxes with can_handle conditions
Green/red indicators showing match/no-match for each handler
Output: the selected handler highlighted
Four example classes cycling through the animation

Interactions: Click different input class types to see which handler catches them. Reorder handlers by dragging to see how priority changes affect dispatch. Toggle "verbose mode" to see the actual can_handle return values.

Learning Objective: (Apply, Bloom Level 3) Students can predict which handler will be selected for a given class type based on the registry's registration order.

SchemaHandlerRegistry¶

The SchemaHandlerRegistry class manages the collection of registered handlers and provides the get_handler method that the rest of the system uses to obtain the appropriate handler for any class.

The registry stores handlers in a list and uses reverse-registration order for priority. When you call register(handler), the new handler is inserted at the front of the list. This means the most recently registered handler gets the first chance to handle any class. The design supports overriding: if you register a more specific handler after a general one, the specific handler wins.

class SchemaHandlerRegistry:
    def __init__(self):
        self._handlers: list[SchemaHandler] = []
        self._plugins_discovered: bool = False

    def register(self, handler: SchemaHandler) -> None:
        self._handlers.insert(0, handler)

    def get_handler(self, cls: type) -> SchemaHandler:
        if not self._plugins_discovered:
            self._discover_plugins()
        for handler in self._handlers:
            if handler.can_handle(cls):
                return handler
        raise UnsupportedSchemaError(...)

If no handler can handle the class, get_handler raises an UnsupportedSchemaError with a diagnostic message listing all registered handlers. This helps users identify whether they forgot to install a format-specific dependency.

Handler Registration¶

Handler registration occurs through two mechanisms: explicit registration via register() and automatic plugin discovery via Python entry points.

Explicit registration happens when the default_registry module is imported. The module-level code registers built-in handlers in order of specificity:

DataclassSchemaHandler() -- registered first (lowest priority, catches all dataclasses)
XsdataSchemaHandler() -- registered second (higher priority for xsdata classes)
PydanticSchemaHandler() -- registered third (catches Pydantic BaseModel subclasses)
ProtobufSchemaHandler() -- registered last (highest priority for protobuf messages)

Optional handlers (Xsdata, Pydantic, Protobuf) are wrapped in try/except ImportError blocks so the library works without those dependencies installed.

Plugin discovery uses the mountainash_treespec.schema_handlers entry point group. Third-party packages can register handlers by declaring an entry point in their pyproject.toml:

[project.entry-points."mountainash_treespec.schema_handlers"]
my_handler = "my_package.handler:MySchemaHandler"

The _discover_plugins method is called lazily on first get_handler invocation. It iterates through all entry points in the group, loads each handler class, instantiates it, and registers it. Plugin handlers get highest priority since they are discovered and registered after built-in handlers.

Default Registry¶

The default_registry is a module-level singleton instance of SchemaHandlerRegistry. It is the registry that the flatten engine, build engine, and Arrow Schema Builder use by default when no explicit registry is provided.

default_registry = SchemaHandlerRegistry()

from mountainash_treespec.formats.json.handler import DataclassSchemaHandler
default_registry.register(DataclassSchemaHandler())

try:
    from mountainash_treespec.formats.xml.handler import XsdataSchemaHandler
    default_registry.register(XsdataSchemaHandler())
except ImportError:
    pass

The default registry provides a "just works" experience for common use cases. A project using plain dataclasses needs no configuration -- the DataclassSchemaHandler handles everything. A project using xsdata classes with xsdata installed gets automatic xsdata support. Only unusual formats or custom handlers require explicit registry manipulation.

The TreeBuilder constructor accepts an optional handler_registry parameter to override the default. This is useful in testing (where you want to control exactly which handlers are available) or when a project needs a custom handler priority order.

Get Fields Method¶

The get_fields method is the primary introspection method. Given a class, it returns a list of SchemaFieldInfo objects describing every field on that class -- their names, types, optionality, collection status, and nesting status.

The information returned by get_fields drives several downstream decisions:

The flattener uses is_nested to distinguish scalar columns from child relationships
The type system uses universal_type to determine PyArrow types
The build engine uses field names to construct instances with the correct keyword arguments
The Arrow Schema Builder uses the complete field list to discover all columns a table will have

Different handlers implement get_fields differently based on their format's introspection mechanisms:

DataclassSchemaHandler uses dataclasses.fields() and typing.get_type_hints() to unwrap Optional, List, and Annotated wrappers
XsdataSchemaHandler reads field metadata dictionaries for richer type and constraint information
PydanticSchemaHandler iterates cls.model_fields and reads Pydantic v2 FieldInfo objects
ProtobufSchemaHandler walks the protobuf Descriptor to extract field definitions

Despite these different mechanisms, all handlers produce the same SchemaFieldInfo protocol-conformant output.

Get Source Name Method¶

The get_source_name method returns the canonical name of a class in its source format. For most handlers, this is simply the class name. For XML handlers, it might be the XML element name from the class's Meta.name attribute.

# DataclassSchemaHandler
def get_source_name(self, cls: type) -> str:
    return cls.__name__

This method is used by the spec derivation logic to suggest table names and node names based on the source schema. It provides the "natural" name that a user would expect when auto-generating a TreeSpec from a class hierarchy.

Get Cardinality Method¶

The get_cardinality method returns the minimum and maximum occurrence bounds for a specific field on a class. The return type is tuple[int, int | None] where None for the maximum means "unbounded" (a list field with no upper limit).

Cardinality mappings:

Required scalar field: (1, 1) -- exactly one occurrence
Optional scalar field: (0, 1) -- zero or one occurrence
Required list field: (0, None) -- zero or more occurrences (lists are always 0..N)
Optional list field: (0, None) -- same as required list

This information is used during spec derivation to suggest appropriate cardinality values for ChildSpec declarations. A field with max=1 suggests cardinality: "one" while max=None suggests cardinality: "many".

Get Choice Groups¶

The get_choice_groups method returns lists of mutually exclusive field names. In XML Schema, a <xs:choice> element means that only one of several possible child elements can appear. This constraint has no direct equivalent in Python type systems, so it must be discovered through format-specific metadata.

# DataclassSchemaHandler (no choice support)
def get_choice_groups(self, cls: type) -> list[list[str]]:
    return []

For the Xsdata handler, choice groups are extracted from field metadata where multiple fields share the same "choice" identifier. This information is passed through to the spec derivation logic, which uses it to generate appropriate routing or conditional configurations.

For handlers that do not support choice semantics (plain dataclasses, Pydantic, Protobuf oneofs are handled differently), the method returns an empty list.

Diagram: SchemaHandler Protocol Methods¶

SchemaHandler Protocol Methods

Type: infographic sim-id: schema-handler-methods
Library: p5.js
Status: Specified

Purpose: Interactive reference card showing all SchemaHandler protocol methods, their signatures, return types, and which engine component consumes each method's output.

Components:

Method cards arranged in a grid, each showing: method name, parameters, return type, description
Arrows from each method to its consumers (Flattener, Builder, Arrow Schema Builder, Spec Validator)
Color coding: blue for introspection methods, green for construction methods, orange for JSON interop
Example return values for a sample Patient dataclass

Interactions: Click a method card to expand its full documentation and example. Click a consumer to highlight which methods it uses. Toggle between "API view" (signatures) and "example view" (sample data).

Learning Objective: (Remember, Bloom Level 1) Students can recall the purpose and signature of each SchemaHandler protocol method.

To FieldSpec Conversion¶

The to_fieldspec method bridges from the handler-specific SchemaFieldInfo representation to the universal FieldSpec type used across the mountainash ecosystem. Each handler implements this conversion, typically normalizing the universal_type to an enum value and extracting constraints from format-specific metadata.

The conversion serves as a standardization point. Different handlers may represent optionality, constraints, and types differently in their SchemaFieldInfo output. The to_fieldspec method normalizes these differences into a single canonical form.

For the DataclassSchemaHandler, the conversion is minimal:

def to_fieldspec(self, field_info: _FieldInfo) -> FieldSpec:
    ut = field_info.universal_type
    if isinstance(ut, str):
        ut = UniversalType(ut)
    constraints = None
    if not field_info.is_optional:
        constraints = FieldConstraints(required=True)
    return FieldSpec(name=field_info.name, type=ut, constraints=constraints)

For the XsdataSchemaHandler, the conversion is richer, extracting min/max length, patterns, and numeric bounds from xsdata field metadata into FieldConstraints.

The FieldSpec objects produced by this method are consumed by code generators, API schema derivation tools, and documentation generators that need format-neutral type information.

Implementing a Custom Handler¶

To add support for a new data format, you implement a class that satisfies the SchemaHandler protocol. Here is a skeleton for a hypothetical MessagePack handler:

class MsgpackSchemaHandler:
    def can_handle(self, cls: type) -> bool:
        return hasattr(cls, "__msgpack_fields__")

    def get_fields(self, cls: type) -> list[SchemaFieldInfo]:
        # Introspect cls.__msgpack_fields__ to build field info
        ...

    def get_source_name(self, cls: type) -> str:
        return cls.__name__

    def get_cardinality(self, cls: type, field_name: str) -> tuple[int, int | None]:
        # Check field's repetition metadata
        ...

    def get_choice_groups(self, cls: type) -> list[list[str]]:
        return []

    def to_fieldspec(self, field_info: SchemaFieldInfo) -> FieldSpec:
        ...

    def construct(self, cls: type, kwargs: dict[str, Any]) -> Any:
        return cls(**kwargs)

    def to_json_dict(self, tree: Any) -> dict[str, Any]:
        ...

    def from_json_dict(self, data: dict, target_class: type) -> Any:
        ...

    def from_json_str(self, raw: str, target_class: type) -> Any:
        ...

Register it with:

from mountainash_treespec.core.schema.registry import default_registry
default_registry.register(MsgpackSchemaHandler())

No inheritance from a base class is needed. The handler must only satisfy the protocol's structural contract.

Handler Priority

Since register() inserts at the front, your custom handler will have higher priority than all previously registered handlers. If your can_handle is too broad (e.g., returning True for all classes), it will intercept classes that should go to built-in handlers. Keep can_handle as specific as possible.

Key Takeaways¶

The SchemaHandler protocol defines a ten-method interface that all format handlers must satisfy, enabling format-agnostic introspection and construction.
The SchemaHandlerRegistry stores handlers in priority order (last registered wins) and dispatches via can_handle to select the appropriate handler for each class.
The default_registry provides built-in support for dataclasses, xsdata, Pydantic, and Protobuf with graceful degradation when optional dependencies are missing.
Handler registration occurs both explicitly at import time and automatically via Python entry points for third-party plugins.
can_handle is the dispatch discriminator; keep it as specific as possible to avoid intercepting classes meant for other handlers.
get_fields is the primary introspection method, producing SchemaFieldInfo objects that drive the flattener, builder, and Arrow Schema Builder.
get_cardinality and get_choice_groups provide schema-level metadata for spec derivation and validation.
to_fieldspec normalizes handler-specific field information into the universal FieldSpec format for ecosystem integration.