to save)
- improved the check for objects being part of a session when the
unit of work seeks to flush() them as part of a relationship..
+- [ticket:280] statement execution supports using the same BindParam
+object more than once in an expression; simplified handling of positional
+parameters. nice job by Bill Noon figuring out the basic idea.
0.2.7
- quoting facilities set up so that database-specific quoting can be
self.bindtemplate = ":%s"
self.paramstyle = dialect.paramstyle
self.positional = dialect.positional
+ self.positiontup = []
self.preparer = dialect.preparer()
def after_compile(self):
if self.paramstyle=='pyformat':
self.strings[self.statement] = re.sub(match, lambda m:'%(' + m.group(1) +')s', self.strings[self.statement])
elif self.positional:
- self.positiontup = []
params = re.finditer(match, self.strings[self.statement])
for p in params:
self.positiontup.append(p.group(1))
bindparams = {}
bindparams.update(params)
- d = sql.ClauseParameters(self.dialect)
- if self.positional:
- for k in self.positiontup:
- b = self.binds[k]
- d.set_parameter(k, b.value, b)
- else:
- for b in self.binds.values():
- d.set_parameter(b.key, b.value, b)
-
+ d = sql.ClauseParameters(self.dialect, self.positiontup)
+ for b in self.binds.values():
+ d.set_parameter(b.key, b.value, b)
+
for key, value in bindparams.iteritems():
try:
b = self.binds[key]
return d
- def get_named_params(self, parameters):
- """given the results of the get_params method, returns the parameters
- in dictionary format. For a named paramstyle, this just returns the
- same dictionary. For a positional paramstyle, the given parameters are
- assumed to be in list format and are converted back to a dictionary.
- """
- if self.positional:
- p = {}
- for i in range(0, len(self.positiontup)):
- p[self.positiontup[i]] = parameters[i]
- return p
- else:
- return parameters
-
def default_from(self):
"""called when a SELECT statement has no froms, and no FROM clause is to be appended.
gives Oracle a chance to tack on a "FROM DUAL" to the string output. """
if parameters is not None:
if self.positional:
if executemany:
- parameters = [p.values() for p in parameters]
+ parameters = [p.get_raw_list() for p in parameters]
else:
- parameters = parameters.values()
+ parameters = parameters.get_raw_list()
else:
if executemany:
parameters = [p.get_raw_dict() for p in parameters]
"""represents the behavior of a particular database. Used by Compiled objects."""
pass
-class ClauseParameters(util.OrderedDict):
- """represents a dictionary/iterator of bind parameter key names/values. Includes parameters compiled with a Compiled object as well as additional arguments passed to the Compiled object's get_params() method. Parameter values will be converted as per the TypeEngine objects present in the bind parameter objects. The non-converted value can be retrieved via the get_original method. For Compiled objects that compile positional parameters, the values() iteration of the object will return the parameter values in the correct order."""
- def __init__(self, dialect):
+class ClauseParameters(dict):
+ """represents a dictionary/iterator of bind parameter key names/values.
+
+ Tracks the original BindParam objects as well as the keys/position of each
+ parameter, and can return parameters as a dictionary or a list.
+ Will process parameter values according to the TypeEngine objects present in
+ the BindParams.
+ """
+ def __init__(self, dialect, positional=None):
super(ClauseParameters, self).__init__(self)
self.dialect=dialect
self.binds = {}
+ self.positional = positional or []
def set_parameter(self, key, value, bindparam):
self[key] = value
self.binds[key] = bindparam
if self.binds.has_key(key):
v = self.binds[key].typeprocess(v, self.dialect)
return v
- def values(self):
- return [self[key] for key in self]
def get_original_dict(self):
return self.copy()
+ def get_raw_list(self):
+ return [self[key] for key in self.positional]
def get_raw_dict(self):
d = {}
for k in self:
finally:
test_table.drop()
+ def test_repeated_bindparams(self):
+ """test that a BindParam can be used more than once.
+ this should be run for dbs with both positional and named paramstyles."""
+ self.users.insert().execute(user_id = 7, user_name = 'jack')
+ self.users.insert().execute(user_id = 8, user_name = 'fred')
+
+ u = bindparam('uid')
+ s = self.users.select(or_(self.users.c.user_name==u, self.users.c.user_name==u))
+ r = s.execute(uid='fred').fetchall()
+ assert len(r) == 1
def testdelete(self):
self.users.insert().execute(user_id = 7, user_name = 'jack')
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
