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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
|
# engine/util.py
# Copyright (C) 2005-2020 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php
from .. import exc
from .. import util
from ..util import collections_abc
from ..util import immutabledict
def connection_memoize(key):
"""Decorator, memoize a function in a connection.info stash.
Only applicable to functions which take no arguments other than a
connection. The memo will be stored in ``connection.info[key]``.
"""
@util.decorator
def decorated(fn, self, connection):
connection = connection.connect()
try:
return connection.info[key]
except KeyError:
connection.info[key] = val = fn(self, connection)
return val
return decorated
_no_tuple = ()
_no_kw = util.immutabledict()
def py_fallback():
# TODO: pass the Connection in so that there can be a standard
# method for warning on parameter format
def _distill_params(connection, multiparams, params): # noqa
r"""Given arguments from the calling form \*multiparams, \**params,
return a list of bind parameter structures, usually a list of
dictionaries.
In the case of 'raw' execution which accepts positional parameters,
it may be a list of tuples or lists.
"""
# C version will fail if this assertion is not true.
# assert isinstance(multiparams, tuple)
if not multiparams:
if params:
connection._warn_for_legacy_exec_format()
return [params]
else:
return []
elif len(multiparams) == 1:
zero = multiparams[0]
if isinstance(zero, (list, tuple)):
if (
not zero
or hasattr(zero[0], "__iter__")
and not hasattr(zero[0], "strip")
):
# execute(stmt, [{}, {}, {}, ...])
# execute(stmt, [(), (), (), ...])
return zero
else:
# this is used by exec_driver_sql only, so a deprecation
# warning would already be coming from passing a plain
# textual statement with positional parameters to
# execute().
# execute(stmt, ("value", "value"))
return [zero]
elif hasattr(zero, "keys"):
# execute(stmt, {"key":"value"})
return [zero]
else:
connection._warn_for_legacy_exec_format()
# execute(stmt, "value")
return [[zero]]
else:
connection._warn_for_legacy_exec_format()
if hasattr(multiparams[0], "__iter__") and not hasattr(
multiparams[0], "strip"
):
return multiparams
else:
return [multiparams]
return locals()
def _distill_cursor_params(connection, multiparams, params):
"""_distill_params without any warnings. more appropriate for
"cursor" params that can include tuple arguments, lists of tuples,
etc.
"""
if not multiparams:
if params:
return [params]
else:
return []
elif len(multiparams) == 1:
zero = multiparams[0]
if isinstance(zero, (list, tuple)):
if (
not zero
or hasattr(zero[0], "__iter__")
and not hasattr(zero[0], "strip")
):
# execute(stmt, [{}, {}, {}, ...])
# execute(stmt, [(), (), (), ...])
return zero
else:
# this is used by exec_driver_sql only, so a deprecation
# warning would already be coming from passing a plain
# textual statement with positional parameters to
# execute().
# execute(stmt, ("value", "value"))
return [zero]
elif hasattr(zero, "keys"):
# execute(stmt, {"key":"value"})
return [zero]
else:
# execute(stmt, "value")
return [[zero]]
else:
if hasattr(multiparams[0], "__iter__") and not hasattr(
multiparams[0], "strip"
):
return multiparams
else:
return [multiparams]
def _distill_params_20(params):
if params is None:
return _no_tuple, _no_kw
elif isinstance(params, list):
# collections_abc.MutableSequence): # avoid abc.__instancecheck__
if params and not isinstance(
params[0], (collections_abc.Mapping, tuple)
):
raise exc.ArgumentError(
"List argument must consist only of tuples or dictionaries"
)
return (params,), _no_kw
elif isinstance(
params,
(tuple, dict, immutabledict),
# avoid abc.__instancecheck__
# (collections_abc.Sequence, collections_abc.Mapping),
):
return (params,), _no_kw
else:
raise exc.ArgumentError("mapping or sequence expected for parameters")
try:
from sqlalchemy.cutils import _distill_params # noqa
except ImportError:
globals().update(py_fallback())
|
