Skip to content

split

trestle.core.commands.split ¤

Trestle Split Command.

logger ¤

trace ¤

Classes¤

AliasTracker (TrestleBaseModel) pydantic-model ¤

Convenience class to track writing out of models.

Source code in trestle/core/commands/split.py
class AliasTracker(TrestleBaseModel):
    """Convenience class to track writing out of models."""

    # This tracks the parts that need to be split from each element
    # and makes sure it is written out once

    aliases: List[str]
    written: bool = False

    def add_alias(self, alias: str) -> None:
        """Add alias."""
        if alias not in self.aliases:
            self.aliases.append(alias)

    def get_aliases(self) -> List[str]:
        """Get the list of aliases."""
        return self.aliases

    def needs_writing(self) -> bool:
        """Need to write the model."""
        return not self.written

    def mark_written(self) -> None:
        """Mark this model as written."""
        self.written = True
aliases: List[str] pydantic-field required ¤
written: bool pydantic-field ¤
Methods¤
add_alias(self, alias) ¤

Add alias.

Source code in trestle/core/commands/split.py
def add_alias(self, alias: str) -> None:
    """Add alias."""
    if alias not in self.aliases:
        self.aliases.append(alias)
get_aliases(self) ¤

Get the list of aliases.

Source code in trestle/core/commands/split.py
def get_aliases(self) -> List[str]:
    """Get the list of aliases."""
    return self.aliases
mark_written(self) ¤

Mark this model as written.

Source code in trestle/core/commands/split.py
def mark_written(self) -> None:
    """Mark this model as written."""
    self.written = True
needs_writing(self) ¤

Need to write the model.

Source code in trestle/core/commands/split.py
def needs_writing(self) -> bool:
    """Need to write the model."""
    return not self.written

SplitCmd (CommandPlusDocs) ¤

Split subcomponents on a trestle model.

