Skip to content

FHIR Mapping Language

FHIR Mapping Language parser and transformation utilities.

FHIRMappingEngineComponent

Path: fhircraft.fhir.mapper.engine.abstract.FHIRMappingEngineComponent

Bases: ABC

Base class for components of the FHIR Mapper Engine.

Methods:

Name Description
process

Process the component within the given mapping scope.

resolve_fhirpath_within_context

Resolve FHIRPath expressions within the given context.

process abstractmethod

process(scope: MappingScope) -> Any

Process the component within the given mapping scope.

Source code in fhircraft/fhir/mapper/engine/abstract.py
@abstractmethod
def process(self, scope: "MappingScope") -> Any:
    """Process the component within the given mapping scope."""
    raise NotImplementedError("Subclasses must implement the process method.")

resolve_fhirpath_within_context staticmethod

resolve_fhirpath_within_context(expression: str, scope: MappingScope) -> FHIRPath

Resolve FHIRPath expressions within the given context.

Source code in fhircraft/fhir/mapper/engine/abstract.py
@staticmethod
def resolve_fhirpath_within_context(
    expression: str, scope: "MappingScope"
) -> FHIRPath:
    """Resolve FHIRPath expressions within the given context."""
    for variable_name, variable_path in scope.variables.items():
        expression = re.sub(
            rf"(?<!\.|\w){variable_name}", str(variable_path), expression
        )
    return fhirpath_parser.parse(expression)

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.

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)

execute

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

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 | StructureMap | 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:

Name Type Description
tuple tuple[BaseModel | dict, ...]

