Fields Mapping

Metaxy provides a few helpers when defining field-level dependencies:

• the default mapping that matches on field names or suffixes with [[metaxy.FieldsMapping.identity]] (the default one)
• specific mapping with [[metaxy.FieldsMapping.default]]
• all mapping with [[metaxy.FieldsMapping.all]]

Always use these classmethods to create instances of lineage relationships. Under the hood, they use Pydantic's discriminated union to ensure that the correct type is constructed based on the provided data.

---

metaxy.models.fields_mapping.FieldsMapping pydantic-model

Bases: BaseModel

Base class for field mapping configurations.

Field mappings define how a field automatically resolves its dependencies based on upstream feature fields. This is separate from explicit field dependencies which are defined directly.

Show JSON schema:

{
  "$defs": {
    "AllFieldsMapping": {
      "description": "Field mapping that explicitly depends on all upstream fields.",
      "properties": {
        "type": {
          "const": "all",
          "default": "all",
          "title": "Type",
          "type": "string"
        }
      },
      "title": "AllFieldsMapping",
      "type": "object"
    },
    "DefaultFieldsMapping": {
      "description": "Default automatic field mapping configuration.\n\nWhen used, automatically maps fields to matching upstream fields based on field keys.\n\nAttributes:\n    match_suffix: If True, allows suffix matching (e.g., \"french\" matches \"audio/french\")\n    exclude_fields: List of field keys to exclude from auto-mapping",
      "properties": {
        "type": {
          "const": "default",
          "default": "default",
          "title": "Type",
          "type": "string"
        },
        "match_suffix": {
          "default": false,
          "title": "Match Suffix",
          "type": "boolean"
        },
        "exclude_fields": {
          "items": {
            "$ref": "#/$defs/FieldKey"
          },
          "title": "Exclude Fields",
          "type": "array"
        }
      },
      "title": "DefaultFieldsMapping",
      "type": "object"
    },
    "FieldKey": {
      "description": "Field key as a sequence of string parts.\n\nHashable for use as dict keys in registries.\nParts cannot contain forward slashes (/) or double underscores (__).\n\nExample:\n\n    ```py\n    FieldKey(\"a/b/c\")  # String format\n    # FieldKey(parts=['a', 'b', 'c'])\n\n    FieldKey([\"a\", \"b\", \"c\"])  # List format\n    # FieldKey(parts=['a', 'b', 'c'])\n\n    FieldKey(FieldKey([\"a\", \"b\", \"c\"]))  # FieldKey copy\n    # FieldKey(parts=['a', 'b', 'c'])\n    ```",
      "items": {
        "type": "string"
      },
      "title": "FieldKey",
      "type": "array"
    },
    "NoneFieldsMapping": {
      "description": "Field mapping that never matches any upstream fields.",
      "properties": {
        "type": {
          "const": "none",
          "default": "none",
          "title": "Type",
          "type": "string"
        }
      },
      "title": "NoneFieldsMapping",
      "type": "object"
    },
    "SpecificFieldsMapping": {
      "description": "Field mapping that explicitly depends on specific upstream fields.",
      "properties": {
        "type": {
          "const": "specific",
          "default": "specific",
          "title": "Type",
          "type": "string"
        },
        "mapping": {
          "additionalProperties": {
            "items": {
              "$ref": "#/$defs/FieldKey"
            },
            "type": "array",
            "uniqueItems": true
          },
          "propertyNames": {
            "$ref": "#/$defs/FieldKey"
          },
          "title": "Mapping",
          "type": "object"
        }
      },
      "required": [
        "mapping"
      ],
      "title": "SpecificFieldsMapping",
      "type": "object"
    }
  },
  "description": "Base class for field mapping configurations.\n\nField mappings define how a field automatically resolves its dependencies\nbased on upstream feature fields. This is separate from explicit field\ndependencies which are defined directly.",
  "properties": {
    "mapping": {
      "discriminator": {
        "mapping": {
          "all": "#/$defs/AllFieldsMapping",
          "default": "#/$defs/DefaultFieldsMapping",
          "none": "#/$defs/NoneFieldsMapping",
          "specific": "#/$defs/SpecificFieldsMapping"
        },
        "propertyName": "type"
      },
      "oneOf": [
        {
          "$ref": "#/$defs/AllFieldsMapping"
        },
        {
          "$ref": "#/$defs/SpecificFieldsMapping"
        },
        {
          "$ref": "#/$defs/NoneFieldsMapping"
        },
        {
          "$ref": "#/$defs/DefaultFieldsMapping"
        }
      ],
      "title": "Mapping"
    }
  },
  "required": [
    "mapping"
  ],
  "title": "FieldsMapping",
  "type": "object"
}

