if c in Connection.executors:
return Connection.executors[c](self, object, multiparams, params)
else:
- raise exceptions.InvalidRequestError("Unexecuteable object type: " + str(type(object)))
+ raise exceptions.InvalidRequestError("Unexecutable object type: " + str(type(object)))
def _execute_default(self, default, multiparams=None, params=None):
return self.__engine.dialect.defaultrunner(self.__create_execution_context()).traverse_single(default)
def _execute_compiled(self, compiled, multiparams=None, params=None, distilled_params=None):
"""Execute a sql.Compiled object."""
if not compiled.can_execute:
- raise exceptions.ArgumentError("Not an executeable clause: %s" % (str(compiled)))
+ raise exceptions.ArgumentError("Not an executable clause: %s" % (str(compiled)))
if distilled_params is None:
distilled_params = self.__distill_params(multiparams, params)
assert users.select().execute().fetchall() == [(7, 'jack'), (8, 'ed'), (9, None)]
def testupdate(self):
-
users.insert().execute(user_id = 7, user_name = 'jack')
assert users.count().scalar() == 1
assert users.select(users.c.user_id==7).execute().fetchone()['user_name'] == 'fred'
def test_lastrow_accessor(self):
- """test the last_inserted_ids() and lastrow_has_id() functions"""
+ """Tests the last_inserted_ids() and lastrow_has_id() functions."""
def insert_values(table, values):
- """insert a row into a table, return the full list of values INSERTed including defaults
- that fired off on the DB side.
-
+ """
+ Inserts a row into a table, returns the full list of values
+ INSERTed including defaults that fired off on the DB side and
detects rows that had defaults and post-fetches.
"""
assert c.execute(s, id=7).fetchall()[0]['user_id'] == 7
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."""
+ """Tests that a BindParam can be used more than once.
+
+ This should be run for DB-APIs with both positional and named
+ paramstyles.
+ """
users.insert().execute(user_id = 7, user_name = 'jack')
users.insert().execute(user_id = 8, user_name = 'fred')
try:
users.join(addresses).execute()
except exceptions.ArgumentError, e:
- assert str(e) == """Not an executeable clause: query_users JOIN query_addresses ON query_users.user_id = query_addresses.user_id"""
+ assert str(e).startswith('Not an executable clause: ')
def test_functions(self):
x = testbase.db.func.current_date().execute().scalar()