1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
|
import copy
import random
import string
def list_or_args(keys, args):
# returns a single new list combining keys and args
try:
iter(keys)
# a string or bytes instance can be iterated, but indicates
# keys wasn't passed as a list
if isinstance(keys, (bytes, str)):
keys = [keys]
else:
keys = list(keys)
except TypeError:
keys = [keys]
if args:
keys.extend(args)
return keys
def nativestr(x):
"""Return the decoded binary string, or a string, depending on type."""
r = x.decode("utf-8", "replace") if isinstance(x, bytes) else x
if r == "null":
return
return r
def delist(x):
"""Given a list of binaries, return the stringified version."""
if x is None:
return x
return [nativestr(obj) for obj in x]
def parse_to_list(response):
"""Optimistically parse the response to a list."""
res = []
if response is None:
return res
for item in response:
try:
res.append(int(item))
except ValueError:
try:
res.append(float(item))
except ValueError:
res.append(nativestr(item))
except TypeError:
res.append(None)
return res
def parse_list_to_dict(response):
res = {}
for i in range(0, len(response), 2):
if isinstance(response[i], list):
res["Child iterators"].append(parse_list_to_dict(response[i]))
elif isinstance(response[i + 1], list):
res["Child iterators"] = [parse_list_to_dict(response[i + 1])]
else:
try:
res[response[i]] = float(response[i + 1])
except (TypeError, ValueError):
res[response[i]] = response[i + 1]
return res
def parse_to_dict(response):
if response is None:
return {}
res = {}
for det in response:
if isinstance(det[1], list):
res[det[0]] = parse_list_to_dict(det[1])
else:
try: # try to set the attribute. may be provided without value
try: # try to convert the value to float
res[det[0]] = float(det[1])
except (TypeError, ValueError):
res[det[0]] = det[1]
except IndexError:
pass
return res
def random_string(length=10):
"""
Returns a random N character long string.
"""
return "".join( # nosec
random.choice(string.ascii_lowercase) for x in range(length)
)
def quote_string(v):
"""
RedisGraph strings must be quoted,
quote_string wraps given v with quotes incase
v is a string.
"""
if isinstance(v, bytes):
v = v.decode()
elif not isinstance(v, str):
return v
if len(v) == 0:
return '""'
v = v.replace('"', '\\"')
return f'"{v}"'
def decodeDictKeys(obj):
"""Decode the keys of the given dictionary with utf-8."""
newobj = copy.copy(obj)
for k in obj.keys():
if isinstance(k, bytes):
newobj[k.decode("utf-8")] = newobj[k]
newobj.pop(k)
return newobj
def stringify_param_value(value):
"""
Turn a parameter value into a string suitable for the params header of
a Cypher command.
You may pass any value that would be accepted by `json.dumps()`.
Ways in which output differs from that of `str()`:
* Strings are quoted.
* None --> "null".
* In dictionaries, keys are _not_ quoted.
:param value: The parameter value to be turned into a string.
:return: string
"""
if isinstance(value, str):
return quote_string(value)
elif value is None:
return "null"
elif isinstance(value, (list, tuple)):
return f'[{",".join(map(stringify_param_value, value))}]'
elif isinstance(value, dict):
return f'{{{",".join(f"{k}:{stringify_param_value(v)}" for k, v in value.items())}}}' # noqa
else:
return str(value)
|