Config:

• frozen: True

Fields:

• mapping (AllFieldsMapping | SpecificFieldsMapping | NoneFieldsMapping | DefaultFieldsMapping)

Functions

metaxy.models.fields_mapping.FieldsMapping.resolve_field_deps

resolve_field_deps(context: FieldsMappingResolutionContext) -> set[FieldKey]

Resolve field dependencies based on upstream feature fields.

Invokes the provided mapping to resolve dependencies.

Parameters:

• context (FieldsMappingResolutionContext) –

  The resolution context containing field key and upstream feature.

Returns:

• set[FieldKey] –

  Set of FieldKey instances for matching fields

Source code in src/metaxy/models/fields_mapping.py

def resolve_field_deps(
    self,
    context: FieldsMappingResolutionContext,
) -> set[FieldKey]:
    """Resolve field dependencies based on upstream feature fields.

    Invokes the provided mapping to resolve dependencies.

    Args:
        context: The resolution context containing field key and upstream feature.

    Returns:
        Set of [FieldKey][metaxy.models.types.FieldKey] instances for matching fields
    """
    return self.mapping.resolve_field_deps(context)

metaxy.models.fields_mapping.FieldsMapping.default classmethod

default(*, match_suffix: bool = False, exclude_fields: list[FieldKey] | None = None) -> Self

Create a default field mapping configuration.

Parameters:

• match_suffix (bool, default: False ) –

  If True, allows suffix matching (e.g., "french" matches "audio/french")

• exclude_fields (list[FieldKey] | None, default: None ) –

  List of field keys to exclude from auto-mapping

Returns:

• Self –

  Configured FieldsMapping instance.

Source code in src/metaxy/models/fields_mapping.py

@classmethod
def default(
    cls,
    *,
    match_suffix: bool = False,
    exclude_fields: list[FieldKey] | None = None,
) -> Self:
    """Create a default field mapping configuration.

    Args:
        match_suffix: If True, allows suffix matching (e.g., "french" matches "audio/french")
        exclude_fields: List of field keys to exclude from auto-mapping

    Returns:
        Configured FieldsMapping instance.
    """
    return cls(
        mapping=DefaultFieldsMapping(
            match_suffix=match_suffix,
            exclude_fields=exclude_fields or [],
        )
    )

metaxy.models.fields_mapping.FieldsMapping.specific classmethod

specific(mapping: dict[CoercibleToFieldKey, set[CoercibleToFieldKey]]) -> Self

Create a field mapping that maps downstream field keys into specific upstream field keys.

Parameters:

• mapping (dict[CoercibleToFieldKey, set[CoercibleToFieldKey]]) –

  Mapping of downstream field keys to sets of upstream field keys. Keys and values can be strings, sequences, or FieldKey instances.

Returns:

• Self –

  Configured FieldsMapping instance.

Source code in src/metaxy/models/fields_mapping.py

@classmethod
def specific(
    cls, mapping: dict[CoercibleToFieldKey, set[CoercibleToFieldKey]]
) -> Self:
    """Create a field mapping that maps downstream field keys into specific upstream field keys.

    Args:
        mapping: Mapping of downstream field keys to sets of upstream field keys.
            Keys and values can be strings, sequences, or FieldKey instances.

    Returns:
        Configured FieldsMapping instance.
    """
    # Validate and coerce the mapping keys and values
    validated_mapping: dict[FieldKey, set[FieldKey]] = {}
    for key, value_set in mapping.items():
        validated_key = ValidatedFieldKeyAdapter.validate_python(key)
        validated_values = {
            ValidatedFieldKeyAdapter.validate_python(v) for v in value_set
        }
        validated_mapping[validated_key] = validated_values

    return cls(mapping=SpecificFieldsMapping(mapping=validated_mapping))

metaxy.models.fields_mapping.FieldsMapping.all classmethod

all() -> Self

Create a field mapping that explicitly depends on all upstream fields.

Returns:

• Self –

  Configured FieldsMapping instance.

Examples:

>>> # Use in field specifications
>>> FieldSpec(
...     key="combined",
...     fields_mapping=FieldsMapping.all()
... )

Source code in src/metaxy/models/fields_mapping.py

@classmethod
def all(cls) -> Self:
    """Create a field mapping that explicitly depends on all upstream fields.

    Returns:
        Configured FieldsMapping instance.

    Examples:
        >>> # Use in field specifications
        >>> FieldSpec(
        ...     key="combined",
        ...     fields_mapping=FieldsMapping.all()
        ... )
    """
    return cls(mapping=AllFieldsMapping())

