FHIR Mapping Language

FHIR Mapping Language parser and transformation utilities.

ArbitraryModel

Path: fhircraft.fhir.mapper.engine.core.ArbitraryModel

Bases: BaseModel

A dynamic Pydantic model that accepts arbitrary fields.

This is used for arbitrary target structures in mappings where no specific structure definition is provided. Unlike plain dicts, this model is compatible with the FHIRPath engine's update mechanisms, allowing complex nested path creation and array operations.

FHIRMappingEngine

Path: fhircraft.fhir.mapper.engine.core.FHIRMappingEngine

FHIRMappingEngine(repository: CompositeStructureDefinitionRepository | None = None, factory: ResourceFactory | None = None)

FHIRMappingEngine is responsible for executing FHIR StructureMap-based transformations between FHIR resources.

This engine validates, processes, and applies mapping rules defined in a StructureMap to transform source FHIR resources into target resources, supporting complex mapping logic, rule dependencies, and FHIRPath-based expressions.

Attributes:

Name	Type	Description
repository	CompositeStructureDefinitionRepository	Repository for FHIR StructureDefinitions.
factory	ResourceFactory	Factory for constructing FHIR resource models.
transformer	MappingTransformer	Executes FHIRPath-based transforms.

Methods:

Name	Description
execute	Executes a FHIR StructureMap transformation using the provided sources and optional targets.
process_group	Processes a StructureMap group by validating input parameters, constructing a local mapping scope,
process_rule	Processes a single StructureMap rule within the given mapping scope.
process_source	Processes a StructureMapGroupRuleSource object within a given MappingScope and returns the variable name
process_target	Processes a StructureMapGroupRuleTarget within the given mapping scope.
validate_structure_map	Validates the structure and content of a given StructureMap instance.

Source code in fhircraft/fhir/mapper/engine/core.py

def __init__(
    self,
    repository: CompositeStructureDefinitionRepository | None = None,
    factory: ResourceFactory | None = None,
):
    self.repository = repository or CompositeStructureDefinitionRepository()
    self.factory = factory or ResourceFactory(repository=self.repository)
    self.transformer = MappingTransformer()

execute

execute(structure_map: StructureMap, sources: tuple[BaseModel | dict], targets: tuple[BaseModel | dict] | None = None, group: str | None = None) -> tuple[BaseModel, ...]

Executes a FHIR StructureMap transformation using the provided sources and optional targets.

This method resolves structure definitions, validates input data, sets up the mapping scope, binds source and target instances to group parameters, and processes the entrypoint group to produce the mapped target instances.

Parameters:

Name	Type	Description	Default
structure_map	StructureMap	The StructureMap resource defining the transformation rules.	required
sources	tuple[BaseModel | dict]	Source data to be mapped, as a tuple of Pydantic models or dictionaries.	required
targets	tuple[BaseModel | dict] | None	Optional target instances to populate. If not provided, new instances are created as needed.	None
group	str | None	The name of the entrypoint group to execute. If not specified, the first group is used.	None

Returns:

Type	Description
tuple[BaseModel, ...]	tuple[BaseModel, ...]: A tuple of resulting target instances after the transformation.

Raises:

Type	Description
NotImplementedError	If StructureMap imports are present (not supported).
ValueError	If a constant in the StructureMap is missing a name or conflicts with a model name.
RuntimeError	If the number of provided sources or targets does not match the group parameters, or if required targets are missing.
TypeError	If provided sources or targets do not match the expected types for the group parameters.

Source code in fhircraft/fhir/mapper/engine/core.py

def execute(
    self,
    structure_map: StructureMap,
    sources: tuple[BaseModel | dict],
    targets: tuple[BaseModel | dict] | None = None,
    group: str | None = None,
) -> tuple[BaseModel, ...]:
    """
    Executes a FHIR StructureMap transformation using the provided sources and optional targets.

    This method resolves structure definitions, validates input data, sets up the mapping scope,
    binds source and target instances to group parameters, and processes the entrypoint group
    to produce the mapped target instances.

    Args:
        structure_map (StructureMap): The StructureMap resource defining the transformation rules.
        sources (tuple[BaseModel | dict]): Source data to be mapped, as a tuple of Pydantic models or dictionaries.
        targets (tuple[BaseModel | dict] | None, optional): Optional target instances to populate. If not provided, new instances are created as needed.
        group (str | None, optional): The name of the entrypoint group to execute. If not specified, the first group is used.

    Returns:
        tuple[BaseModel, ...]: A tuple of resulting target instances after the transformation.

    Raises:
        NotImplementedError: If StructureMap imports are present (not supported).
        ValueError: If a constant in the StructureMap is missing a name or conflicts with a model name.
        RuntimeError: If the number of provided sources or targets does not match the group parameters, or if required targets are missing.
        TypeError: If provided sources or targets do not match the expected types for the group parameters.
    """

    # Ensure sources is a tuple
    if not isinstance(sources, tuple):
        sources = (sources,)

    if structure_map.import_:
        raise NotImplementedError("StructureMap imports are not implemented yet")

    # Resolve structure definitions
    source_models = self._resolve_structure_definitions(
        structure_map, StructureMapModelMode.SOURCE
    )
    target_models = self._resolve_structure_definitions(
        structure_map, StructureMapModelMode.TARGET
    )
    queried_models = self._resolve_structure_definitions(
        structure_map, StructureMapModelMode.QUERIED
    )
    produced_models = self._resolve_structure_definitions(
        structure_map, StructureMapModelMode.PRODUCED
    )

    # Validate source data
    validated_sources = self._validate_source_data(sources, source_models)

    # Create the global mapping scope
    global_scope = MappingScope(
        name="global",
        types={
            **source_models,
            **target_models,
            **queried_models,
            **produced_models,
        },
        groups=OrderedDict(
            [(group.name, group) for group in structure_map.group or []]
        ),
        concept_maps={
            map.name: map
            for map in (structure_map.contained or [])
            if map.resourceType == "ConceptMap" and map.name
        },
    )

    # Build default mapping group registry
    self._build_default_group_registry(structure_map, global_scope)

    # Parse and validate constants
    for const in structure_map.const or []:
        if not const.name:
            raise ValueError("Constant must have a name")
        if const.name in source_models or const.name in target_models:
            raise ValueError(
                f"Constant name '{const.name}' conflicts with existing source or target model"
            )
        # Add the constant as a variable in the global scope
        global_scope.define_variable(const.name, fhirpath_parser.parse(const.value))

    # Determine the entrypoint group
    target_group = (global_scope.groups.get(group) if group else None) or list(
        global_scope.groups.values()
    )[0]

    # Validate the group parameters
    expected_sources = len(
        [
            input
            for input in (target_group.input or [])
            if input.mode == StructureMapModelMode.SOURCE
        ]
    )
    if len(validated_sources) != expected_sources:
        raise RuntimeError(
            f"Entrypoint group {target_group.name} expected {expected_sources} sources, got {len(validated_sources)}."
        )

    # Validate targets if provided
    if targets:
        expected_targets = len(
            [
                input
                for input in (target_group.input or [])
                if input.mode
                in (StructureMapModelMode.TARGET, StructureMapModelMode.PRODUCED)
            ]
        )
        if len(targets) != expected_targets:
            raise RuntimeError(
                f"Entrypoint group {target_group.name} expected {expected_targets} targets, got {len(targets)}."
            )

    # Bind source and target instances to group parameters
    parameters = []
    for input in target_group.input:
        if input.mode == StructureMapModelMode.SOURCE:
            if input.type:
                # Explicit type specified - match by type
                source_instance = validated_sources.get(input.type)
                if not source_instance:
                    raise TypeError(
                        f"Invalid source provided. None of the source arguments matches the '{input.name}' parameter of type {input.type} for the entrypoint group '{target_group.name}'."
                    )
            else:
                # No type specified - use first available source or match by parameter name
                source_instance = (
                    validated_sources.get(input.name) or 
                    validated_sources.get("source") or
                    next(iter(validated_sources.values()), None)
                )
                if source_instance is None:
                    raise TypeError(
                        f"No source data available for parameter '{input.name}'."
                    )
            source_instance_id = f"source_{id(source_instance)}"
            global_scope.source_instances[source_instance_id] = source_instance  # type: ignore
            parameters.append(fhirpath.Element(source_instance_id))

        if input.mode == StructureMapModelMode.TARGET:
            target_type = global_scope.types.get(input.type) if input.type else None

            if target_type is not None:
                # Type specified and model available - create or find typed instance
                if not targets:
                    target_instance = target_type.model_construct()
                else:
                    target_instance = next(
                        (
                            target
                            for target in targets
                            if isinstance(target, target_type)
                        ),
                        None,
                    )
                    if not target_instance:
                        raise TypeError(
                            f"Invalid target provided. None of the target arguments matches the {input.name} parameters of type {input.type} for the entrypoint group '{target_group.name}'."
                        )
            else:
                # No type or type not resolved - use arbitrary target
                if targets:
                    target_instance = targets[0]
                else:
                    # Create ArbitraryModel instance for arbitrary target
                    # This allows FHIRPath engine to work properly with nested paths
                    target_instance = ArbitraryModel()

            target_instance_id = f"source_{id(target_instance)}"
            global_scope.target_instances[target_instance_id] = target_instance  # type: ignore
            parameters.append(fhirpath.Element(target_instance_id))

    # Process the entrypoint group
    self.process_group(target_group, parameters, global_scope)

    # Return the resulting target instances
    return tuple(
        [
            # Convert ArbitraryModel to dict for user consumption
            instance.model_dump() if isinstance(instance, ArbitraryModel)
            # Validate other BaseModel instances
            else instance.model_validate(instance.model_dump()) if isinstance(instance, BaseModel)
            # Pass through non-BaseModel instances (shouldn't happen)
            else instance
            for instance in global_scope.target_instances.values()
        ]
    )

