def get_latest_log_events()

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

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

    def get_latest_log_events(self) -> Generator[Tuple[str, dict], None, None]:
        """
        This method gets all the latest log events for this stream that exist at this moment in time.

        cw_client.get_log_events() always returns a nextForwardToken even if the current batch of events is empty.
        You can keep calling cw_client.get_log_events() with the same token until a new batch of log events exist.

        API Reference: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/logs/client/get_log_events.html

        Returns:
            Generator[tuple[str, dict], None, None]: Generator that yields a tuple that consists for two values
                str: stream_name,
                dict: event dict in format
                    {
                        "ingestionTime": number,
                        "message": "string",
                        "timestamp": number
                    }
        """
        while True:
            if not self.next_token:
                token_args = {}
            else:
                token_args = {"nextToken": self.next_token}

            response = self.cw_client.get_log_events(
                logGroupName=self.log_group_name,
                logStreamName=self.log_stream_name,
                startFromHead=True,
                **token_args,
            )

            self.next_token = response["nextForwardToken"]
            if not response["events"]:
                break

            for event in response["events"]:
                yield self.log_stream_name, event