A tuple of resulting target instances after the transformation, which can be a mixture of BaseModel instances and/or dictionaries.

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: (
        R4_models.StructureMap | R4B_models.StructureMap | R5_models.StructureMap
    ),
    sources: tuple[BaseModel | dict],
    targets: tuple[BaseModel | dict] | None = None,
    group: str | None = None,
) -> tuple[BaseModel | dict, ...]:
    """
    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: The StructureMap resource defining the transformation rules.
        sources: Source data to be mapped, as a tuple of Pydantic models or dictionaries.
        targets: Optional target instances to populate. If not provided, new instances are created as needed.
        group: The name of the entrypoint group to execute. If not specified, the first group is used.

    Returns:
        tuple: A tuple of resulting target instances after the transformation, which can be a mixture of BaseModel instances and/or dictionaries.

    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(
            [(str(group.name), Group(group)) for group in structure_map.group or []]  # type: ignore
        ),
        concept_maps={
            str(map.id): map
            for map in (structure_map.contained or [])
            if isinstance(
                map,
                (R4_models.ConceptMap, R4B_models.ConceptMap, R5_models.ConceptMap),
            )
            and map.id
        },
    )

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

    # Parse and validate constants
    for const in getattr(structure_map, "const", None) 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.inputs)
            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.inputs
                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.inputs:
        if not input.name:
            raise ValueError(
                f"Input in group '{target_group.name}' is missing a name."
            )
        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
    target_group.process(scope=global_scope, parameters=parameters)

    # 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()
        ]
    )

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()

MappingDigestionError

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

Bases: MappingError

Raised during the digestion of StructureMap definitions.

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.

SourceAssertionError

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

Bases: MappingError

Raised when a source's assertion fails.

SourceConditionError

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

Bases: MappingError

Raised when a source's condition fails.

SourceProcessingError

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

Bases: MappingError

Raised when source processing fails.

SourceTypeError

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

Bases: MappingError

Raised when a source's type condition fails.

ValidationError

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

Bases: MappingError

Raised when input data validation fails.

Group

Path: fhircraft.fhir.mapper.engine.group.Group

Group(definition: StructureMapGroup | StructureMapGroup | StructureMapGroup, parent_group=None)

Bases: FHIRMappingEngineComponent

Represents a StructureMap Group with its own processing logic.

Parameters:

Name Type Description Default
source

The StructureMapGroupRuleTarget to initialize from.

required

Raises: SourceProcessingError: If required fields are missing.

Methods:

Name Description
bind_parameters

Bind input parameters to the group scope.

process

Processes a StructureMap group by validating input parameters, constructing a local mapping scope,

Source code in fhircraft/fhir/mapper/engine/group.py
def __init__(
    self,
    definition: "R4_StructureMapGroup | R4B_StructureMapGroup | R5_StructureMapGroup",
    parent_group=None,
):
    """
    Initializes a Rule instance from a StructureMapGroupRuleTarget.

    Args:
        source: The StructureMapGroupRuleTarget to initialize from.
    Raises:
        SourceProcessingError: If required fields are missing.
    """
    self.definition = definition
    self.name = definition.name or f"group-{id(definition)}"
    self.parent_group = parent_group
    self.rules: List[Rule] = [
        Rule(rule, parent_group=self) for rule in definition.rule or []
    ]
    self._organize_rules()
    # Parse inputs
    if not self.definition.input:
        raise MappingDigestionError(
            f"Group '{self.name}' has no input definitions."
        )
    self.inputs = self.definition.input

bind_parameters

bind_parameters(scope: MappingScope, parameters: Sequence[FHIRPath], is_dependent: bool) -> None

Bind input parameters to the group scope.

Source code in fhircraft/fhir/mapper/engine/group.py
def bind_parameters(
    self, scope: "MappingScope", parameters: Sequence[FHIRPath], is_dependent: bool
) -> None:
    """Bind input parameters to the group scope."""
    if len(parameters) != len(self.inputs):
        raise MappingError(
            f"Expected {len(self.inputs)} parameters, got {len(parameters)}"
        )

    for input, parameter in zip(self.inputs, parameters):
        if input.mode == "target" and not is_dependent:
            if not input.type:
                raise MappingError(
                    f"Target input '{input.name}' in group '{self.name}' must have a type specified."
                )

        if input.type:
            try:
                scope.get_type(input.type)
            except MappingError:
                raise MappingError(
                    f"Input '{input.name}' in group '{self.name}' has unknown type '{input.type}'."
                )
        if not input.name:
            raise MappingError(
                f"A {input.mode} input in group '{self.name}' is missing a name."
            )

        scope.define_variable(input.name, parameter)

process

process(scope: MappingScope, parameters: Sequence[FHIRPath], is_dependent: bool = False)

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
parameters Sequence[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/group.py
def process(
    self,
    scope: "MappingScope",
    parameters: Sequence[FHIRPath],
    is_dependent: bool = False,
):
    """
    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:
        parameters: The input parameters to be mapped, corresponding to the group's input definitions.
        scope: 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.
    """
    # Construct local group scope
    group_scope = MappingScope(
        name=self.name,
        parent=scope,
    )

    self.bind_parameters(group_scope, parameters, is_dependent)

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

Rule

Path: fhircraft.fhir.mapper.engine.rule.Rule

Rule(definition: StructureMapGroupRule | StructureMapGroupRule | StructureMapGroupRule, parent_group=None)

Bases: FHIRMappingEngineComponent

Represents a StructureMap Rule with iteration and condition logic.

A Rule contains Sources and Targets, manages iteration over source collections, and handles conditions and dependencies.

Parameters:

Name Type Description Default
source

The StructureMapGroupRuleTarget to initialize from.

required

Raises: SourceProcessingError: If required fields are missing.

Methods:

Name Description
process

Processes the rule within the given mapping scope.

Attributes:

Name Type Description
has_first_target bool

Check if this rule has a target with 'first' list mode.

has_last_target bool

Check if this rule has a target with 'last' list mode.

Source code in fhircraft/fhir/mapper/engine/rule.py
def __init__(
    self,
    definition: "R4_StructureMapGroupRule | R4B_StructureMapGroupRule| R5_StructureMapGroupRule",
    parent_group=None,
):
    """
    Initializes a Rule instance from a StructureMapGroupRuleTarget.

    Args:
        source: The StructureMapGroupRuleTarget to initialize from.
    Raises:
        SourceProcessingError: If required fields are missing.
    """
    self.definition = definition
    self.name = definition.name or f"rule-{id(definition)}"
    self.parent_group = parent_group
    self.sources: List[RuleSource] = []
    self.targets: List[RuleTarget] = []
    self.nested_rules: List[Rule] = []
    self.dependents: List[
        "R4_StructureMapGroupRuleDependent | R4B_StructureMapGroupRuleDependent | R5_StructureMapGroupRuleDependent"
    ] = []

    # Create source objects
    for source_def in self.definition.source or []:
        self.sources.append(RuleSource(source_def, self))

    # Create target objects
    for target_def in self.definition.target or []:
        self.targets.append(RuleTarget(target_def, self))

    # Create nested rules
    for nested_def in self.definition.rule or []:
        self.nested_rules.append(Rule(nested_def, self.parent_group))

    # Extract dependents
    for dependent in self.definition.dependent or []:
        if not dependent.name:
            raise MappingDigestionError("Dependent rule or group must have a name")
        self.dependents.append(dependent)

has_first_target property

has_first_target: bool

Check if this rule has a target with 'first' list mode.

has_last_target property

has_last_target: bool

Check if this rule has a target with 'last' list mode.

process

process(scope: MappingScope)

Processes the rule within the given mapping scope.

Parameters:

Name Type Description Default
scope MappingScope

The current mapping scope.

required
Source code in fhircraft/fhir/mapper/engine/rule.py
def process(
    self,
    scope: "MappingScope",
):
    """
    Processes the rule within the given mapping scope.

    Args:
        scope: The current mapping scope.
    """

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

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

        # Process sources first to determine iteration
        source_iterations = {}

        for source in self.sources:
            try:
                source.process(scope)
            except (SourceTypeError, SourceConditionError):
                logger.debug(
                    f"Source type or condition violated in rule {self.name}. Skipping rule."
                )
                return scope
            except SourceAssertionError:
                raise SourceAssertionError(
                    f"Source assertion failed for rule {self.name}"
                )
            source_iterations[source.variable] = source.iteration_count

        for source_var, iterations in source_iterations.items():
            for source_iteration in range(iterations):
                logger.debug(
                    f"Processing iteration {source_iteration} for rule {self.name}"
                )
                # Create local 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(),
                    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 self.targets:
                    target.process(iteration_scope)

                # Process dependent rules for this iteration
                for dependent in self.dependents:
                    self._process_dependent_group(dependent, iteration_scope)

                # Process nested rules for this iteration
                for nested_rule in self.nested_rules:
                    nested_rule.process(iteration_scope)

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

    finally:
        scope.finish_processing_rule(self.name)
    return scope

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 | ConceptMap | ConceptMap] = dict(), groups: OrderedDict[str, Group] = OrderedDict(), variables: Dict[str, FHIRPath] = dict(), default_groups: Dict[str, Group] = 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 | ConceptMap | 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, str]

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, ForwardRef]

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 | ConceptMap | ConceptMap]

Registry of available concept maps for value transformations

groups OrderedDict[str, Group]

The groups defined on this scope

variables Dict[str, FHIRPath]

Registry of variables mapped to resolved FHIRPath expressions

default_groups Dict[str, Group]

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 | ConceptMap | ConceptMap] = field(default_factory=dict)

Registry of available concept maps for value transformations

groups class-attribute instance-attribute

groups: OrderedDict[str, Group] = 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, Group] = 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 | ConceptMap | 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 | 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
) -> R4_ConceptMap | R4B_ConceptMap | R5_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], Group]

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:

Name Type Description
value Union[FHIRPath, type[BaseModel], Group]

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],
    "Group",
]:
    """
    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:
        value: 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

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 | StructureMapGroup | 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],
        R4_StructureMapGroup | R4B_StructureMapGroup | R5_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')

RuleSource

Path: fhircraft.fhir.mapper.engine.source.RuleSource

Bases: FHIRMappingEngineComponent

Parameters:

Name Type Description Default
source StructureMapGroupRuleSource | StructureMapGroupRuleSource | StructureMapGroupRuleSource

