3
3
4
4
from .exceptions import VersionMismatchException
5
5
from .execution_plan import ExecutionPlan
6
- from .query_result import QueryResult
6
+ from .query_result import AsyncQueryResult , QueryResult
7
+
8
+ PROFILE_CMD = "GRAPH.PROFILE"
9
+ RO_QUERY_CMD = "GRAPH.RO_QUERY"
10
+ QUERY_CMD = "GRAPH.QUERY"
11
+ DELETE_CMD = "GRAPH.DELETE"
12
+ SLOWLOG_CMD = "GRAPH.SLOWLOG"
13
+ CONFIG_CMD = "GRAPH.CONFIG"
14
+ LIST_CMD = "GRAPH.LIST"
15
+ EXPLAIN_CMD = "GRAPH.EXPLAIN"
7
16
8
17
9
18
class GraphCommands :
@@ -52,33 +61,28 @@ def query(self, q, params=None, timeout=None, read_only=False, profile=False):
52
61
query = q
53
62
54
63
# handle query parameters
55
- if params is not None :
56
- query = self ._build_params_header (params ) + query
64
+ query = self ._build_params_header (params ) + query
57
65
58
66
# construct query command
59
67
# ask for compact result-set format
60
68
# specify known graph version
61
69
if profile :
62
- cmd = "GRAPH.PROFILE"
70
+ cmd = PROFILE_CMD
63
71
else :
64
- cmd = "GRAPH.RO_QUERY" if read_only else "GRAPH.QUERY"
72
+ cmd = RO_QUERY_CMD if read_only else QUERY_CMD
65
73
command = [cmd , self .name , query , "--compact" ]
66
74
67
75
# include timeout is specified
68
- if timeout :
69
- if not isinstance ( timeout , int ):
70
- raise Exception ( "Timeout argument must be a positive integer" )
71
- command += [ "timeout" , timeout ]
76
+ if isinstance ( timeout , int ) :
77
+ command . extend ([ " timeout" , timeout ])
78
+ elif timeout is not None :
79
+ raise Exception ( "Timeout argument must be a positive integer" )
72
80
73
81
# issue query
74
82
try :
75
83
response = self .execute_command (* command )
76
84
return QueryResult (self , response , profile )
77
85
except ResponseError as e :
78
- if "wrong number of arguments" in str (e ):
79
- print (
80
- "Note: RedisGraph Python requires server version 2.2.8 or above"
81
- ) # noqa
82
86
if "unknown command" in str (e ) and read_only :
83
87
# `GRAPH.RO_QUERY` is unavailable in older versions.
84
88
return self .query (q , params , timeout , read_only = False )
@@ -106,7 +110,7 @@ def delete(self):
106
110
For more information see `DELETE <https://redis.io/commands/graph.delete>`_. # noqa
107
111
"""
108
112
self ._clear_schema ()
109
- return self .execute_command ("GRAPH.DELETE" , self .name )
113
+ return self .execute_command (DELETE_CMD , self .name )
110
114
111
115
# declared here, to override the built in redis.db.flush()
112
116
def flush (self ):
@@ -146,7 +150,7 @@ def slowlog(self):
146
150
3. The issued query.
147
151
4. The amount of time needed for its execution, in milliseconds.
148
152
"""
149
- return self .execute_command ("GRAPH.SLOWLOG" , self .name )
153
+ return self .execute_command (SLOWLOG_CMD , self .name )
150
154
151
155
def config (self , name , value = None , set = False ):
152
156
"""
@@ -170,14 +174,14 @@ def config(self, name, value=None, set=False):
170
174
raise DataError (
171
175
"``value`` can be provided only when ``set`` is True"
172
176
) # noqa
173
- return self .execute_command ("GRAPH.CONFIG" , * params )
177
+ return self .execute_command (CONFIG_CMD , * params )
174
178
175
179
def list_keys (self ):
176
180
"""
177
181
Lists all graph keys in the keyspace.
178
182
For more information see `GRAPH.LIST <https://redis.io/commands/graph.list>`_. # noqa
179
183
"""
180
- return self .execute_command ("GRAPH.LIST" )
184
+ return self .execute_command (LIST_CMD )
181
185
182
186
def execution_plan (self , query , params = None ):
183
187
"""
@@ -188,10 +192,9 @@ def execution_plan(self, query, params=None):
188
192
query: the query that will be executed
189
193
params: query parameters
190
194
"""
191
- if params is not None :
192
- query = self ._build_params_header (params ) + query
195
+ query = self ._build_params_header (params ) + query
193
196
194
- plan = self .execute_command ("GRAPH.EXPLAIN" , self .name , query )
197
+ plan = self .execute_command (EXPLAIN_CMD , self .name , query )
195
198
if isinstance (plan [0 ], bytes ):
196
199
plan = [b .decode () for b in plan ]
197
200
return "\n " .join (plan )
@@ -206,8 +209,105 @@ def explain(self, query, params=None):
206
209
query: the query that will be executed
207
210
params: query parameters
208
211
"""
209
- if params is not None :
210
- query = self ._build_params_header (params ) + query
212
+ query = self ._build_params_header (params ) + query
213
+
214
+ plan = self .execute_command (EXPLAIN_CMD , self .name , query )
215
+ return ExecutionPlan (plan )
216
+
217
+
218
+ class AsyncGraphCommands (GraphCommands ):
219
+ async def query (self , q , params = None , timeout = None , read_only = False , profile = False ):
220
+ """
221
+ Executes a query against the graph.
222
+ For more information see `GRAPH.QUERY <https://oss.redis.com/redisgraph/master/commands/#graphquery>`_. # noqa
223
+
224
+ Args:
225
+
226
+ q : str
227
+ The query.
228
+ params : dict
229
+ Query parameters.
230
+ timeout : int
231
+ Maximum runtime for read queries in milliseconds.
232
+ read_only : bool
233
+ Executes a readonly query if set to True.
234
+ profile : bool
235
+ Return details on results produced by and time
236
+ spent in each operation.
237
+ """
238
+
239
+ # maintain original 'q'
240
+ query = q
241
+
242
+ # handle query parameters
243
+ query = self ._build_params_header (params ) + query
244
+
245
+ # construct query command
246
+ # ask for compact result-set format
247
+ # specify known graph version
248
+ if profile :
249
+ cmd = PROFILE_CMD
250
+ else :
251
+ cmd = RO_QUERY_CMD if read_only else QUERY_CMD
252
+ command = [cmd , self .name , query , "--compact" ]
253
+
254
+ # include timeout is specified
255
+ if isinstance (timeout , int ):
256
+ command .extend (["timeout" , timeout ])
257
+ elif timeout is not None :
258
+ raise Exception ("Timeout argument must be a positive integer" )
259
+
260
+ # issue query
261
+ try :
262
+ response = await self .execute_command (* command )
263
+ return await AsyncQueryResult ().initialize (self , response , profile )
264
+ except ResponseError as e :
265
+ if "unknown command" in str (e ) and read_only :
266
+ # `GRAPH.RO_QUERY` is unavailable in older versions.
267
+ return await self .query (q , params , timeout , read_only = False )
268
+ raise e
269
+ except VersionMismatchException as e :
270
+ # client view over the graph schema is out of sync
271
+ # set client version and refresh local schema
272
+ self .version = e .version
273
+ self ._refresh_schema ()
274
+ # re-issue query
275
+ return await self .query (q , params , timeout , read_only )
276
+
277
+ async def execution_plan (self , query , params = None ):
278
+ """
279
+ Get the execution plan for given query,
280
+ GRAPH.EXPLAIN returns an array of operations.
281
+
282
+ Args:
283
+ query: the query that will be executed
284
+ params: query parameters
285
+ """
286
+ query = self ._build_params_header (params ) + query
211
287
212
- plan = self .execute_command ("GRAPH.EXPLAIN" , self .name , query )
288
+ plan = await self .execute_command (EXPLAIN_CMD , self .name , query )
289
+ if isinstance (plan [0 ], bytes ):
290
+ plan = [b .decode () for b in plan ]
291
+ return "\n " .join (plan )
292
+
293
+ async def explain (self , query , params = None ):
294
+ """
295
+ Get the execution plan for given query,
296
+ GRAPH.EXPLAIN returns ExecutionPlan object.
297
+
298
+ Args:
299
+ query: the query that will be executed
300
+ params: query parameters
301
+ """
302
+ query = self ._build_params_header (params ) + query
303
+
304
+ plan = await self .execute_command (EXPLAIN_CMD , self .name , query )
213
305
return ExecutionPlan (plan )
306
+
307
+ async def flush (self ):
308
+ """
309
+ Commit the graph and reset the edges and the nodes to zero length.
310
+ """
311
+ await self .commit ()
312
+ self .nodes = {}
313
+ self .edges = []
0 commit comments