def _encode_header_val()

in amazon_transcribe/eventstream.py [0:0]

• 30 lines of code
• 14 McCabe index (conditional complexity)

    def _encode_header_val(self, val: HEADER_SERIALIZATION_VALUE) -> bytes:
        # Handle booleans first to avoid being viewed as ints
        if val is True:
            return b"\x00"
        elif val is False:
            return b"\x01"

        if isinstance(val, int):
            val = self.DEFAULT_INT_TYPE(val)

        if isinstance(val, Int8HeaderValue):
            return b"\x02" + pack("!b", val.value)
        elif isinstance(val, Int16HeaderValue):
            return b"\x03" + pack("!h", val.value)
        elif isinstance(val, Int32HeaderValue):
            return b"\x04" + pack("!i", val.value)
        elif isinstance(val, Int64HeaderValue):
            return b"\x05" + pack("!q", val.value)
        elif isinstance(val, bytes):
            # Byte arrays are prefaced with a 16bit length, but are restricted
            # to a max length of 2**15 - 1, enforce this explicitly
            if len(val) > _MAX_HEADER_VALUE_BYTE_LENGTH:
                raise HeaderValueBytesExceedMaxLength(len(val))
            return b"\x06" + pack("!H", len(val)) + val
        elif isinstance(val, str):
            utf8_string = val.encode("utf-8")
            # Strings are prefaced with a 16bit length, but are restricted
            # to a max length of 2**15 - 1, enforce this explicitly
            if len(utf8_string) > _MAX_HEADER_VALUE_BYTE_LENGTH:
                raise HeaderValueBytesExceedMaxLength(len(utf8_string))
            return b"\x07" + pack("!H", len(utf8_string)) + utf8_string
        elif isinstance(val, datetime.datetime):
            ms_timestamp = int(val.timestamp() * 1000)
            return b"\x08" + pack("!q", ms_timestamp)
        elif isinstance(val, uuid.UUID):
            return b"\x09" + val.bytes
        raise InvalidHeaderValue(val)