def dumps()

in elastic_transport/_serializer.py [0:0]

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

    def dumps(self, data: Any) -> bytes:
        # The body is already encoded to bytes
        # so we forward the request body along.
        if isinstance(data, (bytes, str)):
            data = (data,)

        buffer = bytearray()
        for line in data:
            if isinstance(line, str):
                line = line.encode("utf-8", "surrogatepass")
            if isinstance(line, bytes):
                buffer += line
                # Ensure that there is always a final newline
                if not line.endswith(b"\n"):
                    buffer += b"\n"
            else:
                try:
                    buffer += self.json_dumps(line)
                    buffer += b"\n"
                # This should be captured by the .default()
                # call but just in case we also wrap these.
                except (ValueError, UnicodeError, TypeError) as e:  # pragma: nocover
                    raise SerializationError(
                        message=f"Unable to serialize to NDJSON: {data!r} (type: {type(data).__name__})",
                        errors=(e,),
                    )

        return bytes(buffer)