process_group

process_group(group: StructureMapGroup, parameters: list[FHIRPath] | tuple[FHIRPath], scope: MappingScope)

Processes a StructureMap group by validating input parameters, constructing a local mapping scope, and executing the group's rules in the correct order, handling special list modes ('first' and 'last').

Parameters:

Name	Type	Description	Default
group	StructureMapGroup	The group definition containing mapping rules and input definitions.	required
parameters	list[FHIRPath] | tuple[FHIRPath]	The input parameters to be mapped, corresponding to the group's input definitions.	required
scope	MappingScope	The parent mapping scope to use as the basis for the group's local scope.	required

Raises:

Type	Description
MappingError	If the number of provided parameters does not match the group's input definitions.
RuntimeError	If more than one rule with 'first' or 'last' target list mode is found in the group.
NotImplementedError	If a target list mode other than 'first' or 'last' is encountered.

Source code in fhircraft/fhir/mapper/engine/core.py

def process_group(
    self,
    group: StructureMapGroup,
    parameters: list[FHIRPath] | tuple[FHIRPath],
    scope: MappingScope,
):
    """
    Processes a StructureMap group by validating input parameters, constructing a local mapping scope,
    and executing the group's rules in the correct order, handling special list modes ('first' and 'last').

    Args:
        group (StructureMapGroup): The group definition containing mapping rules and input definitions.
        parameters (list[FHIRPath] | tuple[FHIRPath]): The input parameters to be mapped, corresponding to the group's input definitions.
        scope (MappingScope): The parent mapping scope to use as the basis for the group's local scope.

    Raises:
        MappingError: If the number of provided parameters does not match the group's input definitions.
        RuntimeError: If more than one rule with 'first' or 'last' target list mode is found in the group.
        NotImplementedError: If a target list mode other than 'first' or 'last' is encountered.

    """
    group_name = group.name or f"group_{id(group)}"

    # Construct local group scope
    group_scope = MappingScope(
        name=group_name,
        parent=scope,
    )

    # Validate input parameters
    if len(group.input) != len(parameters):
        raise MappingError(
            f"Invalid number of parameters provided for group '{group_name}'. Expected {len(group.input)}, got {len(parameters)}."
        )
    for input, parameter in zip(group.input, parameters):
        group_scope.define_variable(input.name, parameter)

    # Process rules in order, handling 'first' and 'last' list modes
    rules = []
    first_rule, last_rule = None, None
    for rule in group.rule or []:
        if rule.target and (
            targetMode := next(
                (target.listMode for target in rule.target if target.listMode),
                [None],
            )[0]
        ):
            if targetMode == StructureMapTargetListMode.FIRST:
                if first_rule:
                    raise RuntimeError(
                        'Only one rule with "first" target list mode is allowed per group.'
                    )
                first_rule = rule
            elif targetMode == StructureMapTargetListMode.LAST:
                if last_rule:
                    raise RuntimeError(
                        'Only one rule with "last" target list mode is allowed per group.'
                    )
                last_rule = rule
            else:
                raise NotImplementedError(
                    "Only 'first' and 'last' target list modes are implemented so far. Mode '{}' not implemented"
                )
        else:
            rules.append(rule)
    rules = (
        ([first_rule] if first_rule else [])
        + rules
        + ([last_rule] if last_rule else [])
    )

    # Process each rule
    for rule in rules:
        self.process_rule(rule, group_scope)

process_rule

process_rule(rule: StructureMapGroupRule, scope: MappingScope) -> MappingScope

Processes a single StructureMap rule within the given mapping scope.

This method handles the evaluation and execution of a StructureMapGroupRule, including: - Cycle detection to prevent infinite recursion. - Source processing to determine iteration counts and validate type, condition, and check constraints. - Iterative processing for each source instance, including: - Creating an iteration-specific mapping scope. - Setting indexed FHIRPath variables for the current iteration. - Processing target mappings, dependent rules/groups, and nested rules. - Merging results from each iteration back into the main scope.