The StructureMapGroupRuleSource to initialize from.

required

Raises: MappingDigestionError: If required fields are missing.

Methods:

Name Description
process

Processes the source within the given mapping scope.

Source code in fhircraft/fhir/mapper/engine/source.py
def __init__(
    self,
    source: "R4_StructureMapGroupRuleSource | R4B_StructureMapGroupRuleSource | R5_StructureMapGroupRuleSource",
    parent_rule: "Rule",
):
    """
    Initializes a RuleSource instance from a StructureMapGroupRuleSource.

    Args:
        source: The StructureMapGroupRuleSource to initialize from.
    Raises:
        MappingDigestionError: If required fields are missing.
    """
    self.definition = source
    if self.definition.context is None:
        raise MappingDigestionError("Source context is required")
    self.parent_rule = parent_rule
    self.variable = source.variable or f"source-{id(source)}"
    self.resolved_path: Optional[FHIRPath] = None
    self.iteration_count = 0
    self.condition = (
        source.condition if source.condition else fhirpath.Literal(True)
    )
    self.assertion = source.check if source.check else fhirpath.Literal(True)

process

process(scope: MappingScope) -> None

Processes the source within the given mapping scope.

Source code in fhircraft/fhir/mapper/engine/source.py
def process(
    self,
    scope: "MappingScope",
) -> None:
    """
    Processes the source within the given mapping scope.
    """
    self.resolved_path = scope.resolve_fhirpath(self.definition.context)  # type: ignore
    if self.resolved_path is None:
        raise SourceProcessingError(
            f"Source context {self.definition.context} not found"
        )

    # Apply element path if specified
    if self.definition.element:
        self.resolved_path = self.resolved_path._invoke(
            fhirpath.Element(self.definition.element)
        )
    # Apply list mode if specified
    self._apply_list_mode()

    # Store source FHIRPath in scope
    scope.define_variable(self.variable, self.resolved_path)
    # Store resolved path and calculate iteration count
    self.iteration_count = self.resolved_path.count(scope.get_instances()) or 0

    # Evaluate conditions
    if not self._check_type_condition(scope):
        raise SourceTypeError(
            f"Source type condition not met for source {self.variable}"
        )
    if not self._check_where_condition(scope):
        raise SourceConditionError(
            f"Source condition not met for source {self.variable}"
        )
    if not self._check_assertion_condition(scope):
        raise SourceAssertionError(
            f"Source assertion failed for source {self.variable}"
        )
    if not self._validate_cardinality():
        raise SourceProcessingError("Cardinality constraints violated")

RuleTarget

Path: fhircraft.fhir.mapper.engine.target.RuleTarget

Bases: FHIRMappingEngineComponent

Parameters:

Name Type Description Default
source StructureMapGroupRuleTarget | StructureMapGroupRuleTarget | StructureMapGroupRuleTarget

The StructureMapGroupRuleTarget to initialize from.

required

Raises: SourceProcessingError: If required fields are missing.

Methods:

Name Description
process

Process the target, creating path and executing transforms.

has_list_mode

Check if this target has a specific list mode.

Source code in fhircraft/fhir/mapper/engine/target.py
def __init__(
    self,
    source: "R4_StructureMapGroupRuleTarget | R4B_StructureMapGroupRuleTarget | R5_StructureMapGroupRuleTarget",
    parent_rule: "Rule",
):
    """
    Initializes a RuleSource instance from a StructureMapGroupRuleTarget.

    Args:
        source: The StructureMapGroupRuleTarget to initialize from.
    Raises:
        SourceProcessingError: If required fields are missing.
    """
    self.definition = source
    if self.definition.context is None:
        raise MappingDigestionError("Source context is required")
    self.parent_rule = parent_rule
    self.variable = source.variable or f"target-{id(source)}"
    self.resolved_path: Optional[FHIRPath] = None
    self.transform = self._resolve_transform(
        self.definition.transform, self.definition.parameter
    )

process

process(scope: MappingScope) -> None

Process the target, creating path and executing transforms.

Source code in fhircraft/fhir/mapper/engine/target.py
def process(
    self,
    scope: "MappingScope",
) -> None:
    """Process the target, creating path and executing transforms."""
    # Resolve target path
    self.resolved_path = scope.resolve_fhirpath(self.definition.context)  # type: ignore
    # Apply element path if specified
    if self.definition.element:
        self.resolved_path = self.resolved_path._invoke(
            fhirpath.Element(self.definition.element)
        )

    # Determine insertion index
    insert_index = self.resolved_path.count(scope.get_instances())
    self.resolved_path = self.resolved_path._invoke(fhirpath.Index(insert_index))

    # Apply transform if specified
    if self.transform:
        self.definition.parameter = self.definition.parameter or []  # type: ignore
        # Execute the transform
        transformed_value = self.transform.process(scope)
        # Update the target structure
        self.resolved_path.update_single(scope.get_instances(), transformed_value)

    # Define variable in scope
    scope.define_variable(self.variable, self.resolved_path)

has_list_mode

has_list_mode(mode: str) -> bool

Check if this target has a specific list mode.

Source code in fhircraft/fhir/mapper/engine/target.py
def has_list_mode(self, mode: str) -> bool:
    """Check if this target has a specific list mode."""
    target_mode = getattr(self.definition, "listMode", None)
    if isinstance(target_mode, list):
        return mode in target_mode
    return target_mode == mode

Append

Path: fhircraft.fhir.mapper.engine.transforms.Append

Bases: MappingTransform

Implements the 'append' transform, which appends string representations of parameters.

  • If valueId is present, it resolves the FHIRPath and appends its string value.
  • If valueString is present, it appends the string directly.
  • If neither is present, it raises a RuleProcessingError.

Parameters:

Name Type Description Default
parameters Sequence[StructureMapGroupRuleTargetParameter | StructureMapGroupRuleTargetParameter | StructureMapGroupRuleTargetParameter]

A sequence of StructureMapParameter objects to process.