Source code in trestle/core/commands/split.py
class SplitCmd(CommandPlusDocs):
    """Split subcomponents on a trestle model."""

    name = 'split'

    def _init_arguments(self) -> None:
        self.add_argument(
            f'-{const.ARG_FILE_SHORT}', f'--{const.ARG_FILE}', help=const.ARG_DESC_FILE + ' to split.', required=False
        )
        self.add_argument(
            f'-{const.ARG_ELEMENT_SHORT}',
            f'--{const.ARG_ELEMENT}',
            help=const.ARG_DESC_ELEMENT + ' to split.',
            required=False
        )

    def _run(self, args: argparse.Namespace) -> int:
        """Split an OSCAL file into elements."""
        try:
            log.set_log_level_from_args(args)
            trace.log('Entering trestle split.')
            # get the Model
            args_raw: Dict[str, str] = args.__dict__

            # remove any quotes passed in as on windows platforms
            elements_clean: str = args_raw[const.ARG_ELEMENT].strip("'")

            file_name = ''
            file_name = '' if not args_raw[const.ARG_FILE] else args_raw[const.ARG_FILE]
            # cwd must be in the model directory if file to split is not specified
            effective_cwd = pathlib.Path.cwd()

            return self.perform_split(effective_cwd, file_name, elements_clean, args.trestle_root)
        except Exception as e:  # pragma: no cover
            return handle_generic_command_exception(e, logger, 'Error while performing a split operation')

    @classmethod
    def perform_split(
        cls, effective_cwd: pathlib.Path, file_name: str, elements: str, trestle_root: pathlib.Path
    ) -> int:
        """Perform the split operation.

        Args:
            effective_cwd: effective directory in which the the split operation is performed
            file_name: file name of model to split, or '' if deduced from elements and cwd
            elements: comma separated list of paths to strip from the file, with quotes removed

        Returns:
            0 on success and 1 on failure
        """
        file_path_list: List[Tuple[str, str]] = []

        if file_name:
            file_path_list.append((file_name, elements))
        else:
            # cwd must be in the model directory if file to split is not specified
            # find top directory for this model based on trestle root and cwd
            model_dir = file_utils.extract_project_model_path(effective_cwd)
            if model_dir is None:
                raise TrestleError('Current directory must be within a model directory if file is not specified')

            content_type: FileContentType = FileContentType.dir_to_content_type(model_dir)

            # determine the file needed for each split path
            element_paths = elements.split(',')
            for path in element_paths:
                element_path = ElementPath(path)
                # if element path is relative use directory context to determine absolute path
                element_path.make_absolute(model_dir, effective_cwd)
                file_path = element_path.find_last_file_in_path(content_type, model_dir)
                # now make the element path relative to the model file to be loaded
                if file_path is None or element_path.make_relative(file_path.relative_to(model_dir)) != 0:
                    raise TrestleError(f'Unable to match element path with files in model directory {element_path}')

                file_path_list.append((file_path, element_path.to_string()))

        # match paths to corresponding files since several paths may be split from the same file
        file_path_dict: Dict[str, str] = {}
        for file_path in file_path_list:
            key = file_path[0]
            path = file_path[1]
            if key not in file_path_dict:
                file_path_dict[key] = path
            else:
                current_path = file_path_dict[key]
                file_path_dict[key] = f'{current_path},{path}'

        for raw_file_name, element_path in file_path_dict.items():
            file_path = file_utils.relative_resolve(pathlib.Path(raw_file_name), effective_cwd)
            # this makes assumptions that the path is relative.
            if not file_path.exists():
                raise TrestleError(f'File {file_path} does not exist.')

            content_type = FileContentType.to_content_type(file_path.suffix)

            # find the base directory of the file
            base_dir = file_path.parent
            model_type, _ = ModelUtils.get_stripped_model_type(file_path, trestle_root)

            model: OscalBaseModel = model_type.oscal_read(file_path)

            if cmd_utils.split_is_too_fine(element_path, model):
                raise TrestleError('Cannot split the model to the level of uuids, strings, etc.')

            # use the model itself to resolve any wildcards and create list of element paths
            logger.debug(f'split calling parse_element_args on {element_path}')
            # use contextual mode to parse

            element_paths: List[ElementPath] = cmd_utils.parse_element_args(
                model, element_path.split(','), base_dir.relative_to(trestle_root)
            )

            # analyze the split tree and determine which aliases should be stripped from each file
            aliases_to_strip = cls.find_aliases_to_strip(element_paths)

            # need the file name relative to the base directory
            file_name_no_path = str(file_path.name)

            split_plan = cls.split_model(
                model, element_paths, base_dir, content_type, file_name_no_path, aliases_to_strip
            )
            trash.store(file_path, True)

            try:
                split_plan.execute()
            except Exception as e:
                trash.recover(file_path, True)
                raise TrestleError(f'Split has failed with error: {e}.')

        return CmdReturnCodes.SUCCESS.value

    @classmethod
    def prepare_sub_model_split_actions(
        cls,
        sub_model_item: OscalBaseModel,
        sub_model_dir: pathlib.Path,
        file_prefix: str,
        content_type: FileContentType
    ) -> List[Action]:
        """Create split actions of sub model."""
        actions: List[Action] = []
        file_name = cmd_utils.to_model_file_name(sub_model_item, file_prefix, content_type)
        model_type = classname_to_alias(type(sub_model_item).__name__, AliasMode.JSON)
        sub_model_file = sub_model_dir / file_name
        actions.append(CreatePathAction(sub_model_file))
        actions.append(WriteFileAction(sub_model_file, Element(sub_model_item, model_type), content_type))
        return actions

    @classmethod
    def split_model_at_path_chain(
        cls,
        model_obj: OscalBaseModel,
        element_paths: List[ElementPath],
        base_dir: pathlib.Path,
        content_type: FileContentType,
        cur_path_index: int,
        split_plan: Plan,
        strip_root: bool,
        root_file_name: str,
        aliases_to_strip: Dict[str, AliasTracker],
        last_one: bool = True
    ) -> int:
        """Recursively split the model at the provided chain of element paths.

        It assumes that a chain of element paths starts at the cur_path_index with the first path ending
        with a wildcard (*)

        If the wildcard follows an element that is inherently a list of items, the list of items is extracted.
        But if the wildcard follows a generic model than members of that model class found in the model will be
        split off.  But only the non-trivial elements are removed, i.e. not str, int, datetime, etc.

        Args:
            model_obj: The OscalBaseModel to be split
            element_paths: The List[ElementPath] of elements to split, including embedded wildcards
            base_dir: pathlib.Path of the file being split
            content_type: json or yaml files
            cur_path_index: Index into the list of element paths for the current split operation
            split_plan: The accumulated plan of actions needed to perform the split
            strip_root: Whether to strip elements from the root object
            root_file_name: Filename of root file that gets split into a list of items
            aliases_to_strip: AliasTracker previously loaded with aliases that need to be split from each element
            last_one: bool indicating last item in array has been split and stripped model can now be written

        Returns:
            int representing the index where the chain of the path ends.

        Examples:
            For example, element paths could have a list of paths as below for a `ComponentDefinition` model where
            the first path is the start of the chain.

            For each of the sub model described by the first element path (e.g component-defintion.components.*) in the
            chain, the subsequent paths (e.g component.control-implementations.*) will be applied recursively
            to retrieve the sub-sub models:
            [
                'component-definition.component.*',
                'component.control-implementations.*'
            ]
            for a command like below:
            trestle split -f component.yaml -e component-definition.components.*.control-implementations.*
        """
        if split_plan is None:
            raise TrestleError('Split plan must have been initialized')

        if cur_path_index < 0:
            raise TrestleError('Current index of the chain of paths cannot be less than 0')

        # if there are no more element_paths, return the current plan
        if cur_path_index >= len(element_paths):
            return cur_path_index

        # initialize local variables
        element = Element(model_obj)
        stripped_field_alias: List[str] = []

        # get the sub_model specified by the element_path of this round
        element_path = element_paths[cur_path_index]

        # does the next element_path point back at me
        is_parent = cur_path_index + 1 < len(element_paths) and element_paths[cur_path_index
                                                                              + 1].get_parent() == element_path

        # root dir name for sub models dir
        # 00000__group.json will have the root_dir name as 00000__group for sub models of group
        # catalog.json will have the root_dir name as catalog
        root_dir = ''
        if root_file_name != '':
            root_dir = str(pathlib.Path(root_file_name).with_suffix(''))

        sub_models = element.get_at(element_path, False)  # we call sub_models as in plural, but it can be just one

        # assume cur_path_index is the end of the chain
        # value of this variable may change during recursive split of the sub-models below
        path_chain_end = cur_path_index

        # if wildcard is present in the element_path and the next path in the chain has current path as the parent,
        # Then deal with case of list, or split of arbitrary oscalbasemodel
        if is_parent and element_path.get_last() is not ElementPath.WILDCARD:
            # create dir for all sub model items
            sub_models_dir = base_dir / element_path.to_root_path()
            sub_model_plan = Plan()
            path_chain_end = cls.split_model_at_path_chain(
                sub_models,
                element_paths,
                sub_models_dir,
                content_type,
                cur_path_index + 1,
                sub_model_plan,
                True,
                '',
                aliases_to_strip
            )
            sub_model_actions = sub_model_plan.get_actions()
            split_plan.add_actions(sub_model_actions)
        elif element_path.get_last() == ElementPath.WILDCARD:
            # extract sub-models into a dict with appropriate prefix
            sub_model_items: Dict[str, OscalBaseModel] = {}
            sub_models_dir = base_dir / element_path.to_file_path(root_dir=root_dir)
            if isinstance(sub_models, list):
                for i, sub_model_item in enumerate(sub_models):
                    # e.g. `groups/00000_groups/`
                    prefix = str(i).zfill(const.FILE_DIGIT_PREFIX_LENGTH)
                    sub_model_items[prefix] = sub_model_item

            # process list sub model items
            count = 0
            for key, sub_model_item in sub_model_items.items():
                count += 1
                # recursively split the sub-model if there are more element paths to traverse
                # e.g. split component.control-implementations.*
                require_recursive_split = cur_path_index + 1 < len(element_paths) and element_paths[
                    cur_path_index + 1].get_parent() == element_path

                if require_recursive_split:
                    # prepare individual directory for each sub-model
                    sub_root_file_name = cmd_utils.to_model_file_name(sub_model_item, key, content_type)
                    sub_model_plan = Plan()

                    last_one: bool = count == len(sub_model_items)
                    path_chain_end = cls.split_model_at_path_chain(
                        sub_model_item,
                        element_paths,
                        sub_models_dir,
                        content_type,
                        cur_path_index + 1,
                        sub_model_plan,
                        True,
                        sub_root_file_name,
                        aliases_to_strip,
                        last_one
                    )
                    sub_model_actions = sub_model_plan.get_actions()
                else:
                    sub_model_actions = cls.prepare_sub_model_split_actions(
                        sub_model_item, sub_models_dir, key, content_type
                    )

                split_plan.add_actions(sub_model_actions)
        else:
            # the chain of path ends at the current index.
            # so no recursive call. Let's just write the sub model to the file and get out
            if sub_models is not None:
                sub_model_file = base_dir / element_path.to_file_path(content_type, root_dir=root_dir)
                split_plan.add_action(CreatePathAction(sub_model_file))
                split_plan.add_action(
                    WriteFileAction(sub_model_file, Element(sub_models, element_path.get_element_name()), content_type)
                )

        # Strip the root model and add a WriteAction for the updated model object in the plan
        if strip_root:
            full_path = element_path.get_full()
            path = '.'.join(full_path.split('.')[:-1])
            aliases = [element_path.get_element_name()]
            need_to_write = True
            use_alias_dict = aliases_to_strip is not None and path in aliases_to_strip
            if use_alias_dict:
                aliases = aliases_to_strip[path].get_aliases()
                need_to_write = aliases_to_strip[path].needs_writing()

            stripped_model = model_obj.stripped_instance(stripped_fields_aliases=aliases)
            # can mark it written even if it doesn't need writing since it is empty
            # but if an array only mark it written if it's the last one
            if last_one and use_alias_dict:
                aliases_to_strip[path].mark_written()
            # If it's an empty model after stripping the fields, don't create path and don't write
            field_list = [x for x in model_obj.__fields__.keys() if model_obj.__fields__[x] is not None]
            if set(field_list) == set(stripped_field_alias):
                return path_chain_end

            if need_to_write:
                if root_file_name != '':
                    root_file = base_dir / root_file_name
                else:
                    root_file = base_dir / element_path.to_root_path(content_type)

                split_plan.add_action(CreatePathAction(root_file))
                wrapper_alias = classname_to_alias(stripped_model.__class__.__name__, AliasMode.JSON)
                split_plan.add_action(WriteFileAction(root_file, Element(stripped_model, wrapper_alias), content_type))

        # return the end of the current path chain
        return path_chain_end

    @classmethod
    def split_model(
        cls,
        model_obj: OscalBaseModel,
        element_paths: List[ElementPath],
        base_dir: pathlib.Path,
        content_type: FileContentType,
        root_file_name: str,
        aliases_to_strip: Dict[str, AliasTracker]
    ) -> Plan:
        """Split the model at the provided element paths.

        It returns a plan for the operation
        """
        # initialize plan
        split_plan = Plan()

        # loop through the element path list and update the split_plan
        stripped_field_alias = []
        cur_path_index = 0
        while cur_path_index < len(element_paths):
            # extract the sub element name for each of the root path of the path chain
            element_path = element_paths[cur_path_index]

            if element_path.get_parent() is None and len(element_path.get()) > 1:
                stripped_part = element_path.get()[1]
                if stripped_part == ElementPath.WILDCARD:
                    stripped_field_alias.append('__root__')
                else:
                    if stripped_part not in stripped_field_alias:
                        stripped_field_alias.append(stripped_part)

            # split model at the path chain
            cur_path_index = cls.split_model_at_path_chain(
                model_obj,
                element_paths,
                base_dir,
                content_type,
                cur_path_index,
                split_plan,
                False,
                root_file_name,
                aliases_to_strip
            )

            cur_path_index += 1

        # strip the root model object and add a WriteAction
        stripped_root = model_obj.stripped_instance(stripped_fields_aliases=stripped_field_alias)
        # If it's an empty model after stripping the fields, don't create path and don't write
        if set(model_obj.__fields__.keys()) == set(stripped_field_alias):
            return split_plan
        if root_file_name != '':
            root_file = base_dir / root_file_name
        else:
            root_file = base_dir / element_paths[0].to_root_path(content_type)
        split_plan.add_action(CreatePathAction(root_file, True))
        wrapper_alias = classname_to_alias(stripped_root.__class__.__name__, AliasMode.JSON)
        split_plan.add_action(WriteFileAction(root_file, Element(stripped_root, wrapper_alias), content_type))

        return split_plan

    @classmethod
    def find_aliases_to_strip(cls, element_paths: List[ElementPath]) -> Dict[str, AliasTracker]:
        """Find list of aliases that need to be stripped as each element written out."""
        # A given path may be present in several split actions
        # Need to determine all parts stripped at each node in order to strip them all and
        # write the stripped model only once
        tracker_map: Dict[str, AliasTracker] = {}
        for element_path in element_paths:
            path = element_path.get_full()
            path_parts = path.split('.')
            alias = path_parts[-1]
            if len(path_parts) > 2 and alias != '*':
                root_path = '.'.join(path_parts[:-1])
                if root_path in tracker_map:
                    tracker_map[root_path].add_alias(alias)
                else:
                    tracker_map[root_path] = AliasTracker(aliases=[alias])
        return tracker_map
