def _write_log_entries()

in opentelemetry-exporter-gcp-logging/src/opentelemetry/exporter/cloud_logging/__init__.py [0:0]

• 37 lines of code
• 7 McCabe index (conditional complexity)

    def _write_log_entries(self, log_entries: list[LogEntry]):
        batch: list[LogEntry] = []
        batch_byte_size = 0
        for entry in log_entries:
            msg_size = LogEntry.pb(entry).ByteSize()
            if msg_size > DEFAULT_MAX_ENTRY_SIZE:
                logging.warning(
                    "Cannot write log that is %s bytes which exceeds Cloud Logging's maximum limit of %s bytes.",
                    msg_size,
                    DEFAULT_MAX_ENTRY_SIZE,
                )
                continue
            if msg_size + batch_byte_size > DEFAULT_MAX_REQUEST_SIZE:
                try:
                    self.client.write_log_entries(
                        WriteLogEntriesRequest(
                            entries=batch, partial_success=True
                        )
                    )
                # pylint: disable=broad-except
                except Exception as ex:
                    logging.error(
                        "Error while writing to Cloud Logging", exc_info=ex
                    )
                batch = [entry]
                batch_byte_size = msg_size
            else:
                batch.append(entry)
                batch_byte_size += msg_size
        if batch:
            try:
                self.client.write_log_entries(
                    WriteLogEntriesRequest(entries=batch, partial_success=True)
                )
            # pylint: disable=broad-except
            except Exception as ex:
                logging.error(
                    "Error while writing to Cloud Logging", exc_info=ex
                )