required

Raises:

Type Description
RuleProcessingError

If no parameters are provided, or if a parameter does not have a valid type (valueId or valueString).

Methods:

Name Description
process

Appends the string representations of the provided parameters.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def __init__(
    self,
    parameters: Sequence[
        "R4_StructureMapParameter | R4B_StructureMapParameter | R5_StructureMapParameter"
    ],
):
    """
    Processes a list of `StructureMapParameter` objects, extracting their `valueId` or `valueString` attributes.
    For each parameter:
    - If `valueId` is present, it resolves the FHIRPath and appends its string value.
    - If `valueString` is present, it appends the string directly.
    - If neither is present, it raises a `RuleProcessingError`.

    Args:
        parameters: A sequence of StructureMapParameter objects to process.

    Raises:
        RuleProcessingError: If no parameters are provided, or if a parameter does not have a valid type (`valueId` or `valueString`).
    """
    if len(parameters) < 1:
        raise ValueError("Append transform requires at least one parameter")
    self.elements = []
    for parameter in parameters:
        if not (parameter.valueId or parameter.valueString):
            raise ValueError(
                "Append transform parameters must be of type Id or String"
            )
        is_literal = parameter.valueId is None
        self.elements.append(
            (is_literal, str(parameter.valueId or parameter.valueString))
        )

process

process(scope: MappingScope) -> Any

Appends the string representations of the provided parameters.

Parameters:

Name Type Description Default
scope MappingScope

The current mapping scope used to resolve FHIRPath expressions.

required

Returns:

Name Type Description
str Any

The concatenated string of all parameter values.

Raises:

Type Description
RuleProcessingError

If no parameters are provided, or if a parameter does not have a valid type (valueId or valueString).

Source code in fhircraft/fhir/mapper/engine/transforms.py
def process(self, scope: "MappingScope") -> Any:
    """
    Appends the string representations of the provided parameters.

    Args:
        scope: The current mapping scope used to resolve FHIRPath expressions.

    Returns:
        str: The concatenated string of all parameter values.

    Raises:
        RuleProcessingError: If no parameters are provided, or if a parameter does not have a valid type (`valueId` or `valueString`).
    """
    strings = []
    for is_literal, value in self.elements:
        if is_literal:
            strings.append(value)
        else:
            source_fhirpath = scope.resolve_fhirpath(value)
            strings.append(str(source_fhirpath.single(scope.get_instances())))
    return "".join(strings)

Cast

Path: fhircraft.fhir.mapper.engine.transforms.Cast

Bases: MappingTransform

Implements the 'cast' transform, which casts a value to a specified type.

Methods:

Name Description
process

Casts the value of a FHIRPath expression to a specified target type.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def __init__(
    self,
    parameters: Sequence[
        "R4_StructureMapParameter | R4B_StructureMapParameter | R5_StructureMapParameter"
    ],
):
    if len(parameters) not in [1, 2]:
        raise ValueError("Cast transform requires one or two parameters")
    self.source = parameters[0].value
    if len(parameters) == 2:
        self.type_specifier = (
            parameters[1].value[0].upper() + parameters[1].value[1:]
        )
        self.type_conversion_function = getattr(fp, f"To{self.type_specifier}", None)  # type: ignore
        if not self.type_conversion_function:
            raise ValueError(f"Unsupported type for cast: {self.type_specifier}")
    else:
        self.type_specifier = None
        self.type_conversion_function = None

process

process(scope: MappingScope) -> Any

Casts the value of a FHIRPath expression to a specified target type.

Parameters:

Name Type Description Default
scope MappingScope

The current mapping scope containing context and instance data.

required

Returns:

Name Type Description
Any Any

The value of the source FHIRPath expression cast to the specified type.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def process(self, scope: "MappingScope") -> Any:
    """
    Casts the value of a FHIRPath expression to a specified target type.

    Args:
        scope (MappingScope): The current mapping scope containing context and instance data.

    Returns:
        Any: The value of the source FHIRPath expression cast to the specified type.
    """
    source_fhirpath = scope.resolve_fhirpath(self.source)
    # Explicit casting
    if self.type_conversion_function:
        return source_fhirpath._invoke(self.type_conversion_function()).single(
            scope.get_instances()
        )
    else:
        is_decimal = source_fhirpath._invoke(
            fp.LegacyIs(fp.TypeSpecifier("Decimal"))
        ).single(scope.get_instances())
        is_date = source_fhirpath._invoke(
            fp.LegacyIs(fp.TypeSpecifier("Date"))
        ).single(scope.get_instances())
        if is_decimal:
            return source_fhirpath._invoke(fp.ToDecimal()).single(
                scope.get_instances()
            )
        elif is_date:
            return source_fhirpath._invoke(fp.ToDate()).single(
                scope.get_instances()
            )
        else:
            raise NotImplementedError(
                "Implicit casting is only supported for Decimal and Date types, please specify a target type explicitly."
            )

CodeableConcept

Path: fhircraft.fhir.mapper.engine.transforms.CodeableConcept

Bases: MappingTransform

Implements the 'cc' transform, which creates a CodeableConcept from parameters.

Methods:

Name Description
process

Creates a CodeableConcept instance from the provided Coding parameters.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def __init__(
    self,
    parameters: Sequence[
        "R4_StructureMapParameter | R4B_StructureMapParameter | R5_StructureMapParameter"
    ],
):
    if len(parameters) not in [1, 2, 3]:
        raise ValueError(
            "CodeableConcept transform requires one, two, or three parameters"
        )
    if len(parameters) == 1:
        self.text = (
            fp.Literal(text.value)
            if not (text := parameters[0]).valueId
            else text.valueId
        )
    else:
        self.code = (
            fp.Literal(code.value)
            if not (code := parameters[0]).valueId
            else code.valueId
        )
        self.system = (
            fp.Literal(system.value)
            if not (system := parameters[1]).valueId
            else system.valueId
        )
        if len(parameters) == 3:
            self.display = (
                fp.Literal(display.value)
                if not (display := parameters[2]).valueId
                else display.valueId
            )