Parameters:

Name	Type	Description	Default
rule	StructureMapGroupRule	The rule to process.	required
scope	MappingScope	The current mapping scope.	required

Returns:

Name	Type	Description
MappingScope	MappingScope	The updated mapping scope after processing the rule.

Raises:

Type	Description
RuleProcessingError	If any rule constraints (such as cardinality, type, or checks) are violated, or if required variables or dependent groups are not found.

Source code in fhircraft/fhir/mapper/engine/core.py

def process_rule(
    self, rule: StructureMapGroupRule, scope: MappingScope
) -> MappingScope:
    """
    Processes a single StructureMap rule within the given mapping scope.

    This method handles the evaluation and execution of a StructureMapGroupRule, including:
    - Cycle detection to prevent infinite recursion.
    - Source processing to determine iteration counts and validate type, condition, and check constraints.
    - Iterative processing for each source instance, including:
        - Creating an iteration-specific mapping scope.
        - Setting indexed FHIRPath variables for the current iteration.
        - Processing target mappings, dependent rules/groups, and nested rules.
        - Merging results from each iteration back into the main scope.

    Args:
        rule (StructureMapGroupRule): The rule to process.
        scope (MappingScope): The current mapping scope.

    Returns:
        MappingScope: The updated mapping scope after processing the rule.

    Raises:
        RuleProcessingError: If any rule constraints (such as cardinality, type, or checks) are violated,
            or if required variables or dependent groups are not found.
    """
    rule_name = rule.name or f"rule_{id(rule)}"

    # Check for cycles
    if scope.is_processing_rule(rule_name):
        logger.warning(f"Cycle detected in rule {rule_name}, skipping")
        return scope
    scope.start_processing_rule(rule_name)

    try:
        logger.debug(f"Processing rule: {rule_name}")

        # Process sources first to determine iteration
        source_iterations = {}

        for source in rule.source or []:
            var_name = self.process_source(source, scope)
            source_fhirpath = scope.resolve_fhirpath(var_name)

            if source.type:
                condition_fhirpath = source_fhirpath._invoke(
                    getattr(fhirpath, f"Is{source.type.title()}")
                )
                if not bool(condition_fhirpath.single(scope.get_instances())):
                    logger.debug(
                        f"Source type condition not met for rule {rule_name}"
                    )
                    return scope

            # Where condition
            if source.condition:
                condition_fhirpath = fhirpath_parser.parse(source.condition)
                condition_fhirpath = self._replace_mapping_scope_elements(
                    condition_fhirpath, scope
                )

                if not bool(condition_fhirpath.single(scope.get_instances())):
                    logger.debug(f"Source condition not met for rule {rule_name}")
                    return scope

            # Check condition
            if source.check:
                condition_fhirpath = fhirpath_parser.parse(source.check)
                condition_fhirpath = self._replace_mapping_scope_elements(
                    condition_fhirpath, scope
                )

                if not bool(condition_fhirpath.single(scope.get_instances())):
                    raise RuleProcessingError(
                        f"Source check failed for rule {rule_name}"
                    )

            # Collect source values for iteration
            if source_fhirpath is None:
                raise RuleProcessingError(f"Source variable {var_name} not found")
            source_iterations[var_name] = source_fhirpath.count(
                scope.get_instances()
            )
            if source.min is not None and source_iterations[var_name] < source.min:
                raise RuleProcessingError(
                    f"Source minimum cardinality not met for rule {rule_name}"
                )
            if (
                source.max is not None
                and source.max != "*"
                and source_iterations[var_name] > int(source.max)
            ):
                raise RuleProcessingError(
                    f"Source maximum cardinality exceeded for rule {rule_name}"
                )

        for source_var, iterations in source_iterations.items():
            for source_iteration in range(iterations):

                logger.debug(
                    f"Processing iteration {source_iteration} for rule {rule_name}"
                )
                # Create iteration scope
                iteration_scope = MappingScope(
                    name=f"{scope.name}_iter_{source_iteration}",
                    source_instances=scope.source_instances.copy(),
                    target_instances=scope.target_instances.copy(),
                    types=scope.types.copy(),
                    variables=scope.variables.copy(),  # Copy existing variables
                    parent=scope.parent,
                )

                # Set the source variable to an indexed FHIRPath
                if (rule_source := scope.resolve_fhirpath(source_var)) is None:
                    raise RuleProcessingError(
                        f"Source variable {source_var} not found"
                    )
                iteration_scope.define_variable(
                    source_var,
                    rule_source._invoke(fhirpath.Index(source_iteration)),
                )

                # Process targets for this iteration
                for target in rule.target or []:
                    self.process_target(target, iteration_scope)

                # Process dependent rules for this iteration
                for dependent in rule.dependent or []:
                    dependent_group = iteration_scope.resolve_symbol(dependent.name)
                    if not dependent_group:
                        raise RuleProcessingError(
                            f"Dependent group or rule '{dependent.name}' not found"
                        )
                    if not isinstance(dependent_group, StructureMapGroup):
                        raise RuleProcessingError(
                            f"Dependent '{dependent.name}' is not a group"
                        )
                    parameters = [
                        iteration_scope.resolve_fhirpath(param.value)
                        for param in dependent.parameter
                    ]
                    self.process_group(dependent_group, parameters, iteration_scope)

                # Process nested rules for this iteration
                for nested_rule in rule.rule or []:
                    self.process_rule(nested_rule, iteration_scope)

                # Merge back iteration results to main scope
                scope.target_instances.update(iteration_scope.target_instances)

    finally:
        scope.finish_processing_rule(rule_name)
    return scope

process_source

process_source(source: StructureMapGroupRuleSource, scope: MappingScope) -> str

Processes a StructureMapGroupRuleSource object within a given MappingScope and returns the variable name associated with the resolved FHIRPath expression.

This method resolves the FHIRPath context from the source, applies any specified element path, and modifies the path according to the listMode option (e.g., first, last, not_first, not_last, only_one). The resulting FHIRPath expression is stored in the scope under a variable name, which is either provided by the source or generated uniquely.

Parameters:

Name	Type	Description	Default
source	StructureMapGroupRuleSource	The source mapping definition containing context, element, listMode, and variable.	required
scope	MappingScope	The current mapping scope used to resolve FHIRPath and store variables.	required

Returns:

Name	Type	Description
str	str	The variable name under which the resolved FHIRPath expression is stored in the scope.

Source code in fhircraft/fhir/mapper/engine/core.py