name ¤
Methods¤
find_aliases_to_strip(element_paths) classmethod ¤

Find list of aliases that need to be stripped as each element written out.

Source code in trestle/core/commands/split.py
@classmethod
def find_aliases_to_strip(cls, element_paths: List[ElementPath]) -> Dict[str, AliasTracker]:
    """Find list of aliases that need to be stripped as each element written out."""
    # A given path may be present in several split actions
    # Need to determine all parts stripped at each node in order to strip them all and
    # write the stripped model only once
    tracker_map: Dict[str, AliasTracker] = {}
    for element_path in element_paths:
        path = element_path.get_full()
        path_parts = path.split('.')
        alias = path_parts[-1]
        if len(path_parts) > 2 and alias != '*':
            root_path = '.'.join(path_parts[:-1])
            if root_path in tracker_map:
                tracker_map[root_path].add_alias(alias)
            else:
                tracker_map[root_path] = AliasTracker(aliases=[alias])
    return tracker_map
perform_split(effective_cwd, file_name, elements, trestle_root) classmethod ¤

Perform the split operation.

Parameters:

Name Type Description Default
effective_cwd Path

effective directory in which the the split operation is performed

required
file_name str

file name of model to split, or '' if deduced from elements and cwd

required
elements str

comma separated list of paths to strip from the file, with quotes removed

