in azure/Kqlmagic/kql_response.py [0:0]
def extended_properties(self):
if self._extended_properties is None:
self._extended_properties = {}
if self.endpoint_version == "v2":
for table in self.all_tables:
if table["TableName"] == "@ExtendedProperties" and table["TableKind"] == "QueryProperties":
cols_idx_map = self._map_columns_to_index(table["Columns"])
types = self._get_columns_types(table["Columns"])
key_idx = cols_idx_map.get("Key")
table_id_idx = cols_idx_map.get("TableId")
value_idx = cols_idx_map.get("Value")
if (
key_idx is not None
and table_id_idx is not None
and value_idx is not None
and types[key_idx] == "string"
and types[table_id_idx] == "int"
):
for row in table["Rows"]:
table_id = row[table_id_idx]
key = row[key_idx]
value = row[value_idx]
extended_properties = self._extended_properties.get(table_id, {})
if key == ExtendedPropertiesKeys.VISUALIZATION:
value_obj = self._dynamic_to_object(value)
extended_properties[key] = value_obj
elif key == ExtendedPropertiesKeys.CURSOR:
# Note: although the value column is defined as dynamic the value here is string
extended_properties[key] = str(value)
else:
extended_properties[key] = value
self._extended_properties[table_id] = extended_properties
else:
tables_num = self.json_response["Tables"].__len__()
if tables_num > 1:
last_table = self.json_response["Tables"][tables_num - 1]
for row in last_table["Rows"]:
if row[2] == "@ExtendedProperties" and row[1] == "QueryProperties":
table = self.json_response["Tables"][row[0]]
table_id = 0
value_idx = 0
extended_properties = {}
for row in table["Rows"]:
value = row[value_idx]
value_obj = self._dynamic_to_object(value)
if type(value_obj) == dict:
key = list(value_obj.keys())[0]
if key == ExtendedPropertiesKeys.VISUALIZATION:
extended_properties[key] = value_obj
elif key == ExtendedPropertiesKeys.CURSOR:
extended_properties[key] = str(value_obj.get(ExtendedPropertiesKeys.CURSOR,""))
else:
extended_properties[key] = value_obj
self._extended_properties[table_id] = extended_properties
break
return self._extended_properties