def process_source(
    self, source: StructureMapGroupRuleSource, scope: MappingScope
) -> str:
    """
    Processes a StructureMapGroupRuleSource object within a given MappingScope and returns the variable name
    associated with the resolved FHIRPath expression.

    This method resolves the FHIRPath context from the source, applies any specified element path,
    and modifies the path according to the listMode option (e.g., first, last, not_first, not_last, only_one).
    The resulting FHIRPath expression is stored in the scope under a variable name, which is either
    provided by the source or generated uniquely.

    Args:
        source (StructureMapGroupRuleSource): The source mapping definition containing context, element, listMode, and variable.
        scope (MappingScope): The current mapping scope used to resolve FHIRPath and store variables.

    Returns:
        str: The variable name under which the resolved FHIRPath expression is stored in the scope.
    """
    path = scope.resolve_fhirpath(source.context)
    # Apply element path if specified
    if source.element:
        path = path._invoke(fhirpath.Element(source.element))

    # Apply list-option condition if specified
    if source.listMode == "first":
        path = path._invoke(fhirpath.First())
    elif source.listMode == "not_first":
        path = path._invoke(fhirpath.Tail())
    elif source.listMode == "not_last":
        path = path._invoke(
            fhirpath.Exclude(
                path._invoke(fhirpath.Last()).single(scope.get_instances())
            )
        )
    elif source.listMode == "last":
        path = path._invoke(fhirpath.Last())
    elif source.listMode == "only_one":
        path = path._invoke(fhirpath.Single())

    # Store source FHIRPath
    var_name = source.variable or f"source_{id(source)}"
    scope.define_variable(var_name, path)
    return var_name

process_target

process_target(target: StructureMapGroupRuleTarget, scope: MappingScope) -> Any

Processes a StructureMapGroupRuleTarget within the given mapping scope.

This method resolves the FHIRPath context for the target, applies any specified element path, determines the appropriate insertion index, and stores the resulting FHIRPath in the scope as a variable. If a transform is specified on the target, it executes the transform with the provided parameters and updates the target structure with the transformed value.

Parameters:

Name	Type	Description	Default
target	StructureMapGroupRuleTarget	The mapping target to process, containing context, element, variable, transform, and parameters.	required
scope	MappingScope	The current mapping scope, used for resolving FHIRPath contexts and managing variables.	required

Returns:

Name	Type	Description
Any	Any	The result of processing the target, typically the updated FHIRPath or transformed value.

Raises:

Type	Description
RuleProcessingError	If the target context is not specified.

Source code in fhircraft/fhir/mapper/engine/core.py

def process_target(
    self,
    target: StructureMapGroupRuleTarget,
    scope: MappingScope,
) -> Any:
    """
    Processes a StructureMapGroupRuleTarget within the given mapping scope.

    This method resolves the FHIRPath context for the target, applies any specified element path,
    determines the appropriate insertion index, and stores the resulting FHIRPath in the scope as a variable.
    If a transform is specified on the target, it executes the transform with the provided parameters and
    updates the target structure with the transformed value.

    Args:
        target (StructureMapGroupRuleTarget): The mapping target to process, containing context, element, variable,
            transform, and parameters.
        scope (MappingScope): The current mapping scope, used for resolving FHIRPath contexts and managing variables.

    Returns:
        Any: The result of processing the target, typically the updated FHIRPath or transformed value.

    Raises:
        RuleProcessingError: If the target context is not specified.
    """
    if not target.context:
        raise RuleProcessingError("Target context is required")
    path = scope.resolve_fhirpath(target.context)
    # Apply element path if specified
    if target.element:
        path = path._invoke(fhirpath.Element(target.element))

    insert_index = path.count(scope.get_instances())
    path = path._invoke(fhirpath.Index(insert_index))

    # Store target FHIRPath
    var_name = target.variable or f"target_{id(target)}"
    scope.define_variable(var_name, path)

    transform = target.transform
    if transform:
        target.parameter = target.parameter or []
        # Execute the transform
        transformed_value = self.transformer.execute(
            transform, scope, target.parameter
        )
        # Update the target structure
        path.update_single(scope.get_instances(), transformed_value)

validate_structure_map

validate_structure_map(structure_map: StructureMap) -> List[str]

Validates the structure and content of a given StructureMap instance.

This method checks for the presence of required groups and structure declarations, ensures that both source and target structures are defined, and verifies that each group contains rules. It also delegates rule-specific validation to the _validate_rule method.

Parameters:

Name	Type	Description	Default
structure_map	StructureMap	The StructureMap object to validate.	required

Returns:

Type	Description
List[str]	List[str]: A list of validation issue messages. The list is empty if no issues are found.

Source code in fhircraft/fhir/mapper/engine/core.py

def validate_structure_map(self, structure_map: StructureMap) -> List[str]:
    """
    Validates the structure and content of a given StructureMap instance.

    This method checks for the presence of required groups and structure declarations,
    ensures that both source and target structures are defined, and verifies that each
    group contains rules. It also delegates rule-specific validation to the _validate_rule method.

    Args:
        structure_map (StructureMap): The StructureMap object to validate.

    Returns:
        List[str]: A list of validation issue messages. The list is empty if no issues are found.
    """
    issues = []

    # Check basic structure
    if not structure_map.group:
        issues.append("StructureMap has no groups defined")

    if not structure_map.structure:
        issues.append("StructureMap has no structure declarations")

    # Check structure declarations
    source_structures = [
        s for s in structure_map.structure or [] if s.mode == "source"
    ]
    target_structures = [
        s for s in structure_map.structure or [] if s.mode == "target"
    ]

    if not source_structures:
        issues.append("No source structures defined")
    if not target_structures:
        issues.append("No target structures defined")

    # Check groups and rules
    for group in structure_map.group or []:
        if not group.rule:
            issues.append(f"Group {group.name} has no rules")

        for rule in group.rule or []:
            self._validate_rule(rule, issues)

    return issues

StructureMapModelMode

Path: fhircraft.fhir.mapper.engine.core.StructureMapModelMode

Bases: str, Enum

Enumeration of StructureMap model modes.

StructureMapTargetListMode

Path: fhircraft.fhir.mapper.engine.core.StructureMapTargetListMode

Bases: str, Enum

Enumeration of StructureMap model modes.

mapper module-attribute

mapper = FHIRMappingEngine()

MappingError

Path: fhircraft.fhir.mapper.engine.exceptions.MappingError

Bases: Exception

Base exception for mapping engine errors.

RuleProcessingError

Path: fhircraft.fhir.mapper.engine.exceptions.RuleProcessingError

Bases: MappingError

Raised when rule processing fails.

ValidationError

Path: fhircraft.fhir.mapper.engine.exceptions.ValidationError

Bases: MappingError

Raised when input data validation fails.

MappingScope

Path: fhircraft.fhir.mapper.engine.scope.MappingScope

dataclass

MappingScope(name: str, types: Dict[str, type[BaseModel]] = dict(), source_instances: Dict[str, BaseModel] = dict(), target_instances: Dict[str, BaseModel] = dict(), concept_maps: Dict[str, ConceptMap] = dict(), groups: OrderedDict[str, StructureMapGroup] = OrderedDict(), variables: Dict[str, FHIRPath] = dict(), default_groups: Dict[str, StructureMapGroup] = dict(), processing_rules: Set[str] = set(), parent: Optional[MappingScope] = None)

