def _put_log_event()

in src/cloudformation_cli_python_lib/log_delivery.py [0:0]

• 20 lines of code
• 3 McCabe index (conditional complexity)

    def _put_log_event(self, msg: logging.LogRecord) -> None:
        kwargs = {
            "logGroupName": self.group,
            "logStreamName": self.stream,
            "logEvents": [
                {"timestamp": round(time.time() * 1000), "message": self.format(msg)}
            ],
        }
        if self.sequence_token:
            kwargs["sequenceToken"] = self.sequence_token
        try:
            self.sequence_token = self.client.put_log_events(**kwargs)[
                "nextSequenceToken"
            ]
        except (
            self.client.exceptions.DataAlreadyAcceptedException,
            self.client.exceptions.InvalidSequenceTokenException,
        ) as e:
            self.sequence_token = str(e).rsplit(" ", maxsplit=1)[-1]
            self._put_log_event(msg)