| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573 |
- import sys
- from collections import OrderedDict
- from distutils.util import strtobool
- # from prettytable import PrettyTable
- from redis import ResponseError
- from .edge import Edge
- from .exceptions import VersionMismatchException
- from .node import Node
- from .path import Path
- LABELS_ADDED = "Labels added"
- LABELS_REMOVED = "Labels removed"
- NODES_CREATED = "Nodes created"
- NODES_DELETED = "Nodes deleted"
- RELATIONSHIPS_DELETED = "Relationships deleted"
- PROPERTIES_SET = "Properties set"
- PROPERTIES_REMOVED = "Properties removed"
- RELATIONSHIPS_CREATED = "Relationships created"
- INDICES_CREATED = "Indices created"
- INDICES_DELETED = "Indices deleted"
- CACHED_EXECUTION = "Cached execution"
- INTERNAL_EXECUTION_TIME = "internal execution time"
- STATS = [
- LABELS_ADDED,
- LABELS_REMOVED,
- NODES_CREATED,
- PROPERTIES_SET,
- PROPERTIES_REMOVED,
- RELATIONSHIPS_CREATED,
- NODES_DELETED,
- RELATIONSHIPS_DELETED,
- INDICES_CREATED,
- INDICES_DELETED,
- CACHED_EXECUTION,
- INTERNAL_EXECUTION_TIME,
- ]
- class ResultSetColumnTypes:
- COLUMN_UNKNOWN = 0
- COLUMN_SCALAR = 1
- COLUMN_NODE = 2 # Unused as of RedisGraph v2.1.0, retained for backwards compatibility. # noqa
- COLUMN_RELATION = 3 # Unused as of RedisGraph v2.1.0, retained for backwards compatibility. # noqa
- class ResultSetScalarTypes:
- VALUE_UNKNOWN = 0
- VALUE_NULL = 1
- VALUE_STRING = 2
- VALUE_INTEGER = 3
- VALUE_BOOLEAN = 4
- VALUE_DOUBLE = 5
- VALUE_ARRAY = 6
- VALUE_EDGE = 7
- VALUE_NODE = 8
- VALUE_PATH = 9
- VALUE_MAP = 10
- VALUE_POINT = 11
- class QueryResult:
- def __init__(self, graph, response, profile=False):
- """
- A class that represents a result of the query operation.
- Args:
- graph:
- The graph on which the query was executed.
- response:
- The response from the server.
- profile:
- A boolean indicating if the query command was "GRAPH.PROFILE"
- """
- self.graph = graph
- self.header = []
- self.result_set = []
- # in case of an error an exception will be raised
- self._check_for_errors(response)
- if len(response) == 1:
- self.parse_statistics(response[0])
- elif profile:
- self.parse_profile(response)
- else:
- # start by parsing statistics, matches the one we have
- self.parse_statistics(response[-1]) # Last element.
- self.parse_results(response)
- def _check_for_errors(self, response):
- """
- Check if the response contains an error.
- """
- if isinstance(response[0], ResponseError):
- error = response[0]
- if str(error) == "version mismatch":
- version = response[1]
- error = VersionMismatchException(version)
- raise error
- # If we encountered a run-time error, the last response
- # element will be an exception
- if isinstance(response[-1], ResponseError):
- raise response[-1]
- def parse_results(self, raw_result_set):
- """
- Parse the query execution result returned from the server.
- """
- self.header = self.parse_header(raw_result_set)
- # Empty header.
- if len(self.header) == 0:
- return
- self.result_set = self.parse_records(raw_result_set)
- def parse_statistics(self, raw_statistics):
- """
- Parse the statistics returned in the response.
- """
- self.statistics = {}
- # decode statistics
- for idx, stat in enumerate(raw_statistics):
- if isinstance(stat, bytes):
- raw_statistics[idx] = stat.decode()
- for s in STATS:
- v = self._get_value(s, raw_statistics)
- if v is not None:
- self.statistics[s] = v
- def parse_header(self, raw_result_set):
- """
- Parse the header of the result.
- """
- # An array of column name/column type pairs.
- header = raw_result_set[0]
- return header
- def parse_records(self, raw_result_set):
- """
- Parses the result set and returns a list of records.
- """
- records = [
- [
- self.parse_record_types[self.header[idx][0]](cell)
- for idx, cell in enumerate(row)
- ]
- for row in raw_result_set[1]
- ]
- return records
- def parse_entity_properties(self, props):
- """
- Parse node / edge properties.
- """
- # [[name, value type, value] X N]
- properties = {}
- for prop in props:
- prop_name = self.graph.get_property(prop[0])
- prop_value = self.parse_scalar(prop[1:])
- properties[prop_name] = prop_value
- return properties
- def parse_string(self, cell):
- """
- Parse the cell as a string.
- """
- if isinstance(cell, bytes):
- return cell.decode()
- elif not isinstance(cell, str):
- return str(cell)
- else:
- return cell
- def parse_node(self, cell):
- """
- Parse the cell to a node.
- """
- # Node ID (integer),
- # [label string offset (integer)],
- # [[name, value type, value] X N]
- node_id = int(cell[0])
- labels = None
- if len(cell[1]) > 0:
- labels = []
- for inner_label in cell[1]:
- labels.append(self.graph.get_label(inner_label))
- properties = self.parse_entity_properties(cell[2])
- return Node(node_id=node_id, label=labels, properties=properties)
- def parse_edge(self, cell):
- """
- Parse the cell to an edge.
- """
- # Edge ID (integer),
- # reltype string offset (integer),
- # src node ID offset (integer),
- # dest node ID offset (integer),
- # [[name, value, value type] X N]
- edge_id = int(cell[0])
- relation = self.graph.get_relation(cell[1])
- src_node_id = int(cell[2])
- dest_node_id = int(cell[3])
- properties = self.parse_entity_properties(cell[4])
- return Edge(
- src_node_id, relation, dest_node_id, edge_id=edge_id, properties=properties
- )
- def parse_path(self, cell):
- """
- Parse the cell to a path.
- """
- nodes = self.parse_scalar(cell[0])
- edges = self.parse_scalar(cell[1])
- return Path(nodes, edges)
- def parse_map(self, cell):
- """
- Parse the cell as a map.
- """
- m = OrderedDict()
- n_entries = len(cell)
- # A map is an array of key value pairs.
- # 1. key (string)
- # 2. array: (value type, value)
- for i in range(0, n_entries, 2):
- key = self.parse_string(cell[i])
- m[key] = self.parse_scalar(cell[i + 1])
- return m
- def parse_point(self, cell):
- """
- Parse the cell to point.
- """
- p = {}
- # A point is received an array of the form: [latitude, longitude]
- # It is returned as a map of the form: {"latitude": latitude, "longitude": longitude} # noqa
- p["latitude"] = float(cell[0])
- p["longitude"] = float(cell[1])
- return p
- def parse_null(self, cell):
- """
- Parse a null value.
- """
- return None
- def parse_integer(self, cell):
- """
- Parse the integer value from the cell.
- """
- return int(cell)
- def parse_boolean(self, value):
- """
- Parse the cell value as a boolean.
- """
- value = value.decode() if isinstance(value, bytes) else value
- try:
- scalar = True if strtobool(value) else False
- except ValueError:
- sys.stderr.write("unknown boolean type\n")
- scalar = None
- return scalar
- def parse_double(self, cell):
- """
- Parse the cell as a double.
- """
- return float(cell)
- def parse_array(self, value):
- """
- Parse an array of values.
- """
- scalar = [self.parse_scalar(value[i]) for i in range(len(value))]
- return scalar
- def parse_unknown(self, cell):
- """
- Parse a cell of unknown type.
- """
- sys.stderr.write("Unknown type\n")
- return None
- def parse_scalar(self, cell):
- """
- Parse a scalar value from a cell in the result set.
- """
- scalar_type = int(cell[0])
- value = cell[1]
- scalar = self.parse_scalar_types[scalar_type](value)
- return scalar
- def parse_profile(self, response):
- self.result_set = [x[0 : x.index(",")].strip() for x in response]
- def is_empty(self):
- return len(self.result_set) == 0
- @staticmethod
- def _get_value(prop, statistics):
- for stat in statistics:
- if prop in stat:
- return float(stat.split(": ")[1].split(" ")[0])
- return None
- def _get_stat(self, stat):
- return self.statistics[stat] if stat in self.statistics else 0
- @property
- def labels_added(self):
- """Returns the number of labels added in the query"""
- return self._get_stat(LABELS_ADDED)
- @property
- def labels_removed(self):
- """Returns the number of labels removed in the query"""
- return self._get_stat(LABELS_REMOVED)
- @property
- def nodes_created(self):
- """Returns the number of nodes created in the query"""
- return self._get_stat(NODES_CREATED)
- @property
- def nodes_deleted(self):
- """Returns the number of nodes deleted in the query"""
- return self._get_stat(NODES_DELETED)
- @property
- def properties_set(self):
- """Returns the number of properties set in the query"""
- return self._get_stat(PROPERTIES_SET)
- @property
- def properties_removed(self):
- """Returns the number of properties removed in the query"""
- return self._get_stat(PROPERTIES_REMOVED)
- @property
- def relationships_created(self):
- """Returns the number of relationships created in the query"""
- return self._get_stat(RELATIONSHIPS_CREATED)
- @property
- def relationships_deleted(self):
- """Returns the number of relationships deleted in the query"""
- return self._get_stat(RELATIONSHIPS_DELETED)
- @property
- def indices_created(self):
- """Returns the number of indices created in the query"""
- return self._get_stat(INDICES_CREATED)
- @property
- def indices_deleted(self):
- """Returns the number of indices deleted in the query"""
- return self._get_stat(INDICES_DELETED)
- @property
- def cached_execution(self):
- """Returns whether or not the query execution plan was cached"""
- return self._get_stat(CACHED_EXECUTION) == 1
- @property
- def run_time_ms(self):
- """Returns the server execution time of the query"""
- return self._get_stat(INTERNAL_EXECUTION_TIME)
- @property
- def parse_scalar_types(self):
- return {
- ResultSetScalarTypes.VALUE_NULL: self.parse_null,
- ResultSetScalarTypes.VALUE_STRING: self.parse_string,
- ResultSetScalarTypes.VALUE_INTEGER: self.parse_integer,
- ResultSetScalarTypes.VALUE_BOOLEAN: self.parse_boolean,
- ResultSetScalarTypes.VALUE_DOUBLE: self.parse_double,
- ResultSetScalarTypes.VALUE_ARRAY: self.parse_array,
- ResultSetScalarTypes.VALUE_NODE: self.parse_node,
- ResultSetScalarTypes.VALUE_EDGE: self.parse_edge,
- ResultSetScalarTypes.VALUE_PATH: self.parse_path,
- ResultSetScalarTypes.VALUE_MAP: self.parse_map,
- ResultSetScalarTypes.VALUE_POINT: self.parse_point,
- ResultSetScalarTypes.VALUE_UNKNOWN: self.parse_unknown,
- }
- @property
- def parse_record_types(self):
- return {
- ResultSetColumnTypes.COLUMN_SCALAR: self.parse_scalar,
- ResultSetColumnTypes.COLUMN_NODE: self.parse_node,
- ResultSetColumnTypes.COLUMN_RELATION: self.parse_edge,
- ResultSetColumnTypes.COLUMN_UNKNOWN: self.parse_unknown,
- }
- class AsyncQueryResult(QueryResult):
- """
- Async version for the QueryResult class - a class that
- represents a result of the query operation.
- """
- def __init__(self):
- """
- To init the class you must call self.initialize()
- """
- pass
- async def initialize(self, graph, response, profile=False):
- """
- Initializes the class.
- Args:
- graph:
- The graph on which the query was executed.
- response:
- The response from the server.
- profile:
- A boolean indicating if the query command was "GRAPH.PROFILE"
- """
- self.graph = graph
- self.header = []
- self.result_set = []
- # in case of an error an exception will be raised
- self._check_for_errors(response)
- if len(response) == 1:
- self.parse_statistics(response[0])
- elif profile:
- self.parse_profile(response)
- else:
- # start by parsing statistics, matches the one we have
- self.parse_statistics(response[-1]) # Last element.
- await self.parse_results(response)
- return self
- async def parse_node(self, cell):
- """
- Parses a node from the cell.
- """
- # Node ID (integer),
- # [label string offset (integer)],
- # [[name, value type, value] X N]
- labels = None
- if len(cell[1]) > 0:
- labels = []
- for inner_label in cell[1]:
- labels.append(await self.graph.get_label(inner_label))
- properties = await self.parse_entity_properties(cell[2])
- node_id = int(cell[0])
- return Node(node_id=node_id, label=labels, properties=properties)
- async def parse_scalar(self, cell):
- """
- Parses a scalar value from the server response.
- """
- scalar_type = int(cell[0])
- value = cell[1]
- try:
- scalar = await self.parse_scalar_types[scalar_type](value)
- except TypeError:
- # Not all of the functions are async
- scalar = self.parse_scalar_types[scalar_type](value)
- return scalar
- async def parse_records(self, raw_result_set):
- """
- Parses the result set and returns a list of records.
- """
- records = []
- for row in raw_result_set[1]:
- record = [
- await self.parse_record_types[self.header[idx][0]](cell)
- for idx, cell in enumerate(row)
- ]
- records.append(record)
- return records
- async def parse_results(self, raw_result_set):
- """
- Parse the query execution result returned from the server.
- """
- self.header = self.parse_header(raw_result_set)
- # Empty header.
- if len(self.header) == 0:
- return
- self.result_set = await self.parse_records(raw_result_set)
- async def parse_entity_properties(self, props):
- """
- Parse node / edge properties.
- """
- # [[name, value type, value] X N]
- properties = {}
- for prop in props:
- prop_name = await self.graph.get_property(prop[0])
- prop_value = await self.parse_scalar(prop[1:])
- properties[prop_name] = prop_value
- return properties
- async def parse_edge(self, cell):
- """
- Parse the cell to an edge.
- """
- # Edge ID (integer),
- # reltype string offset (integer),
- # src node ID offset (integer),
- # dest node ID offset (integer),
- # [[name, value, value type] X N]
- edge_id = int(cell[0])
- relation = await self.graph.get_relation(cell[1])
- src_node_id = int(cell[2])
- dest_node_id = int(cell[3])
- properties = await self.parse_entity_properties(cell[4])
- return Edge(
- src_node_id, relation, dest_node_id, edge_id=edge_id, properties=properties
- )
- async def parse_path(self, cell):
- """
- Parse the cell to a path.
- """
- nodes = await self.parse_scalar(cell[0])
- edges = await self.parse_scalar(cell[1])
- return Path(nodes, edges)
- async def parse_map(self, cell):
- """
- Parse the cell to a map.
- """
- m = OrderedDict()
- n_entries = len(cell)
- # A map is an array of key value pairs.
- # 1. key (string)
- # 2. array: (value type, value)
- for i in range(0, n_entries, 2):
- key = self.parse_string(cell[i])
- m[key] = await self.parse_scalar(cell[i + 1])
- return m
- async def parse_array(self, value):
- """
- Parse array value.
- """
- scalar = [await self.parse_scalar(value[i]) for i in range(len(value))]
- return scalar
|