metaxy.models.fields_mapping.FieldsMapping.none classmethod

none() -> Self

Create a field mapping that explicitly depends on no upstream fields.

This is typically useful when explicitly defining FieldSpec.deps instead.

Returns:

• Self –

  Configured FieldsMapping instance.

Source code in src/metaxy/models/fields_mapping.py

@classmethod
def none(cls) -> Self:
    """Create a field mapping that explicitly depends on no upstream fields.

    This is typically useful when explicitly defining [FieldSpec.deps][metaxy.models.field.FieldSpec] instead.

    Returns:
        Configured FieldsMapping instance.
    """
    return cls(mapping=NoneFieldsMapping())

---

metaxy.models.fields_mapping.FieldsMappingType

Bases: str, Enum

Type of fields mapping between a field key and the upstream field keys.

---

metaxy.models.fields_mapping.DefaultFieldsMapping pydantic-model

Bases: BaseFieldsMapping

Default automatic field mapping configuration.

When used, automatically maps fields to matching upstream fields based on field keys.

Attributes:

• match_suffix (bool) –

  If True, allows suffix matching (e.g., "french" matches "audio/french")

• exclude_fields (list[FieldKey]) –

  List of field keys to exclude from auto-mapping

Show JSON schema:

{
  "$defs": {
    "FieldKey": {
      "description": "Field key as a sequence of string parts.\n\nHashable for use as dict keys in registries.\nParts cannot contain forward slashes (/) or double underscores (__).\n\nExample:\n\n    ```py\n    FieldKey(\"a/b/c\")  # String format\n    # FieldKey(parts=['a', 'b', 'c'])\n\n    FieldKey([\"a\", \"b\", \"c\"])  # List format\n    # FieldKey(parts=['a', 'b', 'c'])\n\n    FieldKey(FieldKey([\"a\", \"b\", \"c\"]))  # FieldKey copy\n    # FieldKey(parts=['a', 'b', 'c'])\n    ```",
      "items": {
        "type": "string"
      },
      "title": "FieldKey",
      "type": "array"
    }
  },
  "description": "Default automatic field mapping configuration.\n\nWhen used, automatically maps fields to matching upstream fields based on field keys.\n\nAttributes:\n    match_suffix: If True, allows suffix matching (e.g., \"french\" matches \"audio/french\")\n    exclude_fields: List of field keys to exclude from auto-mapping",
  "properties": {
    "type": {
      "const": "default",
      "default": "default",
      "title": "Type",
      "type": "string"
    },
    "match_suffix": {
      "default": false,
      "title": "Match Suffix",
      "type": "boolean"
    },
    "exclude_fields": {
      "items": {
        "$ref": "#/$defs/FieldKey"
      },
      "title": "Exclude Fields",
      "type": "array"
    }
  },
  "title": "DefaultFieldsMapping",
  "type": "object"
}

Fields:

• type (Literal[DEFAULT])
• match_suffix (bool)
• exclude_fields (list[FieldKey])

Functions

metaxy.models.fields_mapping.DefaultFieldsMapping.resolve_field_deps

resolve_field_deps(context: FieldsMappingResolutionContext) -> set[FieldKey]

Resolve automatic field mapping to explicit FieldDep list.

This method should be overridden by concrete implementations.

Parameters:

• context (FieldsMappingResolutionContext) –

  The resolution context containing field key and upstream feature.

Returns:

• set[FieldKey] –

  Set of FieldKey instances for matching fields

Source code in src/metaxy/models/fields_mapping.py

def resolve_field_deps(
    self,
    context: FieldsMappingResolutionContext,
) -> set[FieldKey]:
    res = set()

    for upstream_field_key in context.upstream_feature_fields:
        # Skip excluded fields
        if upstream_field_key in self.exclude_fields:
            continue

        # Check for exact match
        if upstream_field_key == context.field_key:
            res.add(upstream_field_key)
        # Check for suffix match if enabled
        elif self.match_suffix and self._is_suffix_match(
            context.field_key, upstream_field_key
        ):
            res.add(upstream_field_key)

    # If no fields matched, return ALL fields from this upstream feature
    # (excluding any explicitly excluded fields)
    if not res:
        for upstream_field_key in context.upstream_feature_fields:
            if upstream_field_key not in self.exclude_fields:
                res.add(upstream_field_key)

    return res

metaxy.models.fields_mapping.SpecificFieldsMapping pydantic-model

Bases: BaseFieldsMapping

Field mapping that explicitly depends on specific upstream fields.

Show JSON schema:

{
  "$defs": {
    "FieldKey": {
      "description": "Field key as a sequence of string parts.\n\nHashable for use as dict keys in registries.\nParts cannot contain forward slashes (/) or double underscores (__).\n\nExample:\n\n    ```py\n    FieldKey(\"a/b/c\")  # String format\n    # FieldKey(parts=['a', 'b', 'c'])\n\n    FieldKey([\"a\", \"b\", \"c\"])  # List format\n    # FieldKey(parts=['a', 'b', 'c'])\n\n    FieldKey(FieldKey([\"a\", \"b\", \"c\"]))  # FieldKey copy\n    # FieldKey(parts=['a', 'b', 'c'])\n    ```",
      "items": {
        "type": "string"
      },
      "title": "FieldKey",
      "type": "array"
    }
  },
  "description": "Field mapping that explicitly depends on specific upstream fields.",
  "properties": {
    "type": {
      "const": "specific",
      "default": "specific",
      "title": "Type",
      "type": "string"
    },
    "mapping": {
      "additionalProperties": {
        "items": {
          "$ref": "#/$defs/FieldKey"
        },
        "type": "array",
        "uniqueItems": true
      },
      "propertyNames": {
        "$ref": "#/$defs/FieldKey"
      },
      "title": "Mapping",
      "type": "object"
    }
  },
  "required": [
    "mapping"
  ],
  "title": "SpecificFieldsMapping",
  "type": "object"
}

Fields:

• type (Literal[SPECIFIC])
• mapping (dict[FieldKey, set[FieldKey]])

Functions

metaxy.models.fields_mapping.SpecificFieldsMapping.resolve_field_deps

resolve_field_deps(context: FieldsMappingResolutionContext) -> set[FieldKey]

Resolve automatic field mapping to explicit FieldDep list.

This method should be overridden by concrete implementations.

Parameters:

• context (FieldsMappingResolutionContext) –

  The resolution context containing field key and upstream feature.

Returns:

• set[FieldKey] –

  Set of FieldKey instances for matching fields

Source code in src/metaxy/models/fields_mapping.py

def resolve_field_deps(
    self,
    context: FieldsMappingResolutionContext,
) -> set[FieldKey]:
    desired_upstream_fields = self.mapping.get(context.field_key, set())
    return desired_upstream_fields & context.upstream_feature_fields

metaxy.models.fields_mapping.AllFieldsMapping pydantic-model

Bases: BaseFieldsMapping

Field mapping that explicitly depends on all upstream fields.

Show JSON schema:

{
  "description": "Field mapping that explicitly depends on all upstream fields.",
  "properties": {
    "type": {
      "const": "all",
      "default": "all",
      "title": "Type",
      "type": "string"
    }
  },
  "title": "AllFieldsMapping",
  "type": "object"
}

Fields:

• type (Literal[ALL])

Functions

metaxy.models.fields_mapping.AllFieldsMapping.resolve_field_deps

resolve_field_deps(context: FieldsMappingResolutionContext) -> set[FieldKey]

Resolve automatic field mapping to explicit FieldDep list.

This method should be overridden by concrete implementations.

Parameters:

• context (FieldsMappingResolutionContext) –

  The resolution context containing field key and upstream feature.

Returns:

• set[FieldKey] –

  Set of FieldKey instances for matching fields

Source code in src/metaxy/models/fields_mapping.py

def resolve_field_deps(
    self,
    context: FieldsMappingResolutionContext,
) -> set[FieldKey]:
    return context.upstream_feature_fields

metaxy.models.fields_mapping.NoneFieldsMapping pydantic-model

Bases: BaseFieldsMapping

Field mapping that never matches any upstream fields.

Show JSON schema:

{
  "description": "Field mapping that never matches any upstream fields.",
  "properties": {
    "type": {
      "const": "none",
      "default": "none",
      "title": "Type",
      "type": "string"
    }
  },
  "title": "NoneFieldsMapping",
  "type": "object"
}

Fields:

• type (Literal[NONE])

Functions

metaxy.models.fields_mapping.NoneFieldsMapping.resolve_field_deps

resolve_field_deps(context: FieldsMappingResolutionContext) -> set[FieldKey]

Resolve automatic field mapping to explicit FieldDep list.

This method should be overridden by concrete implementations.

Parameters:

• context (FieldsMappingResolutionContext) –

  The resolution context containing field key and upstream feature.

Returns:

• set[FieldKey] –

  Set of FieldKey instances for matching fields

Source code in src/metaxy/models/fields_mapping.py

def resolve_field_deps(
    self,
    context: FieldsMappingResolutionContext,
) -> set[FieldKey]:
    return set()

---