in watchman/python/pywatchman/pybser.py [0:0]
def loads_recursive(self, buf, pos):
val_type = _buf_pos(buf, pos)
if (
val_type == BSER_INT8
or val_type == BSER_INT16
or val_type == BSER_INT32
or val_type == BSER_INT64
):
return self.unser_int(buf, pos)
elif val_type == BSER_REAL:
val = struct.unpack_from(b"=d", buf, pos + 1)[0]
return (val, pos + 9)
elif val_type == BSER_TRUE:
return (True, pos + 1)
elif val_type == BSER_FALSE:
return (False, pos + 1)
elif val_type == BSER_NULL:
return (None, pos + 1)
elif val_type == BSER_BYTESTRING:
return self.unser_bytestring(buf, pos)
elif val_type == BSER_UTF8STRING:
return self.unser_utf8_string(buf, pos)
elif val_type == BSER_ARRAY:
return self.unser_array(buf, pos)
elif val_type == BSER_OBJECT:
return self.unser_object(buf, pos)
elif val_type == BSER_TEMPLATE:
return self.unser_template(buf, pos)
else:
raise ValueError(
"unhandled bser opcode 0x%s"
% binascii.hexlify(val_type).decode("ascii")
)