required

Returns:

Type Description
int

0 on success and 1 on failure

Source code in trestle/core/commands/split.py
@classmethod
def perform_split(
    cls, effective_cwd: pathlib.Path, file_name: str, elements: str, trestle_root: pathlib.Path
) -> int:
    """Perform the split operation.

    Args:
        effective_cwd: effective directory in which the the split operation is performed
        file_name: file name of model to split, or '' if deduced from elements and cwd
        elements: comma separated list of paths to strip from the file, with quotes removed

    Returns:
        0 on success and 1 on failure
    """
    file_path_list: List[Tuple[str, str]] = []

    if file_name:
        file_path_list.append((file_name, elements))
    else:
        # cwd must be in the model directory if file to split is not specified
        # find top directory for this model based on trestle root and cwd
        model_dir = file_utils.extract_project_model_path(effective_cwd)
        if model_dir is None:
            raise TrestleError('Current directory must be within a model directory if file is not specified')

        content_type: FileContentType = FileContentType.dir_to_content_type(model_dir)

        # determine the file needed for each split path
        element_paths = elements.split(',')
        for path in element_paths:
            element_path = ElementPath(path)
            # if element path is relative use directory context to determine absolute path
            element_path.make_absolute(model_dir, effective_cwd)
            file_path = element_path.find_last_file_in_path(content_type, model_dir)
            # now make the element path relative to the model file to be loaded
            if file_path is None or element_path.make_relative(file_path.relative_to(model_dir)) != 0:
                raise TrestleError(f'Unable to match element path with files in model directory {element_path}')

            file_path_list.append((file_path, element_path.to_string()))

    # match paths to corresponding files since several paths may be split from the same file
    file_path_dict: Dict[str, str] = {}
    for file_path in file_path_list:
        key = file_path[0]
        path = file_path[1]
        if key not in file_path_dict:
            file_path_dict[key] = path
        else:
            current_path = file_path_dict[key]
            file_path_dict[key] = f'{current_path},{path}'

    for raw_file_name, element_path in file_path_dict.items():
        file_path = file_utils.relative_resolve(pathlib.Path(raw_file_name), effective_cwd)
        # this makes assumptions that the path is relative.
        if not file_path.exists():
            raise TrestleError(f'File {file_path} does not exist.')

        content_type = FileContentType.to_content_type(file_path.suffix)

        # find the base directory of the file
        base_dir = file_path.parent
        model_type, _ = ModelUtils.get_stripped_model_type(file_path, trestle_root)

        model: OscalBaseModel = model_type.oscal_read(file_path)

        if cmd_utils.split_is_too_fine(element_path, model):
            raise TrestleError('Cannot split the model to the level of uuids, strings, etc.')

        # use the model itself to resolve any wildcards and create list of element paths
        logger.debug(f'split calling parse_element_args on {element_path}')
        # use contextual mode to parse

        element_paths: List[ElementPath] = cmd_utils.parse_element_args(
            model, element_path.split(','), base_dir.relative_to(trestle_root)
        )

        # analyze the split tree and determine which aliases should be stripped from each file
        aliases_to_strip = cls.find_aliases_to_strip(element_paths)

        # need the file name relative to the base directory
        file_name_no_path = str(file_path.name)

        split_plan = cls.split_model(
            model, element_paths, base_dir, content_type, file_name_no_path, aliases_to_strip
        )
        trash.store(file_path, True)

        try:
            split_plan.execute()
        except Exception as e:
            trash.recover(file_path, True)
            raise TrestleError(f'Split has failed with error: {e}.')

    return CmdReturnCodes.SUCCESS.value