A scope defines the visibility and accessibility of identifiers (variables, types, etc.)

Parameters:

Name	Type	Description	Default
name	str	Name of the scope	required
types	Dict[str, type[BaseModel]]	dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)	<class 'dict'>
source_instances	Dict[str, BaseModel]	dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)	<class 'dict'>
target_instances	Dict[str, BaseModel]	dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)	<class 'dict'>
concept_maps	Dict[str, ConceptMap]	dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)	<class 'dict'>
groups	OrderedDict[str, StructureMapGroup]	Dictionary that remembers insertion order	<class 'collections.OrderedDict'>
variables	Dict[str, FHIRPath]	dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)	<class 'dict'>
default_groups	Dict[str, StructureMapGroup]	dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)	<class 'dict'>
processing_rules	Set[str]	Build an unordered collection of unique elements.	<dynamic>
parent	ForwardRef | None	Parent mapping scope	None

Methods:

Name	Description
define_variable	Defines a new variable in the current scope.
get_instances	Returns a dictionary containing all instances from the current scope, including those inherited from the parent scope (if any), as well as target and source instances.
get_concept_map	Retrieve a ConceptMap by its identifier from the current scope or any parent scopes.
get_target_instance	Retrieve a target instance by its identifier from the current scope or any parent scopes.
get_source_instance	Retrieve a source instance by its identifier from the current scope or any parent scopes.
get_type	Retrieve the type associated with the given identifier from the current scope or its parent scopes.
resolve_symbol	Resolves a symbol (variable, type, or group) by its identifier from the current scope or any parent scopes.
has_symbol	Checks if a symbol with the given identifier exists in the current scope.
has_local_symbol	Checks if a symbol with the given identifier exists in the current scope only.
resolve_fhirpath	Resolve a symbol as a FHIRPath expression.
get_all_visible_symbols	Retrieves all visible symbols in the current scope, including those inherited from parent scopes.
get_scope_path	Returns the hierarchical path of scope names from the root to the current scope as a list of strings.
get_scope_depth	Returns the depth of the current scope within the scope hierarchy.
create_child_scope	Creates and returns a new child MappingScope with the specified name, setting the current scope as its parent.
is_processing_rule	Check if a given rule name is present in the list of processing rules.
start_processing_rule	Marks the beginning of processing for a specific rule by adding its name to the set of currently processing rules.
finish_processing_rule	Marks the specified rule as finished by removing it from the set of currently processing rules.

Attributes:

Name	Type	Description
name	str	Name of the scope
types	Dict[str, type[BaseModel]]	Registry of available FHIR types by identifier
source_instances	Dict[str, BaseModel]	The source instances being mapped
target_instances	Dict[str, BaseModel]	The target instances being mapped
concept_maps	Dict[str, ConceptMap]	Registry of available concept maps for value transformations
groups	OrderedDict[str, StructureMapGroup]	The groups defined on this scope
variables	Dict[str, FHIRPath]	Registry of variables mapped to resolved FHIRPath expressions
default_groups	Dict[str, StructureMapGroup]	Registry of default mapping groups by type signature
processing_rules	Set[str]	Set of currently processing rules
parent	Optional[MappingScope]	Parent mapping scope

name instance-attribute

name: str

Name of the scope

types class-attribute instance-attribute

types: Dict[str, type[BaseModel]] = field(default_factory=dict)

Registry of available FHIR types by identifier

source_instances class-attribute instance-attribute

source_instances: Dict[str, BaseModel] = field(default_factory=dict)

The source instances being mapped

target_instances class-attribute instance-attribute

target_instances: Dict[str, BaseModel] = field(default_factory=dict)

The target instances being mapped

concept_maps class-attribute instance-attribute

concept_maps: Dict[str, ConceptMap] = field(default_factory=dict)

Registry of available concept maps for value transformations

groups class-attribute instance-attribute

groups: OrderedDict[str, StructureMapGroup] = field(default_factory=OrderedDict)

The groups defined on this scope

variables class-attribute instance-attribute

variables: Dict[str, FHIRPath] = field(default_factory=dict)

Registry of variables mapped to resolved FHIRPath expressions

default_groups class-attribute instance-attribute

default_groups: Dict[str, StructureMapGroup] = field(default_factory=dict)

Registry of default mapping groups by type signature

processing_rules class-attribute instance-attribute

processing_rules: Set[str] = field(default_factory=set)

Set of currently processing rules

parent class-attribute instance-attribute

parent: Optional[MappingScope] = None

Parent mapping scope

define_variable

define_variable(identifier: str, value: FHIRPath) -> None

Defines a new variable in the current scope.

Parameters:

Name	Type	Description	Default
identifier	str	The name of the variable to define.	required
value	FHIRPath	The FHIRPath instance to assign to the variable.	required

Raises:

Type	Description
ValueError	If the provided value is not an instance of FHIRPath.

Source code in fhircraft/fhir/mapper/engine/scope.py

def define_variable(self, identifier: str, value: FHIRPath) -> None:
    """
    Defines a new variable in the current scope.

    Args:
        identifier (str): The name of the variable to define.
        value (FHIRPath): The FHIRPath instance to assign to the variable.

    Raises:
        ValueError: If the provided value is not an instance of FHIRPath.
    """
    if not isinstance(value, FHIRPath):
        raise ValueError("Variables can only be assigned to a FHIRPath instance")
    self.variables[identifier] = value

get_instances

get_instances() -> Dict[str, BaseModel]

Returns a dictionary containing all instances from the current scope, including those inherited from the parent scope (if any), as well as target and source instances.

Returns:

Type	Description
Dict[str, BaseModel]	Dict[str, BaseModel]: A dictionary mapping instance names to their corresponding BaseModel objects, aggregated from the parent scope, target instances, and source instances.

Source code in fhircraft/fhir/mapper/engine/scope.py

def get_instances(self) -> Dict[str, BaseModel]:
    """
    Returns a dictionary containing all instances from the current scope, including those inherited from the parent scope (if any), as well as target and source instances.

    Returns:
        Dict[str, BaseModel]: A dictionary mapping instance names to their corresponding BaseModel objects, aggregated from the parent scope, target instances, and source instances.
    """
    return {
        **(self.parent.get_instances() if self.parent else {}),
        **self.target_instances,
        **self.source_instances,
    }

get_concept_map

get_concept_map(identifier: str) -> ConceptMap

Retrieve a ConceptMap by its identifier from the current scope or any parent scopes.

Parameters:

Name	Type	Description	Default
identifier	str	The unique identifier of the ConceptMap to retrieve.	required

Returns:

Name	Type	Description
ConceptMap	ConceptMap	The ConceptMap instance associated with the given identifier.

Raises:

