haive.core.schema.compatibility.field_mapping

Advanced field mapping with path resolution and transformations.

Classes

FieldMapper	Manages field mappings between schemas.
FieldMapping	Represents a mapping between fields with transformation.

Functions

create_mapping(mappings[, computed_fields])

Create a field mapper from simple mapping dict.

Module Contents

class haive.core.schema.compatibility.field_mapping.FieldMapper

Manages field mappings between schemas.

Init .

Returns:

Add return description]

Return type:

[TODO

add_aggregate_field(sources, target, aggregator, default=None)

Add an aggregate field from multiple sources.

Parameters:

• sources (list[str])

• target (str)

• aggregator (collections.abc.Callable[[list[Any]], Any])

• default (Any)

Return type:

FieldMapping

add_computed_field(target, generator, condition=None)

Add a computed field with no source.

Parameters:

• target (str)

• generator (collections.abc.Callable[[], Any])

• condition (collections.abc.Callable[[dict[str, Any]], bool] | None)

Return type:

FieldMapping

add_mapping(source, target, transformer=None, condition=None, default=None, validator=None)

Add a field mapping.

Parameters:

• source (str | list[str])

• target (str)

• transformer (collections.abc.Callable[[Any], Any] | None)

• condition (collections.abc.Callable[[dict[str, Any]], bool] | None)

• default (Any)

• validator (collections.abc.Callable[[Any], bool] | None)

Return type:

FieldMapping

get_mapping_for_target(target_field)

Get mapping for a target field.

Parameters:

target_field (str)

Return type:

FieldMapping | None

get_targets_for_source(source_field)

Get all target fields that use a source field.

Parameters:

source_field (str)

Return type:

set[str]

map_data(source_data, target_fields=None, include_unmapped=False, context=None)

Map source data to target schema.

Parameters:

• source_data (dict[str, Any]) – Source data dictionary

• target_fields (set[str] | None) – Specific fields to map (None = all)

• include_unmapped (bool) – Include unmapped source fields

• context (haive.core.schema.compatibility.types.ConversionContext | None) – Conversion context for tracking

Returns:

Mapped data dictionary

Return type:

dict[str, Any]

to_dict()

Export mappings as dictionary.

Return type:

dict[str, dict[str, Any]]

validate_mappings(source_fields, target_fields)

Validate that mappings are complete and valid.

Returns:

Tuple of (is_valid, issues)

Parameters:

• source_fields (set[str])

• target_fields (set[str])

Return type:

tuple[bool, list[str]]

class haive.core.schema.compatibility.field_mapping.FieldMapping

Represents a mapping between fields with transformation.

apply(source_data)

Apply mapping to source data.

Returns:

Tuple of (success, value)

Parameters:

source_data (dict[str, Any])

Return type:

tuple[bool, Any]

haive.core.schema.compatibility.field_mapping.create_mapping(mappings, computed_fields=None)

Create a field mapper from simple mapping dict.

Parameters:

• mappings (dict[str, str | tuple[str, collections.abc.Callable]]) – Dict of target -> source or (source, transformer)

• computed_fields (dict[str, collections.abc.Callable] | None) – Dict of target -> generator function

Return type:

FieldMapper