prepare_sub_model_split_actions(sub_model_item, sub_model_dir, file_prefix, content_type) classmethod ¤

Create split actions of sub model.

Source code in trestle/core/commands/split.py
@classmethod
def prepare_sub_model_split_actions(
    cls,
    sub_model_item: OscalBaseModel,
    sub_model_dir: pathlib.Path,
    file_prefix: str,
    content_type: FileContentType
) -> List[Action]:
    """Create split actions of sub model."""
    actions: List[Action] = []
    file_name = cmd_utils.to_model_file_name(sub_model_item, file_prefix, content_type)
    model_type = classname_to_alias(type(sub_model_item).__name__, AliasMode.JSON)
    sub_model_file = sub_model_dir / file_name
    actions.append(CreatePathAction(sub_model_file))
    actions.append(WriteFileAction(sub_model_file, Element(sub_model_item, model_type), content_type))
    return actions
split_model(model_obj, element_paths, base_dir, content_type, root_file_name, aliases_to_strip) classmethod ¤

Split the model at the provided element paths.

It returns a plan for the operation

Source code in trestle/core/commands/split.py
@classmethod
def split_model(
    cls,
    model_obj: OscalBaseModel,
    element_paths: List[ElementPath],
    base_dir: pathlib.Path,
    content_type: FileContentType,
    root_file_name: str,
    aliases_to_strip: Dict[str, AliasTracker]
) -> Plan:
    """Split the model at the provided element paths.

    It returns a plan for the operation
    """
    # initialize plan
    split_plan = Plan()

    # loop through the element path list and update the split_plan
    stripped_field_alias = []
    cur_path_index = 0
    while cur_path_index < len(element_paths):
        # extract the sub element name for each of the root path of the path chain
        element_path = element_paths[cur_path_index]

        if element_path.get_parent() is None and len(element_path.get()) > 1:
            stripped_part = element_path.get()[1]
            if stripped_part == ElementPath.WILDCARD:
                stripped_field_alias.append('__root__')
            else:
                if stripped_part not in stripped_field_alias:
                    stripped_field_alias.append(stripped_part)

        # split model at the path chain
        cur_path_index = cls.split_model_at_path_chain(
            model_obj,
            element_paths,
            base_dir,
            content_type,
            cur_path_index,
            split_plan,
            False,
            root_file_name,
            aliases_to_strip
        )

        cur_path_index += 1

    # strip the root model object and add a WriteAction
    stripped_root = model_obj.stripped_instance(stripped_fields_aliases=stripped_field_alias)
    # If it's an empty model after stripping the fields, don't create path and don't write
    if set(model_obj.__fields__.keys()) == set(stripped_field_alias):
        return split_plan
    if root_file_name != '':
        root_file = base_dir / root_file_name
    else:
        root_file = base_dir / element_paths[0].to_root_path(content_type)
    split_plan.add_action(CreatePathAction(root_file, True))
    wrapper_alias = classname_to_alias(stripped_root.__class__.__name__, AliasMode.JSON)
    split_plan.add_action(WriteFileAction(root_file, Element(stripped_root, wrapper_alias), content_type))

    return split_plan