process

process(scope: MappingScope) -> Any

Creates a CodeableConcept instance from the provided Coding parameters.

Parameters:

Name Type Description Default
scope MappingScope

The current mapping scope (unused in this function).

required

Returns:

Name Type Description
CodeableConcept Any

A codeableConcept dictionary containing the specified Codings.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def process(self, scope: "MappingScope") -> Any:
    """
    Creates a CodeableConcept instance from the provided Coding parameters.

    Args:
        scope: The current mapping scope (unused in this function).

    Returns:
        CodeableConcept: A codeableConcept dictionary containing the specified Codings.
    """
    if self.text:
        text = (
            scope.resolve_fhirpath(self.text)
            if isinstance(self.text, str)
            else self.text
        ).single(scope.get_instances())
        return {"text": text}
    elif self.code and self.system:
        code = (
            scope.resolve_fhirpath(self.code)
            if isinstance(self.code, str)
            else self.code
        ).single(scope.get_instances())
        system = (
            scope.resolve_fhirpath(self.system)
            if isinstance(self.system, str)
            else self.system
        ).single(scope.get_instances())
        display = None
        if self.display:
            display = (
                scope.resolve_fhirpath(self.display)
                if isinstance(self.display, str)
                else self.display
            ).single(scope.get_instances())
        return {
            "coding": [
                {
                    "code": code,
                    "system": system,
                    "display": display,
                }
            ]
        }
    else:
        raise ValueError(
            "CodeableConcept transform requires either a text parameter or code and system parameters"
        )

Coding

Path: fhircraft.fhir.mapper.engine.transforms.Coding

Bases: MappingTransform

Implements the 'c' transform, which creates a Coding from parameters.

Methods:

Name Description
process

Creates a CodeableConcept instance from the provided Coding parameters.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def __init__(
    self,
    parameters: Sequence[
        "R4_StructureMapParameter | R4B_StructureMapParameter | R5_StructureMapParameter"
    ],
):
    if len(parameters) not in [2, 3]:
        raise ValueError("Coding transform requires two or three parameters")
    self.code = (
        fp.Literal(code.value)
        if not (code := parameters[0]).valueId
        else code.valueId
    )
    self.system = (
        fp.Literal(system.value)
        if not (system := parameters[1]).valueId
        else system.valueId
    )
    if len(parameters) == 3:
        self.display = (
            fp.Literal(display.value)
            if not (display := parameters[2]).valueId
            else display.valueId
        )

process

process(scope: MappingScope) -> Any

Creates a CodeableConcept instance from the provided Coding parameters.

Parameters:

Name Type Description Default
scope MappingScope

The current mapping scope (unused in this function).

required

Returns:

Name Type Description
Coding Any

A coding dictionary containing the specified Codings.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def process(self, scope: "MappingScope") -> Any:
    """
    Creates a CodeableConcept instance from the provided Coding parameters.

    Args:
        scope: The current mapping scope (unused in this function).

    Returns:
        Coding: A coding dictionary containing the specified Codings.
    """
    if self.code and self.system:
        code = (
            scope.resolve_fhirpath(self.code)
            if isinstance(self.code, str)
            else self.code
        ).single(scope.get_instances())
        system = (
            scope.resolve_fhirpath(self.system)
            if isinstance(self.system, str)
            else self.system
        ).single(scope.get_instances())
        display = None
        if self.display:
            display = (
                scope.resolve_fhirpath(self.display)
                if isinstance(self.display, str)
                else self.display
            ).single(scope.get_instances())
        return {
            "code": code,
            "system": system,
            "display": display,
        }
    else:
        raise ValueError(
            "CodeableConcept transform requires either a text parameter or code and system parameters"
        )

ContactPoint

Path: fhircraft.fhir.mapper.engine.transforms.ContactPoint

Bases: MappingTransform

Implements the 'cp' transform, which creates a ContactPoint

Methods:

Name Description
process

Creates an Identifier instance from the provided parameters.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def __init__(
    self,
    parameters: Sequence[
        "R4_StructureMapParameter | R4B_StructureMapParameter | R5_StructureMapParameter"
    ],
):
    if len(parameters) not in [1, 2]:
        raise ValueError("Identifier transform requires one or two parameters")
    if len(parameters) == 2:
        self.system = (
            fp.Literal(system.value)
            if not (system := parameters[0]).valueId
            else system.valueId
        )
    self.value = (
        fp.Literal(value.value)
        if not (value := parameters[-1]).valueId
        else value.valueId
    )

process

process(scope: MappingScope) -> Any

Creates an Identifier instance from the provided parameters.

Parameters:

Name Type Description Default
scope MappingScope

The current mapping scope (unused in this function).

required

Returns:

Name Type Description
Identifier Any

An identifier dictionary containing the specified system, value, and code.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def process(self, scope: "MappingScope") -> Any:
    """
    Creates an Identifier instance from the provided parameters.

    Args:
        scope: The current mapping scope (unused in this function).

    Returns:
        Identifier: An identifier dictionary containing the specified system, value, and code.
    """
    value = (
        scope.resolve_fhirpath(self.value)
        if isinstance(self.value, str)
        else self.value
    ).single(scope.get_instances())
    if self.system:
        system = (
            scope.resolve_fhirpath(self.system)
            if isinstance(self.system, str)
            else self.system
        ).single(scope.get_instances())
    else:
        # Determine through regex which system type to use
        if re.match(r"^[+]{1}(?:[0-9\-\$$\$$\/\.]\s?){6,15}[0-9]{1}$", value):
            system = "phone"
        elif re.match(r"^[\w\-\.]+@([\w-]+\.)+[\w-]{2,4}$", value):
            system = "email"
        elif re.match(r"^\+1[2-9][0-9]{9}$", value):
            system = "fax"
        elif re.match(
            r"^(https?:\/\/)?(www\.)?[-a-zA-Z0-9@:%._\+~#=]{2,256}\.[a-z]{2,4}\b([-a-zA-Z0-9@:%_\+.~#?&//=]*)$",
            value,
        ):
            system = "url"
        else:
            system = "other"
    return {
        "system": system,
        "value": value,
    }