Type	Description
MappingError	If the ConceptMap with the specified identifier is not found in the current or any parent scopes.

Source code in fhircraft/fhir/mapper/engine/scope.py

def get_concept_map(self, identifier: str) -> ConceptMap:
    """
    Retrieve a ConceptMap by its identifier from the current scope or any parent scopes.

    Args:
        identifier (str): The unique identifier of the ConceptMap to retrieve.

    Returns:
        ConceptMap: The ConceptMap instance associated with the given identifier.

    Raises:
        MappingError: If the ConceptMap with the specified identifier is not found in the current or any parent scopes.
    """
    concept_map = self.concept_maps.get(identifier)
    if concept_map:
        return concept_map

    if self.parent:
        return self.parent.get_concept_map(identifier)

    raise MappingError(
        f"Concept map '{identifier}' not found in current or parent scopes."
    )

get_target_instance

get_target_instance(identifier: str) -> BaseModel

Retrieve a target instance by its identifier from the current scope or any parent scopes.

Parameters:

Name	Type	Description	Default
identifier	str	The unique identifier of the target instance to retrieve.	required

Returns:

Name	Type	Description
BaseModel	BaseModel	The target instance associated with the given identifier.

Raises:

Type	Description
MappingError	If the target instance is not found in the current or any parent scopes.

Source code in fhircraft/fhir/mapper/engine/scope.py

def get_target_instance(self, identifier: str) -> BaseModel:
    """
    Retrieve a target instance by its identifier from the current scope or any parent scopes.

    Args:
        identifier (str): The unique identifier of the target instance to retrieve.

    Returns:
        BaseModel: The target instance associated with the given identifier.

    Raises:
        MappingError: If the target instance is not found in the current or any parent scopes.
    """
    instance = self.target_instances.get(identifier)
    if instance:
        return instance

    if self.parent:
        return self.parent.get_target_instance(identifier)

    raise MappingError(
        f"Target instance '{identifier}' not found in current or parent scopes."
    )

get_source_instance

get_source_instance(identifier: str) -> BaseModel

Retrieve a source instance by its identifier from the current scope or any parent scopes.

Parameters:

Name	Type	Description	Default
identifier	str	The unique identifier of the source instance to retrieve.	required

Returns:

Name	Type	Description
BaseModel	BaseModel	The source instance associated with the given identifier.

Raises:

Type	Description
MappingError	If the source instance is not found in the current or any parent scopes.

Source code in fhircraft/fhir/mapper/engine/scope.py

def get_source_instance(self, identifier: str) -> BaseModel:
    """
    Retrieve a source instance by its identifier from the current scope or any parent scopes.

    Args:
        identifier (str): The unique identifier of the source instance to retrieve.

    Returns:
        BaseModel: The source instance associated with the given identifier.

    Raises:
        MappingError: If the source instance is not found in the current or any parent scopes.
    """
    instance = self.source_instances.get(identifier)
    if instance:
        return instance

    if self.parent:
        return self.parent.get_source_instance(identifier)

    raise MappingError(
        f"Source instance '{identifier}' not found in current or parent scopes."
    )

get_type

get_type(identifier: str) -> type[BaseModel]

Retrieve the type associated with the given identifier from the current scope or its parent scopes.

Parameters:

Name	Type	Description	Default
identifier	str	The identifier for which to retrieve the associated type.	required

Returns:

Type	Description
type[BaseModel]	type[BaseModel]: The type associated with the identifier.

Raises:

Type	Description
MappingError	If the identifier is not found in the current or any parent scope.

Source code in fhircraft/fhir/mapper/engine/scope.py

def get_type(self, identifier: str) -> type[BaseModel]:
    """
    Retrieve the type associated with the given identifier from the current scope or its parent scopes.

    Args:
        identifier (str): The identifier for which to retrieve the associated type.

    Returns:
        type[BaseModel]: The type associated with the identifier.

    Raises:
        MappingError: If the identifier is not found in the current or any parent scope.
    """
    type_ = self.types.get(identifier)
    if type_:
        return type_

    if self.parent:
        return self.parent.get_type(identifier)

    raise MappingError(
        f"Type '{identifier}' not found in current or parent scopes."
    )

resolve_symbol

resolve_symbol(identifier: str) -> Union[FHIRPath, type[BaseModel], StructureMapGroup]

Resolves a symbol (variable, type, or group) by its identifier from the current scope or any parent scopes.

Parameters:

Name	Type	Description	Default
identifier	str	The name of the symbol to resolve.	required

Returns:

Type	Description
Union[FHIRPath, type[BaseModel], StructureMapGroup]	Union[FHIRPath, type[BaseModel], StructureMapGroup]: The resolved symbol, which can be a variable, a type, or a group.

Raises:

Type	Description
MappingError	If the symbol cannot be found in the current or any parent scopes.

Source code in fhircraft/fhir/mapper/engine/scope.py

def resolve_symbol(
    self, identifier: str
) -> Union[FHIRPath, type[BaseModel], StructureMapGroup]:
    """
    Resolves a symbol (variable, type, or group) by its identifier from the current scope or any parent scopes.

    Args:
        identifier (str): The name of the symbol to resolve.

    Returns:
        Union[FHIRPath, type[BaseModel], StructureMapGroup]: The resolved symbol, which can be a variable, a type, or a group.

    Raises:
        MappingError: If the symbol cannot be found in the current or any parent scopes.
    """
    # Handle special _DefaultMappingGroup_ symbol
    if identifier == "_DefaultMappingGroup_":
        return self._resolve_default_mapping_group()

    # Check local scope first
    if identifier in self.variables:
        return self.variables[identifier]
    elif identifier in self.types:
        return self.types[identifier]
    elif identifier in self.groups:
        return self.groups[identifier]

    # Check parent scope
    if self.parent:
        try:
            return self.parent.resolve_symbol(identifier)
        except MappingError:
            pass

    raise MappingError(
        f"Symbol '{identifier}' not found in current or parent scopes."
    )

has_symbol

has_symbol(identifier: str) -> bool

Checks if a symbol with the given identifier exists in the current scope.

Parameters:

Name	Type	Description	Default
identifier	str	The name of the symbol to check.	required

Returns:

Name	Type	Description
bool	bool	True if the symbol exists, False otherwise.

Source code in fhircraft/fhir/mapper/engine/scope.py

def has_symbol(self, identifier: str) -> bool:
    """
    Checks if a symbol with the given identifier exists in the current scope.

    Args:
        identifier (str): The name of the symbol to check.

    Returns:
        bool: True if the symbol exists, False otherwise.
    """
    try:
        self.resolve_symbol(identifier)
        return True
    except MappingError:
        return False

has_local_symbol

has_local_symbol(identifier: str) -> bool

Checks if a symbol with the given identifier exists in the current scope only.

Parameters:

Name	Type	Description	Default
identifier	str	The name of the symbol to check.	required

Returns:

Name	Type	Description
bool	bool	True if the symbol exists, False otherwise.

Source code in fhircraft/fhir/mapper/engine/scope.py

def has_local_symbol(self, identifier: str) -> bool:
    """
    Checks if a symbol with the given identifier exists in the current scope only.

    Args:
        identifier (str): The name of the symbol to check.

    Returns:
        bool: True if the symbol exists, False otherwise.
    """
    return (
        identifier in self.variables
        or identifier in self.types
        or identifier in self.groups
    )

resolve_fhirpath

resolve_fhirpath(identifier: str) -> FHIRPath

Resolve a symbol as a FHIRPath expression.

Parameters:

Name	Type	Description	Default
identifier	str	The symbol identifier to resolve	required

Returns: The resolved FHIRPath expression Raises: MappingError: If the symbol is not found or is not a FHIRPath

Source code in fhircraft/fhir/mapper/engine/scope.py

def resolve_fhirpath(self, identifier: str) -> FHIRPath:
    """
    Resolve a symbol as a FHIRPath expression.

    Args:
        identifier: The symbol identifier to resolve
    Returns:
        The resolved FHIRPath expression
    Raises:
        MappingError: If the symbol is not found or is not a FHIRPath
    """
    symbol = self.resolve_symbol(identifier)
    if not isinstance(symbol, FHIRPath):
        raise MappingError(f"Symbol '{identifier}' is not a FHIRPath expression.")
    return symbol

get_all_visible_symbols

get_all_visible_symbols() -> Dict[str, Union[FHIRPath, type[BaseModel], StructureMapGroup]]

Retrieves all visible symbols in the current scope, including those inherited from parent scopes.

This method aggregates symbols from the parent scope (if present) and then overrides them with symbols defined in the current scope. The symbols include variables, types, and groups.

Returns:

Type	Description
Dict[str, Union[FHIRPath, type[BaseModel], StructureMapGroup]]	Dict[str, Union[FHIRPath, type[BaseModel], StructureMapGroup]]: A dictionary mapping symbol names to their corresponding objects, representing all symbols visible in the current scope.

Source code in fhircraft/fhir/mapper/engine/scope.py

def get_all_visible_symbols(
    self,
) -> Dict[str, Union[FHIRPath, type[BaseModel], StructureMapGroup]]:
    """
    Retrieves all visible symbols in the current scope, including those inherited from parent scopes.

    This method aggregates symbols from the parent scope (if present) and then overrides them with
    symbols defined in the current scope. The symbols include variables, types, and groups.

    Returns:
        Dict[str, Union[FHIRPath, type[BaseModel], StructureMapGroup]]:
            A dictionary mapping symbol names to their corresponding objects, representing all
            symbols visible in the current scope.
    """
    all_symbols = {}

    # Start with parent symbols (if any)
    if self.parent:
        all_symbols.update(self.parent.get_all_visible_symbols())

    # Override with local symbols
    all_symbols.update(self.variables)
    all_symbols.update(self.types)
    all_symbols.update(self.groups)

    return all_symbols

get_scope_path

get_scope_path() -> List[str]

Returns the hierarchical path of scope names from the root to the current scope as a list of strings.

If the current scope has a parent, the method recursively retrieves the parent's scope path and appends the current scope's name. If there is no parent, returns a list containing only the current scope's name.

Returns:

Type	Description
List[str]	List[str]: The list of scope names representing the path from the root to the current scope.

Source code in fhircraft/fhir/mapper/engine/scope.py

def get_scope_path(self) -> List[str]:
    """
    Returns the hierarchical path of scope names from the root to the current scope as a list of strings.

    If the current scope has a parent, the method recursively retrieves the parent's scope path and appends the current scope's name.
    If there is no parent, returns a list containing only the current scope's name.

    Returns:
        List[str]: The list of scope names representing the path from the root to the current scope.
    """
    if self.parent:
        return self.parent.get_scope_path() + [self.name]
    return [self.name]

get_scope_depth

get_scope_depth() -> int

Returns the depth of the current scope within the scope hierarchy.

Traverses up the parent scopes recursively, incrementing the depth count for each parent until the root scope is reached.

Returns:

Name	Type	Description
int	int	The depth of the current scope, where the root scope has a depth of 0.

Source code in fhircraft/fhir/mapper/engine/scope.py

def get_scope_depth(self) -> int:
    """
    Returns the depth of the current scope within the scope hierarchy.

    Traverses up the parent scopes recursively, incrementing the depth count
    for each parent until the root scope is reached.

    Returns:
        int: The depth of the current scope, where the root scope has a depth of 0.
    """
    if self.parent:
        return self.parent.get_scope_depth() + 1
    return 0

create_child_scope

create_child_scope(name: str) -> MappingScope

Creates and returns a new child MappingScope with the specified name, setting the current scope as its parent.

Parameters:

Name	Type	Description	Default
name	str	The name of the child scope to be created.	required

Returns:

Name	Type	Description
MappingScope	MappingScope	A new instance of MappingScope with the given name and the current scope as its parent.

Source code in fhircraft/fhir/mapper/engine/scope.py

def create_child_scope(self, name: str) -> "MappingScope":
    """
    Creates and returns a new child MappingScope with the specified name, setting the current scope as its parent.

    Args:
        name (str): The name of the child scope to be created.

    Returns:
        MappingScope: A new instance of MappingScope with the given name and the current scope as its parent.
    """
    return MappingScope(name=name, parent=self)

is_processing_rule

is_processing_rule(rule_name: str) -> bool

Check if a given rule name is present in the list of processing rules.

Parameters:

Name	Type	Description	Default
rule_name	str	The name of the rule to check.	required

Returns:

Name	Type	Description
bool	bool	True if the rule is in the processing rules, False otherwise.

Source code in fhircraft/fhir/mapper/engine/scope.py

def is_processing_rule(self, rule_name: str) -> bool:
    """
    Check if a given rule name is present in the list of processing rules.

    Args:
        rule_name (str): The name of the rule to check.

    Returns:
        bool: True if the rule is in the processing rules, False otherwise.
    """
    return rule_name in self.processing_rules

start_processing_rule

start_processing_rule(rule_name: str) -> None

Marks the beginning of processing for a specific rule by adding its name to the set of currently processing rules.

Parameters:

Name	Type	Description	Default
rule_name	str	The name of the rule to start processing.	required

Source code in fhircraft/fhir/mapper/engine/scope.py

def start_processing_rule(self, rule_name: str) -> None:
    """
    Marks the beginning of processing for a specific rule by adding its name to the set of currently processing rules.

    Args:
        rule_name (str): The name of the rule to start processing.
    """
    self.processing_rules.add(rule_name)

finish_processing_rule

finish_processing_rule(rule_name: str) -> None

Marks the specified rule as finished by removing it from the set of currently processing rules.

Parameters:

Name	Type	Description	Default
rule_name	str	The name of the rule to mark as finished.	required