split_model_at_path_chain(model_obj, element_paths, base_dir, content_type, cur_path_index, split_plan, strip_root, root_file_name, aliases_to_strip, last_one=True) classmethod ¤

Recursively split the model at the provided chain of element paths.

It assumes that a chain of element paths starts at the cur_path_index with the first path ending with a wildcard (*)

If the wildcard follows an element that is inherently a list of items, the list of items is extracted. But if the wildcard follows a generic model than members of that model class found in the model will be split off. But only the non-trivial elements are removed, i.e. not str, int, datetime, etc.

Parameters:

Name Type Description Default
model_obj OscalBaseModel

The OscalBaseModel to be split

required
element_paths List[trestle.core.models.elements.ElementPath]

The List[ElementPath] of elements to split, including embedded wildcards

required
base_dir Path

pathlib.Path of the file being split

required
content_type FileContentType

json or yaml files

required
cur_path_index int

Index into the list of element paths for the current split operation

required
split_plan Plan

The accumulated plan of actions needed to perform the split

required
strip_root bool

Whether to strip elements from the root object

required
root_file_name str

Filename of root file that gets split into a list of items

required
aliases_to_strip Dict[str, trestle.core.commands.split.AliasTracker]

AliasTracker previously loaded with aliases that need to be split from each element