Copy

Path: fhircraft.fhir.mapper.engine.transforms.Copy

Bases: MappingTransform

Implements the 'copy' transform, which copies the source value to the target.

Methods:

Name Description
process

Copies a value from the source or uses a provided literal.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def __init__(
    self,
    parameters: Sequence[
        "R4_StructureMapParameter | R4B_StructureMapParameter | R5_StructureMapParameter"
    ],
):
    if len(parameters) != 1:
        raise ValueError("Copy transform requires exactly one parameter")
    parameter = parameters[0]
    if parameter.valueId:
        self.source = parameter.valueId
    else:
        self.literal = parameter.value

process

process(scope: MappingScope) -> Any

Copies a value from the source or uses a provided literal.

This function attempts to copy a value from the specified source using FHIRPath resolution within the given scope. If source is not provided, it returns the given literal value instead.

Parameters:

Name Type Description Default
scope MappingScope

The mapping scope containing context and instances for FHIRPath resolution.

required

Returns:

Name Type Description
Any Any

The copied value from the source or the provided literal.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def process(self, scope: "MappingScope") -> Any:
    """
    Copies a value from the source or uses a provided literal.

    This function attempts to copy a value from the specified `source` using FHIRPath resolution within the given `scope`.
    If `source` is not provided, it returns the given `literal` value instead.

    Args:
        scope (MappingScope): The mapping scope containing context and instances for FHIRPath resolution.

    Returns:
        Any: The copied value from the source or the provided literal.

    """
    # Just copy the source value or use the literal
    if self.source:
        source_fhirpath = scope.resolve_fhirpath(self.source)
        return source_fhirpath.single(scope.get_instances())
    elif self.literal:
        return self.literal

    def __repr__(self):
        if hasattr(self, "source"):
            return f"CopyTransform(source={self.source})"
        else:
            return f"CopyTransform(literal={self.literal})"

Create

Path: fhircraft.fhir.mapper.engine.transforms.Create

Bases: MappingTransform

Implements the 'create' transform, which creates a new instance of the specified type.

Methods:

Name Description
process

Creates and returns a new instance of a model of the specified type using the provided mapping scope.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def __init__(
    self,
    parameters: Sequence[
        "R4_StructureMapParameter | R4B_StructureMapParameter | R5_StructureMapParameter"
    ],
):
    if len(parameters) != 1:
        raise ValueError("Create transform requires exactly one parameter")
    self.type_specifier = parameters[0].value

process

process(scope: MappingScope) -> Any

Creates and returns a new instance of a model of the specified type using the provided mapping scope.

Parameters:

Name Type Description Default
scope MappingScope

The mapping scope containing type definitions and model constructors.

required

Returns:

Name Type Description
BaseModel Any

A newly constructed instance of the specified model type.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def process(self, scope: "MappingScope") -> Any:
    """
    Creates and returns a new instance of a model of the specified type using the provided mapping scope.

    Args:
        scope (MappingScope): The mapping scope containing type definitions and model constructors.

    Returns:
        BaseModel: A newly constructed instance of the specified model type.
    """
    return scope.get_type(self.type_specifier).model_construct()

Evaluate

Path: fhircraft.fhir.mapper.engine.transforms.Evaluate

Bases: MappingTransform

Implements the 'evaluate' transform, which evaluates a FHIRPath expression.

Methods:

Name Description
process

Evaluates a FHIRPath expression against a specified source within the given mapping scope.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def __init__(
    self,
    parameters: Sequence[
        "R4_StructureMapParameter | R4B_StructureMapParameter | R5_StructureMapParameter"
    ],
):
    if len(parameters) not in [1, 2]:
        raise ValueError(
            "Evaluate transform requires exactly one or two parameters"
        )
    if len(parameters) == 2:
        self.source = parameters[0].value
    self.expression = parameters[-1].value

process

process(scope: MappingScope) -> Any

Evaluates a FHIRPath expression against a specified source within the given mapping scope.

Parameters:

Name Type Description Default
scope MappingScope

The current mapping scope containing context and instances.

required

Returns:

Name Type Description
Any Any

The result of evaluating the FHIRPath expression against the source context.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def process(self, scope: "MappingScope") -> Any:
    """
    Evaluates a FHIRPath expression against a specified source within the given mapping scope.

    Args:
        scope: The current mapping scope containing context and instances.

    Returns:
        Any: The result of evaluating the FHIRPath expression against the source context.
    """
    if self.source:
        context = scope.resolve_fhirpath(self.source).single(scope.get_instances())
    else:
        context = scope.get_instances()
    expression = self.resolve_fhirpath_within_context(self.expression, scope)
    transformed_values = expression.values(context)
    if len(transformed_values) == 1:
        return transformed_values[0]
    elif len(transformed_values) > 1:
        return transformed_values
    else:
        return None

Identifier

Path: fhircraft.fhir.mapper.engine.transforms.Identifier

Bases: MappingTransform

Implements the 'id' transform, which creates an Identifier

Methods:

Name Description
process

Creates an Identifier instance from the provided parameters.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def __init__(
    self,
    parameters: Sequence[
        "R4_StructureMapParameter | R4B_StructureMapParameter | R5_StructureMapParameter"
    ],
):
    if len(parameters) != 3:
        raise ValueError("Identifier transform requires three parameters")
    self.system = (
        fp.Literal(system.value)
        if not (system := parameters[0]).valueId
        else system.valueId
    )
    self.value = (
        fp.Literal(value.value)
        if not (value := parameters[1]).valueId
        else value.valueId
    )
    self.type = (
        fp.Literal(type_.value)
        if not (type_ := parameters[2]).valueId
        else type_.valueId
    )

process

process(scope: MappingScope) -> Any

Creates an Identifier instance from the provided parameters.

Parameters:

Name Type Description Default
scope MappingScope

