def _extract_resource_plan_as_dataframe()

in src/sagemaker_core/tools/resources_extractor.py [0:0]

• 85 lines of code
• 23 McCabe index (conditional complexity)

    def _extract_resource_plan_as_dataframe(self):
        """
        Builds a DataFrame containing resource information.

        Returns:
            None
        """
        self.df = pd.DataFrame(
            columns=[
                "resource_name",
                "type",
                "class_methods",
                "object_methods",
                "chain_resource_name",
                "additional_methods",
                "raw_actions",
                "resource_status_chain",
                "resource_states",
            ]
        )

        for resource, actions in sorted(self.resource_actions.items()):
            class_methods = set()
            object_methods = set()
            additional_methods = set()
            chain_resource_names = set()
            resource_status_chain = set()
            resource_states = set()

            for action in actions:
                action_low = action.lower()
                resource_low = resource.lower()

                if action_low.split(resource_low)[0] == "describe":
                    class_methods.add("get")
                    object_methods.add("refresh")

                    output_shape_name = self.operations[action]["output"]["shape"]
                    output_members_data = self.shapes[output_shape_name]["members"]

                    resource_status_chain, resource_states = self.get_status_chain_and_states(
                        resource
                    )

                    if resource_low.endswith("job") or resource_low.endswith("jobv2"):
                        object_methods.add("wait")
                    elif resource_states and resource_low != "action":
                        object_methods.add("wait_for_status")

                    if "Deleting" in resource_states or "DELETING" in resource_states:
                        object_methods.add("wait_for_delete")

                    continue

                if action_low.split(resource_low)[0] == "create":
                    shape_name = self.operations[action]["input"]["shape"]
                    input = self.shapes[shape_name]
                    for member in input["members"]:
                        if member.endswith("Name") or member.endswith("Names"):
                            chain_resource_name = member[: -len("Name")]

                            if (
                                chain_resource_name != resource
                                and chain_resource_name in self.resources
                            ):
                                chain_resource_names.add(chain_resource_name)
                action_split = action_low.split(resource_low)
                if action_split[0] == "invoke":
                    if not action_split[1]:
                        invoke_method = "invoke"
                    elif action_split[1] == "async":
                        invoke_method = "invoke_async"
                    else:
                        invoke_method = "invoke_with_response_stream"
                    object_methods.add(invoke_method)
                elif action_split[0] in CLASS_METHODS:
                    if action_low.split(resource_low)[0] == "list":
                        class_methods.add("get_all")
                    else:
                        class_methods.add(action_low.split(resource_low)[0])
                elif action_split[0] in OBJECT_METHODS:
                    object_methods.add(action_split[0])
                else:
                    additional_methods.add(action)

            if resource in self.RESOURCE_TO_ADDITIONAL_METHODS:
                additional_methods.update(self.RESOURCE_TO_ADDITIONAL_METHODS[resource])

            new_row = pd.DataFrame(
                {
                    "resource_name": [resource],
                    "type": ["resource"],
                    "class_methods": [list(sorted(class_methods))],
                    "object_methods": [list(sorted(object_methods))],
                    "chain_resource_name": [list(sorted(chain_resource_names))],
                    "additional_methods": [list(sorted(additional_methods))],
                    "raw_actions": [list(sorted(actions))],
                    "resource_status_chain": [list(resource_status_chain)],
                    "resource_states": [list(resource_states)],
                }
            )

            self.df = pd.concat([self.df, new_row], ignore_index=True)

        self.df.to_csv("resource_plan.csv", index=False)