def __next__()

in src/sagemaker_core/main/utils.py [0:0]

• 35 lines of code
• 17 McCabe index (conditional complexity)

    def __next__(self) -> T:

        # If there are summaries in the summary_list, return the next summary
        if len(self.summary_list) > 0 and self.index < len(self.summary_list):
            # Get the next summary from the resource summary_list
            summary = self.summary_list[self.index]
            self.index += 1

            # Initialize the resource object
            if is_primitive_class(self.resource_cls):
                # If the resource class is a primitive class, there will be only one element in the summary
                resource_object = list(summary.values())[0]
            else:
                # Transform the resource summary into format to initialize object
                init_data = transform(summary, self.summary_name)

                if self.custom_key_mapping:
                    init_data = {self.custom_key_mapping.get(k, k): v for k, v in init_data.items()}

                # Filter out the fields that are not in the resource class
                fields = self.resource_cls.__annotations__
                init_data = {k: v for k, v in init_data.items() if k in fields}

                resource_object = self.resource_cls(**init_data)

            # If the resource object has refresh method, refresh and return it
            if hasattr(resource_object, "refresh"):
                resource_object.refresh()
            return resource_object

        # If index reached the end of summary_list, and there is no next token, raise StopIteration
        elif (
            len(self.summary_list) > 0
            and self.index >= len(self.summary_list)
            and self.next_token is None
        ):
            raise StopIteration

        # Otherwise, get the next page of summaries by calling the list method with the next token if available
        else:
            if self.next_token is not None:
                response = getattr(self.client, self.list_method)(
                    NextToken=self.next_token, **self.list_method_kwargs
                )
            else:
                response = getattr(self.client, self.list_method)(**self.list_method_kwargs)

            self.summary_list = response.get(self.summaries_key, [])
            self.next_token = response.get("NextToken", None)
            self.index = 0

            # If list_method returned an empty list, raise StopIteration
            if len(self.summary_list) == 0:
                raise StopIteration

            return self.__next__()