The current mapping scope (unused in this function).

required

Returns:

Name Type Description
Identifier Any

An identifier dictionary containing the specified system, value, and code.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def process(self, scope: "MappingScope") -> Any:
    """
    Creates an Identifier instance from the provided parameters.

    Args:
        scope: The current mapping scope (unused in this function).

    Returns:
        Identifier: An identifier dictionary containing the specified system, value, and code.
    """
    system = (
        scope.resolve_fhirpath(self.system)
        if isinstance(self.system, str)
        else self.system
    ).single(scope.get_instances())
    value = (
        scope.resolve_fhirpath(self.value)
        if isinstance(self.value, str)
        else self.value
    ).single(scope.get_instances())
    type_ = (
        scope.resolve_fhirpath(self.type)
        if isinstance(self.type, str)
        else self.type
    ).single(scope.get_instances())
    return {
        "system": system,
        "value": value,
        "type": {
            "coding": [
                {
                    "code": type_,
                    "system": "http://hl7.org/fhir/ValueSet/identifier-type",
                }
            ]
        },
    }

MappingTransform

Path: fhircraft.fhir.mapper.engine.transforms.MappingTransform

Quantity

Path: fhircraft.fhir.mapper.engine.transforms.Quantity

Bases: MappingTransform

Implements the 'qty' transform, which creates a Quantity

Methods:

Name Description
process

Transforms quantity-related input parameters into a FHIR Quantity object.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def __init__(
    self,
    parameters: Sequence[
        "R4_StructureMapParameter | R4B_StructureMapParameter | R5_StructureMapParameter"
    ],
):
    if len(parameters) not in [1, 2, 4]:
        raise ValueError("Quantity transform requires one, two, or four parameters")
    if len(parameters) == 1:
        self.text = (
            fp.Literal(text.value)
            if not (text := parameters[0]).valueId
            else text.valueId
        )
    else:
        self.value = (
            fp.Literal(value.value)
            if not (value := parameters[0]).valueId
            else value.valueId
        )
        self.unit = (
            fp.Literal(unit.value)
            if not (unit := parameters[1]).valueId
            else unit.valueId
        )
        if len(parameters) == 4:
            self.system = (
                fp.Literal(system.value)
                if not (system := parameters[2]).valueId
                else system.valueId
            )
            self.code = (
                fp.Literal(code.value)
                if not (code := parameters[3]).valueId
                else code.valueId
            )

process

process(scope: MappingScope) -> Any

Transforms quantity-related input parameters into a FHIR Quantity object.

This function supports two modes of operation: 1. Parsing a single text parameter of the form '[<|<=|>=|>|ad] ', extracting the comparator, value, and unit. 2. Using explicit value and unit parameters (with optional system and code).

Parameters:

Name Type Description Default
scope MappingScope

The mapping scope context (not used in this function).

required

Returns:

Name Type Description
Quantity Any

A FHIR Quantity object constructed from the provided parameters.

Raises:

Type Description
RuleProcessingError

If the text parameter does not match the expected format.

AssertionError

If neither text nor both value and unit are provided.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def process(self, scope: "MappingScope") -> Any:
    """
    Transforms quantity-related input parameters into a FHIR Quantity object.

    This function supports two modes of operation:
    1. Parsing a single `text` parameter of the form '[<|<=|>=|>|ad]<number> <unit>', extracting the comparator, value, and unit.
    2. Using explicit `value` and `unit` parameters (with optional `system` and `code`).

    Args:
        scope (MappingScope): The mapping scope context (not used in this function).

    Returns:
        Quantity: A FHIR Quantity object constructed from the provided parameters.

    Raises:
        RuleProcessingError: If the `text` parameter does not match the expected format.
        AssertionError: If neither `text` nor both `value` and `unit` are provided.
    """
    if self.text:
        text = (
            scope.resolve_fhirpath(self.text)
            if isinstance(self.text, str)
            else self.text
        ).single(scope.get_instances())
        matches = re.search(r"(<|<=|>=|>|ad)?(\d+((\.|\,)\d+)?) (.*)", text)
        if not matches:
            raise MappingError(
                "The 'qty' transform single parameter must be of the form '[<|<=|>=|>|ad]<number> <unit>'"
            )

        return dict(
            comparator=matches.group(1) if matches.group(1) else None,
            value=float(matches.group(2).replace(",", ".")),
            unit=matches.group(5),
            system=None,
            code=None,
        )
    else:
        assert self.value and self.unit
        return dict(
            value=(
                scope.resolve_fhirpath(self.value)
                if isinstance(self.value, str)
                else self.value
            ).single(scope.get_instances()),
            unit=(
                scope.resolve_fhirpath(self.unit)
                if isinstance(self.unit, str)
                else self.unit
            ).single(scope.get_instances()),
            system=(
                (
                    scope.resolve_fhirpath(self.system)
                    if isinstance(self.system, str)
                    else self.system
                ).single(scope.get_instances())
                if self.system
                else None
            ),
            code=(
                (
                    scope.resolve_fhirpath(self.code)
                    if isinstance(self.code, str)
                    else self.code
                ).single(scope.get_instances())
                if self.code
                else None
            ),
        )

Reference

Path: fhircraft.fhir.mapper.engine.transforms.Reference

Bases: MappingTransform

Implements the 'reference' transform, which creates a FHIR reference from a given source.

Methods:

Name Description
process

Transforms a FHIR resource reference by extracting its resource type and ID from the given source.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def __init__(
    self,
    parameters: Sequence[
        "R4_StructureMapParameter | R4B_StructureMapParameter | R5_StructureMapParameter"
    ],
):
    if len(parameters) != 1:
        raise ValueError("Reference transform requires exactly one parameter")
    if param := parameters[0].valueId:
        self.source = param
    else:
        raise ValueError("Reference transform parameter must be of type Id")

process

process(scope: MappingScope) -> Any

Transforms a FHIR resource reference by extracting its resource type and ID from the given source.

Parameters:

Name Type Description Default
scope MappingScope

