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

__class_vars__ special

__custom_root_type__ special

__doc__ special

__exclude_fields__ special

__fields__ special

__include_fields__ special

__post_root_validators__ special

__pre_root_validators__ special

__private_attributes__ special

__schema_cache__ special

__signature__ special

__slots__: Tuple[str, ...] special

__validators__ special

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]  # type: ignore
            path = file_path[1]  # type: ignore
            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]  # type: ignore
        path = file_path[1]  # type: ignore
        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