required
last_one bool

bool indicating last item in array has been split and stripped model can now be written

True

Returns:

Type Description
int

int representing the index where the chain of the path ends.

Examples:

For example, element paths could have a list of paths as below for a ComponentDefinition model where the first path is the start of the chain.

For each of the sub model described by the first element path (e.g component-defintion.components.) in the chain, the subsequent paths (e.g component.control-implementations.) will be applied recursively to retrieve the sub-sub models: [ 'component-definition.component.', 'component.control-implementations.' ] for a command like below: trestle split -f component.yaml -e component-definition.components..control-implementations.

Source code in trestle/core/commands/split.py
@classmethod
def split_model_at_path_chain(
    cls,
    model_obj: OscalBaseModel,
    element_paths: List[ElementPath],
    base_dir: pathlib.Path,
    content_type: FileContentType,
    cur_path_index: int,
    split_plan: Plan,
    strip_root: bool,
    root_file_name: str,
    aliases_to_strip: Dict[str, AliasTracker],
    last_one: bool = True
) -> int:
    """Recursively split the model at the provided chain of element paths.

    It assumes that a chain of element paths starts at the cur_path_index with the first path ending
    with a wildcard (*)

    If the wildcard follows an element that is inherently a list of items, the list of items is extracted.
    But if the wildcard follows a generic model than members of that model class found in the model will be
    split off.  But only the non-trivial elements are removed, i.e. not str, int, datetime, etc.

    Args:
        model_obj: The OscalBaseModel to be split
        element_paths: The List[ElementPath] of elements to split, including embedded wildcards
        base_dir: pathlib.Path of the file being split
        content_type: json or yaml files
        cur_path_index: Index into the list of element paths for the current split operation
        split_plan: The accumulated plan of actions needed to perform the split
        strip_root: Whether to strip elements from the root object
        root_file_name: Filename of root file that gets split into a list of items
        aliases_to_strip: AliasTracker previously loaded with aliases that need to be split from each element
        last_one: bool indicating last item in array has been split and stripped model can now be written

    Returns:
        int representing the index where the chain of the path ends.

    Examples:
        For example, element paths could have a list of paths as below for a `ComponentDefinition` model where
        the first path is the start of the chain.

        For each of the sub model described by the first element path (e.g component-defintion.components.*) in the
        chain, the subsequent paths (e.g component.control-implementations.*) will be applied recursively
        to retrieve the sub-sub models:
        [
            'component-definition.component.*',
            'component.control-implementations.*'
        ]
        for a command like below:
        trestle split -f component.yaml -e component-definition.components.*.control-implementations.*
    """
    if split_plan is None:
        raise TrestleError('Split plan must have been initialized')

    if cur_path_index < 0:
        raise TrestleError('Current index of the chain of paths cannot be less than 0')

    # if there are no more element_paths, return the current plan
    if cur_path_index >= len(element_paths):
        return cur_path_index

    # initialize local variables
    element = Element(model_obj)
    stripped_field_alias: List[str] = []

    # get the sub_model specified by the element_path of this round
    element_path = element_paths[cur_path_index]

    # does the next element_path point back at me
    is_parent = cur_path_index + 1 < len(element_paths) and element_paths[cur_path_index
                                                                          + 1].get_parent() == element_path

    # root dir name for sub models dir
    # 00000__group.json will have the root_dir name as 00000__group for sub models of group
    # catalog.json will have the root_dir name as catalog
    root_dir = ''
    if root_file_name != '':
        root_dir = str(pathlib.Path(root_file_name).with_suffix(''))

    sub_models = element.get_at(element_path, False)  # we call sub_models as in plural, but it can be just one

    # assume cur_path_index is the end of the chain
    # value of this variable may change during recursive split of the sub-models below
    path_chain_end = cur_path_index

    # if wildcard is present in the element_path and the next path in the chain has current path as the parent,
    # Then deal with case of list, or split of arbitrary oscalbasemodel
    if is_parent and element_path.get_last() is not ElementPath.WILDCARD:
        # create dir for all sub model items
        sub_models_dir = base_dir / element_path.to_root_path()
        sub_model_plan = Plan()
        path_chain_end = cls.split_model_at_path_chain(
            sub_models,
            element_paths,
            sub_models_dir,
            content_type,
            cur_path_index + 1,
            sub_model_plan,
            True,
            '',
            aliases_to_strip
        )
        sub_model_actions = sub_model_plan.get_actions()
        split_plan.add_actions(sub_model_actions)
    elif element_path.get_last() == ElementPath.WILDCARD:
        # extract sub-models into a dict with appropriate prefix
        sub_model_items: Dict[str, OscalBaseModel] = {}
        sub_models_dir = base_dir / element_path.to_file_path(root_dir=root_dir)
        if isinstance(sub_models, list):
            for i, sub_model_item in enumerate(sub_models):
                # e.g. `groups/00000_groups/`
                prefix = str(i).zfill(const.FILE_DIGIT_PREFIX_LENGTH)
                sub_model_items[prefix] = sub_model_item

        # process list sub model items
        count = 0
        for key, sub_model_item in sub_model_items.items():
            count += 1
            # recursively split the sub-model if there are more element paths to traverse
            # e.g. split component.control-implementations.*
            require_recursive_split = cur_path_index + 1 < len(element_paths) and element_paths[
                cur_path_index + 1].get_parent() == element_path

            if require_recursive_split:
                # prepare individual directory for each sub-model
                sub_root_file_name = cmd_utils.to_model_file_name(sub_model_item, key, content_type)
                sub_model_plan = Plan()

                last_one: bool = count == len(sub_model_items)
                path_chain_end = cls.split_model_at_path_chain(
                    sub_model_item,
                    element_paths,
                    sub_models_dir,
                    content_type,
                    cur_path_index + 1,
                    sub_model_plan,
                    True,
                    sub_root_file_name,
                    aliases_to_strip,
                    last_one
                )
                sub_model_actions = sub_model_plan.get_actions()
            else:
                sub_model_actions = cls.prepare_sub_model_split_actions(
                    sub_model_item, sub_models_dir, key, content_type
                )

            split_plan.add_actions(sub_model_actions)
    else:
        # the chain of path ends at the current index.
        # so no recursive call. Let's just write the sub model to the file and get out
        if sub_models is not None:
            sub_model_file = base_dir / element_path.to_file_path(content_type, root_dir=root_dir)
            split_plan.add_action(CreatePathAction(sub_model_file))
            split_plan.add_action(
                WriteFileAction(sub_model_file, Element(sub_models, element_path.get_element_name()), content_type)
            )

    # Strip the root model and add a WriteAction for the updated model object in the plan
    if strip_root:
        full_path = element_path.get_full()
        path = '.'.join(full_path.split('.')[:-1])
        aliases = [element_path.get_element_name()]
        need_to_write = True
        use_alias_dict = aliases_to_strip is not None and path in aliases_to_strip
        if use_alias_dict:
            aliases = aliases_to_strip[path].get_aliases()
            need_to_write = aliases_to_strip[path].needs_writing()

        stripped_model = model_obj.stripped_instance(stripped_fields_aliases=aliases)
        # can mark it written even if it doesn't need writing since it is empty
        # but if an array only mark it written if it's the last one
        if last_one and use_alias_dict:
            aliases_to_strip[path].mark_written()
        # If it's an empty model after stripping the fields, don't create path and don't write
        field_list = [x for x in model_obj.__fields__.keys() if model_obj.__fields__[x] is not None]
        if set(field_list) == set(stripped_field_alias):
            return path_chain_end

        if need_to_write:
            if root_file_name != '':
                root_file = base_dir / root_file_name
            else:
                root_file = base_dir / element_path.to_root_path(content_type)

            split_plan.add_action(CreatePathAction(root_file))
            wrapper_alias = classname_to_alias(stripped_model.__class__.__name__, AliasMode.JSON)
            split_plan.add_action(WriteFileAction(root_file, Element(stripped_model, wrapper_alias), content_type))

    # return the end of the current path chain
    return path_chain_end

handler: python