def ready()

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

• 33 lines of code
• 10 McCabe index (conditional complexity)

    def ready(self) -> bool:
        """
        Checks whether or not MultiLogStreamHandler is ready to serve new log events at this moment.

        If self.streams is already set, return True.
        Otherwise, check if the current number of log streams in the log group match the exptected stream count.

        Returns:
            bool: Whether or not MultiLogStreamHandler is ready to serve new log events.
        """

        if len(self.streams) >= self.expected_stream_count:
            return True

        try:
            response = self.cw_client.describe_log_streams(
                logGroupName=self.log_group_name,
                logStreamNamePrefix=self.log_stream_name_prefix + "/",
                orderBy="LogStreamName",
            )
            stream_names = [stream["logStreamName"] for stream in response["logStreams"]]

            next_token = response.get("nextToken")
            while next_token:
                response = self.cw_client.describe_log_streams(
                    logGroupName=self.log_group_name,
                    logStreamNamePrefix=self.log_stream_name_prefix + "/",
                    orderBy="LogStreamName",
                    nextToken=next_token,
                )
                stream_names.extend([stream["logStreamName"] for stream in response["logStreams"]])
                next_token = response.get("nextToken", None)

            if len(stream_names) >= self.expected_stream_count:
                self.streams = [
                    LogStreamHandler(self.log_group_name, log_stream_name, index)
                    for index, log_stream_name in enumerate(stream_names)
                ]

                return True
            else:
                # Log streams are created whenever a container starts writing to stdout/err,
                # so if the stream count is less than the expected number, return False
                return False

        except botocore.exceptions.ClientError as e:
            # On the very first training job run on an account, there's no log group until
            # the container starts logging, so ignore any errors thrown about that
            if e.response["Error"]["Code"] == "ResourceNotFoundException":
                return False
            else:
                raise