Source code in fhircraft/fhir/mapper/engine/scope.py

def finish_processing_rule(self, rule_name: str) -> None:
    """
    Marks the specified rule as finished by removing it from the set of currently processing rules.

    Args:
        rule_name (str): The name of the rule to mark as finished.
    """
    self.processing_rules.discard(rule_name)

T module-attribute

T = TypeVar('T')

MappingTransformer

Path: fhircraft.fhir.mapper.engine.transformer.MappingTransformer

MappingTransformer()

MappingTransformer is responsible for executing various FHIR Mapping Language transforms within a mapping scope. It provides implementations for a set of standard transforms (such as copy, create, truncate, cast, append, reference, uuid, translate, evaluate, etc.) used in FHIR StructureMap-based data transformations.

Attributes:

Name	Type	Description
_transforms	dict	A mapping of transform names to their corresponding handler methods.

Raises:

Type	Description
MappingError	If an invalid or unsupported transform is requested.
NotImplementedError	If a requested transform or feature is not implemented.
RuleProcessingError	For invalid parameters or processing errors in transforms.

Methods:

Name	Description
execute	Executes a registered FHIR Mapping Language transform by name.

Source code in fhircraft/fhir/mapper/engine/transformer.py

def __init__(self):
    self._transforms = {
        "copy": self._copy_transform,
        "create": self._create_transform,
        "truncate": self._truncate_transform,
        "escape": None,  # Not implemented yet
        "cast": self._cast_transform,
        "append": self._append_transform,
        "reference": self._reference_transform,
        "dateOp": None,  # Not implemented yet
        "uuid": self._uuid_transform,
        "pointer": None,  # Not implemented yet
        "translate": self._translate_transform,
        "evaluate": self._evaluate_transform,
        "cc": self._cc_transform,
        "c": self._c_transform,
        "qty": self._qty_transform,
        "id": self._id_transform,
        "cp": self._cp_transform,
    }

execute

execute(name: str, scope: MappingScope, parameters: List[StructureMapGroupRuleTargetParameter]) -> Any

Executes a registered FHIR Mapping Language transform by name.

Parameters:

Name	Type	Description	Default
name	str	The name of the transform to execute.	required
scope	MappingScope	The current execution scope or context for the transform.	required
parameters	List[StructureMapGroupRuleTargetParameter]	Parameters to be passed to the transform function.	required

Returns:

Name	Type	Description
result	Any	The result of the executed transform function.

Raises:

Type	Description
MappingError	If the specified transform name is not registered.
NotImplementedError	If the transform is registered but not implemented.

Source code in fhircraft/fhir/mapper/engine/transformer.py

def execute(
    self, name: str, scope: MappingScope, parameters: List[StructureMapParameter]
) -> Any:
    """
    Executes a registered FHIR Mapping Language transform by name.

    Args:
        name: The name of the transform to execute.
        scope: The current execution scope or context for the transform.
        parameters: Parameters to be passed to the transform function.

    Returns:
        result (Any): The result of the executed transform function.

    Raises:
        MappingError: If the specified transform name is not registered.
        NotImplementedError: If the transform is registered but not implemented.
    """
    if name not in self._transforms:
        raise MappingError(f"Invalid FHIR Mapping Language transform: {name}")
    if not (transform := self._transforms[name]):
        raise NotImplementedError(f"Transform '{name}' is not implemented.")
    return transform(scope, parameters)

TransformParameter

Path: fhircraft.fhir.mapper.engine.transformer.TransformParameter

dataclass

TransformParameter(name: str, type: str, is_optional: bool = False)

Represents a parameter used in a transformation process.

Parameters:

Name	Type	Description	Default
name	str	The name of the parameter.	required
type	str	The data type of the parameter.	required
is_optional	bool	Indicates whether the parameter is optional. Defaults to False.	False

Attributes:

Name	Type	Description
name	str	The name of the parameter.
type	str	The data type of the parameter.
is_optional	bool	Indicates whether the parameter is optional. Defaults to False.

name instance-attribute

name: str

The name of the parameter.

type instance-attribute

type: str

The data type of the parameter.

is_optional class-attribute instance-attribute

is_optional: bool = False

Indicates whether the parameter is optional. Defaults to False.

validate_transform_parameters

validate_transform_parameters(*signatures: list[TransformParameter]) -> Callable

A decorator to validate the parameters passed to a transformation function against one or more expected signatures.

Each signature is a list of TransformParameter objects that define the expected parameter names, types, and whether they are optional. The decorator checks if the provided parameters match any of the given signatures in terms of count and type. If a match is found, the decorated function is called with the validated and unpacked parameters. If no signature matches, a RuleProcessingError is raised.

Parameters:

Name	Type	Description	Default
*signatures	list[TransformParameter]	One or more lists of TransformParameter objects representing valid parameter signatures.	()

Returns:

Name	Type	Description
Callable	Callable	A decorator that wraps the target function with parameter validation logic.

Raises:

Type	Description
RuleProcessingError	If the provided parameters do not match any of the given signatures.

Source code in fhircraft/fhir/mapper/engine/transformer.py

def validate_transform_parameters(
    *signatures: list[TransformParameter],
) -> Callable:
    """
    A decorator to validate the parameters passed to a transformation function against one or more expected signatures.

    Each signature is a list of `TransformParameter` objects that define the expected parameter names, types, and whether they are optional.
    The decorator checks if the provided parameters match any of the given signatures in terms of count and type.
    If a match is found, the decorated function is called with the validated and unpacked parameters.
    If no signature matches, a `RuleProcessingError` is raised.

    Args:
        *signatures (list[TransformParameter]): One or more lists of `TransformParameter` objects representing valid parameter signatures.

    Returns:
        Callable: A decorator that wraps the target function with parameter validation logic.

    Raises:
        RuleProcessingError: If the provided parameters do not match any of the given signatures.
    """

    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(
            scope: MappingScope, parameters: List[StructureMapParameter]
        ) -> Any:
            error = ""
            if not signatures:
                # No validation required
                return func(scope)
            # Try all signatures; only raise if none match
            for signature in signatures:
                min_count = sum(1 for param in signature if not param.is_optional)
                max_count = len(signature)
                if len(parameters) < min_count or len(parameters) > max_count:
                    error = f"{func.__name__} requires between {min_count} and {max_count} parameters, but got {len(parameters)}."
                    continue
                transform_arguments = {}
                for param, expected in zip(parameters, signature):
                    if (value := getattr(param, f"value{expected.type}", None)) is None:
                        error = f"Parameter '{expected.name}' expected type '{expected.type}', but got '{type(param)}'."
                        break
                    transform_arguments[expected.name] = value
                else:
                    # All parameters match this signature
                    return func(scope, **transform_arguments)
            # No signature matched
            raise RuleProcessingError(
                f"Parameters did not match any valid signature for {func.__name__}. {error}"
            )

        return wrapper

    return decorator