def set_field()

in datahub/batch/binary_record.py [0:0]

• 25 lines of code
• 9 McCabe index (conditional complexity)

    def set_field(self, pos, value):
        self.__set_none(pos)
        if self._schema is None:    # Blob
            if not isinstance(value, bytes):
                raise InvalidParameterException("Only support write bytes for no schema")
            self._fields[0] = self.__set_byte_field(value)
        else:                       # Tuple
            field = self._schema.get_field(pos)
            field_type = field.type
            if field_type is FieldType.STRING or field_type is FieldType.DECIMAL:
                value_byte = self.__set_byte_field(to_binary(str(value)))
            elif field_type is FieldType.BOOLEAN:
                value_byte = bool2byte(value)
                value_byte += PADDING_BYTES * 7
            elif field_type is FieldType.FLOAT:
                value_byte = float2byte(value)
                value_byte += PADDING_BYTES * 4
            elif field_type is FieldType.DOUBLE:
                value_byte = double2byte(value)
            elif field_type in (FieldType.TINYINT, FieldType.SMALLINT, FieldType.INTEGER,
                                FieldType.BIGINT, FieldType.TIMESTAMP):
                value_byte = int2byte(value, size=8)
            else:
                raise DatahubException("Error field type. {}".format(field_type))
            self._fields[pos] = value_byte