The current mapping scope containing context and instances.

required

Returns:

Name Type Description
str Any

A string in the format "ResourceType/ResourceId" representing the FHIR reference.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def process(self, scope: "MappingScope") -> Any:
    """
    Transforms a FHIR resource reference by extracting its resource type and ID from the given source.

    Args:
        scope (MappingScope): The current mapping scope containing context and instances.

    Returns:
        str: A string in the format "ResourceType/ResourceId" representing the FHIR reference.

    Raises:
        Any exception raised by `scope.resolve_fhirpath` or the FHIRPath evaluation methods if the resource type or ID cannot be resolved.
    """
    source_fhirpath = scope.resolve_fhirpath(self.source)
    resource_type = (
        source_fhirpath._invoke(fp.Element("resourceType")).single(
            scope.get_instances()
        )
        or source_fhirpath.single(scope.get_instances()).__class__.__name__
    )
    resource_id = source_fhirpath._invoke(fp.Element("id")).single(
        scope.get_instances()
    )
    return f"{resource_type}/{resource_id}"

Translate

Path: fhircraft.fhir.mapper.engine.transforms.Translate

Bases: MappingTransform

Implements the 'translate' transform, which translates a code using a concept map.

Methods:

Name Description
process

Translates a source code using a specified FHIR ConceptMap.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def __init__(
    self,
    parameters: Sequence[
        "R4_StructureMapParameter | R4B_StructureMapParameter | R5_StructureMapParameter"
    ],
):
    if len(parameters) not in [2, 3]:
        raise ValueError(
            "Translate transform requires exactly two or three parameters"
        )
    self.source = parameters[0].value
    self.concept_map_name = parameters[1].value
    if len(parameters) == 3:
        self.output_type = parameters[2].value
        if self.output_type != "code":
            raise NotImplementedError(
                f"Output mode '{self.output_type}' for translate operation is not yet supported."
            )

process

process(scope: MappingScope) -> Any

Translates a source code using a specified FHIR ConceptMap.

Parameters:

Name Type Description Default
scope MappingScope

The current mapping scope, providing access to FHIRPath resolution and ConceptMaps.

required

Returns:

Name Type Description
str Any

The translated target code from the ConceptMap.

Raises:

Type Description
MappingError

If the ConceptMap has no groups, if a target code is not defined for the source code, or if the source code cannot be mapped using the ConceptMap.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def process(self, scope: "MappingScope") -> Any:
    """
    Translates a source code using a specified FHIR ConceptMap.

    Args:
        scope (MappingScope): The current mapping scope, providing access to FHIRPath resolution and ConceptMaps.

    Returns:
        str: The translated target code from the ConceptMap.

    Raises:
        MappingError: If the ConceptMap has no groups, if a target code is not defined for the source code,
                      or if the source code cannot be mapped using the ConceptMap.
    """
    source_code = scope.resolve_fhirpath(self.source).single(scope.get_instances())
    concept_map = scope.get_concept_map(self.concept_map_name.lstrip("#"))
    if concept_map.group is None:
        raise MappingError(
            f"Concept map '{self.concept_map_name}' has no groups defined."
        )
    for group in concept_map.group:
        for element in group.element or []:
            if element.target is None:
                continue
            for element_target in element.target:
                if element.code == source_code:
                    if self.output_type == "code":
                        if element_target.code is None:
                            raise MappingError(
                                f"Concept map '{self.concept_map_name}' does not define a target code for source code '{source_code}'."
                            )
                        return element_target.code
    else:
        raise MappingError(
            f"Could not map source code '{source_code}' using concept map '{self.concept_map_name}'."
        )

Truncate

Path: fhircraft.fhir.mapper.engine.transforms.Truncate

Bases: MappingTransform

Implements the 'truncate' transform, which truncates a string to a specified length.

Methods:

Name Description
process

Truncates the value obtained from a FHIRPath expression to a specified length.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def __init__(
    self,
    parameters: Sequence[
        "R4_StructureMapParameter | R4B_StructureMapParameter | R5_StructureMapParameter"
    ],
):
    if len(parameters) != 2:
        raise ValueError("Truncate transform requires exactly two parameters")
    self.source = parameters[0].value
    self.length = int(parameters[1].value)

process

process(scope: MappingScope) -> Any

Truncates the value obtained from a FHIRPath expression to a specified length.

Parameters:

Name Type Description Default
scope MappingScope

The current mapping scope containing context and data.

required

Returns:

Name Type Description
str Any

The truncated string result from the FHIRPath expression.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def process(self, scope: "MappingScope") -> Any:
    """
    Truncates the value obtained from a FHIRPath expression to a specified length.

    Args:
        scope (MappingScope): The current mapping scope containing context and data.

    Returns:
        str: The truncated string result from the FHIRPath expression.
    """
    source_fhirpath = scope.resolve_fhirpath(self.source)
    return source_fhirpath._invoke(fp.Substring(0, int(self.length))).single(
        scope.get_instances()
    )

UUID

Path: fhircraft.fhir.mapper.engine.transforms.UUID

Bases: MappingTransform

Implements the 'uuid' transform, which generates a UUID string.

Methods:

Name Description
process

Generates a new UUID string.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def __init__(
    self,
    parameters: Sequence[
        "R4_StructureMapParameter | R4B_StructureMapParameter | R5_StructureMapParameter"
    ],
):
    if len(parameters) != 0:
        raise ValueError("UUID transform does not take any parameters")

process

process(scope: MappingScope) -> Any

Generates a new UUID string.

Parameters:

Name Type Description Default
scope MappingScope

The current mapping scope (unused in this function).

required

Returns:

Name Type Description
str Any

A newly generated UUID as a string.

Source code in fhircraft/fhir/mapper/engine/transforms.py
def process(self, scope: "MappingScope") -> Any:
    """
    Generates a new UUID string.

    Args:
        scope (MappingScope): The current mapping scope (unused in this function).

    Returns:
        str: A newly generated UUID as a string.
    """
    return str(uuid.uuid4())