.. changelog::
:version: 0.8.0b2
+ .. change::
+ :tags: sql, feature
+ :tickets: 2623
+
+ The :class:`.Insert` construct now supports multi-valued inserts,
+ that is, an INSERT that renders like
+ "INSERT INTO table VALUES (...), (...), ...".
+ Supported by Postgresql, SQLite, and MySQL.
+ Big thanks to Idan Kamara for doing the legwork on this one.
+
.. change::
:tags: oracle, bug
:tickets: 2620
def visit_insert(self, insert_stmt, **kw):
self.isinsert = True
- cols, params = self._get_colparams(insert_stmt)
+ colparams = self._get_colparams(insert_stmt)
- if not cols and \
+ if not colparams and \
not self.dialect.supports_default_values and \
not self.dialect.supports_empty_insert:
- raise exc.CompileError("The version of %s you are using does "
- "not support empty inserts." %
+ raise exc.CompileError("The '%s' dialect with current database "
+ "version settings does not support empty "
+ "inserts." %
self.dialect.name)
- if insert_stmt.multi_parameters and not self.dialect.supports_multirow_insert:
- raise exc.CompileError("The version of %s you are using does "
- "not support multirow inserts." %
+ if insert_stmt._has_multi_parameters:
+ if not self.dialect.supports_multirow_insert:
+ raise exc.CompileError("The '%s' dialect with current database "
+ "version settings does not support "
+ "in-place multirow inserts." %
self.dialect.name)
+ colparams_single = colparams[0]
+ else:
+ colparams_single = colparams
+
preparer = self.preparer
supports_default_values = self.dialect.supports_default_values
text += table_text
- if cols or not supports_default_values:
- text += " (%s)" % ', '.join([preparer.format_column(c)
- for c in cols])
+ if colparams_single or not supports_default_values:
+ text += " (%s)" % ', '.join([preparer.format_column(c[0])
+ for c in colparams_single])
if self.returning or insert_stmt._returning:
self.returning = self.returning or insert_stmt._returning
if self.returning_precedes_values:
text += " " + returning_clause
- if not cols and supports_default_values:
+ if not colparams and supports_default_values:
text += " DEFAULT VALUES"
+ elif insert_stmt._has_multi_parameters:
+ text += " VALUES %s" % (
+ ", ".join(
+ "(%s)" % (
+ ', '.join(c[1] for c in colparam_set)
+ )
+ for colparam_set in colparams
+ )
+ )
else:
- values = []
- for row in params:
- values.append('(%s)' % ', '.join(row))
- text += " VALUES %s" % \
- ', '.join(values)
+ text += " VALUES (%s)" % \
+ ', '.join([c[1] for c in colparams])
if self.returning and not self.returning_precedes_values:
text += " " + returning_clause
extra_froms = update_stmt._extra_froms
- cols, params = self._get_colparams(update_stmt, extra_froms)
+ colparams = self._get_colparams(update_stmt, extra_froms)
text = "UPDATE "
text += ' SET '
include_table = extra_froms and \
self.render_table_with_column_in_update_from
- colparams = []
- if params:
- colparams = zip(cols, params[0])
text += ', '.join(
- c._compiler_dispatch(self,
+ c[0]._compiler_dispatch(self,
include_table=include_table) +
- '=' + p for c, p in colparams
+ '=' + c[1] for c in colparams
)
if update_stmt._returning:
# no parameters in the statement, no parameters in the
# compiled params - return binds for all columns
if self.column_keys is None and stmt.parameters is None:
- values = [self._create_crud_bind_param(c, None, required=True)
- for c in stmt.table.columns]
- return list(stmt.table.columns), [values]
+ return [
+ (c, self._create_crud_bind_param(c,
+ None, required=True))
+ for c in stmt.table.columns
+ ]
required = object()
+ if stmt._has_multi_parameters:
+ stmt_parameters = stmt.parameters[0]
+ else:
+ stmt_parameters = stmt.parameters
+
# if we have statement parameters - set defaults in the
# compiled params
if self.column_keys is None:
else:
parameters = dict((sql._column_as_key(key), required)
for key in self.column_keys
- if not stmt.parameters or
- key not in stmt.parameters)
+ if not stmt_parameters or
+ key not in stmt_parameters)
# create a list of column assignment clauses as tuples
- columns = []
values = []
- if stmt.parameters is not None:
- for k, v in stmt.parameters.iteritems():
+ if stmt_parameters is not None:
+ for k, v in stmt_parameters.iteritems():
colkey = sql._column_as_key(k)
if colkey is not None:
parameters.setdefault(colkey, v)
else:
v = self.process(v.self_group())
- columns.append(k)
- values.append(v)
+ values.append((k, v))
need_pks = self.isinsert and \
not self.inline and \
check_columns = {}
# special logic that only occurs for multi-table UPDATE
# statements
- if extra_tables and stmt.parameters:
+ if extra_tables and stmt_parameters:
normalized_params = dict(
(sql._clause_element_as_expr(c), param)
- for c, param in stmt.parameters.items()
+ for c, param in stmt_parameters.items()
)
assert self.isupdate
affected_tables = set()
else:
self.postfetch.append(c)
value = self.process(value.self_group())
- columns.append(c)
- values.append(value)
+ values.append((c, value))
# determine tables which are actually
# to be updated - process onupdate and
# server_onupdate for these
continue
elif c.onupdate is not None and not c.onupdate.is_sequence:
if c.onupdate.is_clause_element:
- columns.apppend(c)
- values.append(self.process(c.onupdate.arg.self_group()))
+ values.append(
+ (c, self.process(c.onupdate.arg.self_group()))
+ )
self.postfetch.append(c)
else:
- columns.append(c)
- values.append(self._create_crud_bind_param(c, None))
+ values.append(
+ (c, self._create_crud_bind_param(c, None))
+ )
self.prefetch.append(c)
elif c.server_onupdate is not None:
self.postfetch.append(c)
value = parameters.pop(c.key)
if sql._is_literal(value):
value = self._create_crud_bind_param(
- c, value, required=value is required)
+ c, value, required=value is required,
+ name=c.key
+ if not stmt._has_multi_parameters
+ else "%s_0" % c.key
+ )
elif c.primary_key and implicit_returning:
self.returning.append(c)
value = self.process(value.self_group())
else:
self.postfetch.append(c)
value = self.process(value.self_group())
- columns.append(c)
- values.append(value)
+ values.append((c, value))
elif self.isinsert:
if c.primary_key and \
(not c.default.optional or \
not self.dialect.sequences_optional):
proc = self.process(c.default)
- columns.append(c)
- values.append(proc)
+ values.append((c, proc))
self.returning.append(c)
elif c.default.is_clause_element:
- columns.append(c)
- values.append(self.process(c.default.arg.self_group()))
+ values.append(
+ (c,
+ self.process(c.default.arg.self_group()))
+ )
self.returning.append(c)
else:
- columns.append(c)
- values.append(self._create_crud_bind_param(c, None))
+ values.append(
+ (c, self._create_crud_bind_param(c, None))
+ )
self.prefetch.append(c)
else:
self.returning.append(c)
self.dialect.preexecute_autoincrement_sequences
):
- columns.append(c)
- values.append(self._create_crud_bind_param(c, None))
+ values.append(
+ (c, self._create_crud_bind_param(c, None))
+ )
+
self.prefetch.append(c)
elif c.default is not None:
(not c.default.optional or \
not self.dialect.sequences_optional):
proc = self.process(c.default)
- columns.append(c)
- values.append(proc)
+ values.append((c, proc))
if not c.primary_key:
self.postfetch.append(c)
elif c.default.is_clause_element:
- columns.append(c)
- values.append(self.process(c.default.arg.self_group()))
+ values.append(
+ (c, self.process(c.default.arg.self_group()))
+ )
if not c.primary_key:
# dont add primary key column to postfetch
self.postfetch.append(c)
else:
- columns.append(c)
- values.append(self._create_crud_bind_param(c, None))
+ values.append(
+ (c, self._create_crud_bind_param(c, None))
+ )
self.prefetch.append(c)
elif c.server_default is not None:
if not c.primary_key:
elif self.isupdate:
if c.onupdate is not None and not c.onupdate.is_sequence:
if c.onupdate.is_clause_element:
- columns.append(c)
- values.append(self.process(c.onupdate.arg.self_group()))
+ values.append(
+ (c, self.process(c.onupdate.arg.self_group()))
+ )
self.postfetch.append(c)
else:
- columns.append(c)
- values.append(self._create_crud_bind_param(c, None))
+ values.append(
+ (c, self._create_crud_bind_param(c, None))
+ )
self.prefetch.append(c)
elif c.server_onupdate is not None:
self.postfetch.append(c)
- if parameters and stmt.parameters:
+ if parameters and stmt_parameters:
check = set(parameters).intersection(
sql._column_as_key(k) for k in stmt.parameters
).difference(check_columns)
(", ".join(check))
)
- if values:
+ if stmt._has_multi_parameters:
+ values_0 = values
values = [values]
- for i, row in enumerate(stmt.multi_parameters):
- r = []
- for c in columns:
- r.append(self._create_crud_bind_param(c, row[c.key],
- name=c.key + str(i)))
- values.append(r)
+ values.extend(
+ [
+ (
+ c,
+ self._create_crud_bind_param(
+ c, row[c.key],
+ name="%s_%d" % (c.key, i + 1)
+ )
+ if c.key in row else param
+ )
+ for (c, param) in values_0
+ ]
+ for i, row in enumerate(stmt.parameters[1:])
+ )
- return columns, values
+ return values
def visit_delete(self, delete_stmt, **kw):
self.stack.append({'from': set([delete_stmt.table])})
:class:`~.schema.Table`.
- :param table: The table to be inserted into.
+ :param table: :class:`.TableClause` which is the subject of the insert.
- :param values: A dictionary which specifies the column specifications of
- the ``INSERT``, and is optional. If left as None, the column
- specifications are determined from the bind parameters used during the
- compile phase of the ``INSERT`` statement. If the bind parameters also
- are None during the compile phase, then the column specifications will be
- generated from the full list of table columns. Note that the
- :meth:`~Insert.values()` generative method may also be used for this.
+ :param values: collection of values to be inserted; see
+ :meth:`.Insert.values` for a description of allowed formats here.
+ Can be omitted entirely; a :class:`.Insert` construct will also
+ dynamically render the VALUES clause at execution time based on
+ the parameters passed to :meth:`.Connection.execute`.
:param inline: if True, SQL defaults will be compiled 'inline' into the
statement and not pre-executed.
``INSERT`` statement's table, the statement will be correlated
against the ``INSERT`` statement.
- See also:
+ .. seealso::
:ref:`coretutorial_insert_expressions` - SQL Expression Tutorial
def _process_colparams(self, parameters):
def process_single(p):
if isinstance(p, (list, tuple)):
- pp = {}
- for i, c in enumerate(self.table.c):
- pp[c.key] = p[i]
- return pp
+ return dict(
+ (c.key, pval)
+ for c, pval in zip(self.table.c, p)
+ )
else:
return p
if isinstance(parameters, (list, tuple)) and \
isinstance(parameters[0], (list, tuple, dict)):
- return process_single(parameters[0]), \
- [process_single(p) for p in parameters[1:]]
+
+ if not self._supports_multi_parameters:
+ raise exc.InvalidRequestError(
+ "This construct does not support "
+ "multiple parameter sets.")
+
+ return [process_single(p) for p in parameters], True
else:
- return process_single(parameters), []
+ return process_single(parameters), False
def params(self, *arg, **kw):
"""Set the parameters for the statement.
self._bind = bind
bind = property(bind, _set_bind)
- _returning_re = re.compile(r'(?:firebird|postgres(?:ql)?)_returning')
-
- def _process_deprecated_kw(self, kwargs):
- for k in list(kwargs):
- m = self._returning_re.match(k)
- if m:
- self._returning = kwargs.pop(k)
- util.warn_deprecated(
- "The %r argument is deprecated. Please "
- "use statement.returning(col1, col2, ...)" % k
- )
- return kwargs
-
@_generative
def returning(self, *cols):
"""Add a RETURNING or equivalent clause to this statement.
__visit_name__ = 'values_base'
+ _supports_multi_parameters = False
+ _has_multi_parameters = False
+
def __init__(self, table, values, prefixes):
self.table = table
- self.parameters, self.multi_parameters = self._process_colparams(values)
+ self.parameters, self._has_multi_parameters = \
+ self._process_colparams(values)
if prefixes:
self._setup_prefixes(prefixes)
@_generative
def values(self, *args, **kwargs):
- """specify the VALUES clause for an INSERT statement, or the SET
+ """specify a fixed VALUES clause for an INSERT statement, or the SET
clause for an UPDATE.
+ Note that the :class:`.Insert` and :class:`.Update` constructs support
+ per-execution time formatting of the VALUES and/or SET clauses,
+ based on the arguments passed to :meth:`.Connection.execute`. However,
+ the :meth:`.ValuesBase.values` method can be used to "fix" a particular
+ set of parameters into the statement.
+
+ Multiple calls to :meth:`.ValuesBase.values` will produce a new
+ construct, each one with the parameter list modified to include
+ the new parameters sent. In the typical case of a single
+ dictionary of parameters, the newly passed keys will replace
+ the same keys in the previous construct. In the case of a list-based
+ "multiple values" construct, each new list of values is extended
+ onto the existing list of values.
+
:param \**kwargs: key value pairs representing the string key
of a :class:`.Column` mapped to the value to be rendered into the
VALUES or SET clause::
users.update().where(users.c.id==5).values(name="some name")
- :param \*args: A single dictionary can be sent as the first positional
- argument. This allows non-string based keys, such as Column
- objects, to be used::
+ :param \*args: Alternatively, a dictionary, tuple or list
+ of dictionaries or tuples can be passed as a single positional
+ argument in order to form the VALUES or
+ SET clause of the statement. The single dictionary form
+ works the same as the kwargs form::
- users.insert().values({users.c.name : "some name"})
+ users.insert().values({"name": "some name"})
- users.update().where(users.c.id==5).values(
- {users.c.name: "some name"})
+ If a tuple is passed, the tuple should contain the same number
+ of columns as the target :class:`.Table`::
- See also:
+ users.insert().values((5, "some name"))
+
+ The :class:`.Insert` construct also supports multiply-rendered VALUES
+ construct, for those backends which support this SQL syntax
+ (SQLite, Postgresql, MySQL). This mode is indicated by passing a list
+ of one or more dictionaries/tuples::
+
+ users.insert().values([
+ {"name": "some name"},
+ {"name": "some other name"},
+ {"name": "yet another name"},
+ ])
+
+ In the case of an :class:`.Update`
+ construct, only the single dictionary/tuple form is accepted,
+ else an exception is raised. It is also an exception case to
+ attempt to mix the single-/multiple- value styles together,
+ either through multiple :meth:`.ValuesBase.values` calls
+ or by sending a list + kwargs at the same time.
+
+ .. note::
+
+ Passing a multiple values list is *not* the same
+ as passing a multiple values list to the :meth:`.Connection.execute`
+ method. Passing a list of parameter sets to :meth:`.ValuesBase.values`
+ produces a construct of this form::
+
+ INSERT INTO table (col1, col2, col3) VALUES
+ (col1_0, col2_0, col3_0),
+ (col1_1, col2_1, col3_1),
+ ...
+
+ whereas a multiple list passed to :meth:`.Connection.execute`
+ has the effect of using the DBAPI
+ `executemany() <http://www.python.org/dev/peps/pep-0249/#id18>`_
+ method, which provides a high-performance system of invoking
+ a single-row INSERT statement many times against a series
+ of parameter sets. The "executemany" style is supported by
+ all database backends, as it does not depend on a special SQL
+ syntax.
+
+ .. versionadded:: 0.8
+ Support for multiple-VALUES INSERT statements.
+
+
+ .. seealso::
:ref:`inserts_and_updates` - SQL Expression
Language Tutorial
:func:`~.expression.update` - produce an ``UPDATE`` statement
"""
- if self.multi_parameters and kwargs:
- assert False
+ if self._has_multi_parameters and kwargs:
+ raise exc.InvalidRequestError(
+ "This construct already has multiple parameter sets.")
+
if args:
+ if len(args) > 1:
+ raise exc.ArgumentError(
+ "Only a single dictionary/tuple or list of "
+ "dictionaries/tuples is accepted positionally.")
v = args[0]
else:
v = {}
if self.parameters is None:
- self.parameters, self.multi_parameters = self._process_colparams(v)
- self.parameters.update(kwargs)
+ self.parameters, self._has_multi_parameters = \
+ self._process_colparams(v)
else:
- self.parameters = self.parameters.copy()
- p, mp = self._process_colparams(v)
- self.parameters.update(p)
- for p in mp:
- self.multi_parameters.update(p)
- self.parameters.update(kwargs)
+ if self._has_multi_parameters:
+ self.parameters = list(self.parameters)
+ p, self._has_multi_parameters = self._process_colparams(v)
+ if not self._has_multi_parameters:
+ raise exc.ArgumentError(
+ "Can't mix single-values and multiple values "
+ "formats in one statement")
+
+ self.parameters.extend(p)
+ else:
+ self.parameters = self.parameters.copy()
+ p, self._has_multi_parameters = self._process_colparams(v)
+ if self._has_multi_parameters:
+ raise exc.ArgumentError(
+ "Can't mix single-values and multiple values "
+ "formats in one statement")
+ self.parameters.update(p)
+
+ if kwargs:
+ if self._has_multi_parameters:
+ raise exc.ArgumentError(
+ "Can't pass kwargs and multiple parameter sets "
+ "simultaenously")
+ else:
+ self.parameters.update(kwargs)
class Insert(ValuesBase):
"""
__visit_name__ = 'insert'
+ _supports_multi_parameters = True
+
def __init__(self,
table,
values=None,
self.select = None
self.inline = inline
self._returning = returning
-
- if kwargs:
- self.kwargs = self._process_deprecated_kw(kwargs)
+ self.kwargs = kwargs
def get_children(self, **kwargs):
if self.select is not None:
else:
self._whereclause = None
self.inline = inline
+ self.kwargs = kwargs
- if kwargs:
- self.kwargs = self._process_deprecated_kw(kwargs)
def get_children(self, **kwargs):
if self._whereclause is not None:
else:
self._whereclause = None
- if kwargs:
- self.kwargs = self._process_deprecated_kw(kwargs)
+ self.kwargs = kwargs
def get_children(self, **kwargs):
if self._whereclause is not None:
"Backend does not require denormalized names."
)
+ @property
+ def multirow_inserts(self):
+ """target database must support multiple VALUES clauses in an
+ INSERT statement."""
+
+ return exclusions.skip_if(
+ lambda: not self.db.dialect.supports_multirow_insert,
+ "Backend does not support multirow inserts."
+ )
+
+
@property
def implements_get_lastrowid(self):
""""target dialect implements the executioncontext.get_lastrowid()
'(%(name)s) RETURNING length(mytable.name) '
'AS length_1', dialect=dialect)
- @testing.uses_deprecated('.*argument is deprecated. Please use '
- 'statement.returning.*')
- def test_old_returning_names(self):
- dialect = postgresql.dialect()
- table1 = table('mytable', column('myid', Integer), column('name'
- , String(128)), column('description',
- String(128)))
- u = update(table1, values=dict(name='foo'),
- postgres_returning=[table1.c.myid, table1.c.name])
- self.assert_compile(u,
- 'UPDATE mytable SET name=%(name)s '
- 'RETURNING mytable.myid, mytable.name',
- dialect=dialect)
- u = update(table1, values=dict(name='foo'),
- postgresql_returning=[table1.c.myid, table1.c.name])
- self.assert_compile(u,
- 'UPDATE mytable SET name=%(name)s '
- 'RETURNING mytable.myid, mytable.name',
- dialect=dialect)
- i = insert(table1, values=dict(name='foo'),
- postgres_returning=[table1.c.myid, table1.c.name])
- self.assert_compile(i,
- 'INSERT INTO mytable (name) VALUES '
- '(%(name)s) RETURNING mytable.myid, '
- 'mytable.name', dialect=dialect)
def test_create_partial_index(self):
m = MetaData()
c = cast(self.column, Integer)
self._do_test(c)
+
class CRUDTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = 'default'
"INSERT C D INTO mytable (myid, name, description) "
"VALUES (:myid, :name, :description)")
- def test_inline_insert(self):
+ def test_inline_default_insert(self):
metadata = MetaData()
table = Table('sometable', metadata,
Column('id', Integer, primary_key=True),
table.insert(inline=True),
"INSERT INTO sometable (foo) VALUES (foobar())", params={})
- def test_multirow_insert(self):
- data = [(1, 'a', 'b'), (2, 'a', 'b')]
+ def test_empty_insert_default(self):
+ stmt = table1.insert().values()
+ self.assert_compile(stmt, "INSERT INTO mytable () VALUES ()")
+
+ def test_empty_insert_default_values(self):
+ stmt = table1.insert().values()
+ dialect = default.DefaultDialect()
+ dialect.supports_empty_insert = dialect.supports_default_values = True
+ self.assert_compile(stmt, "INSERT INTO mytable DEFAULT VALUES",
+ dialect=dialect)
+
+ def test_empty_insert_not_supported(self):
+ stmt = table1.insert().values()
+ dialect = default.DefaultDialect()
+ dialect.supports_empty_insert = dialect.supports_default_values = False
+ assert_raises_message(
+ exc.CompileError,
+ "The 'default' dialect with current database version "
+ "settings does not support empty inserts.",
+ stmt.compile, dialect=dialect
+ )
+
+ def test_multirow_insert_not_supported(self):
+ stmt = table1.insert().values([{"myid": 1}, {"myid": 2}])
+ dialect = default.DefaultDialect()
+ assert_raises_message(
+ exc.CompileError,
+ "The 'default' dialect with current database version settings "
+ "does not support in-place multirow inserts.",
+ stmt.compile, dialect=dialect
+ )
+
+ def test_multirow_insert_named(self):
+ stmt = table1.insert().\
+ values([{"myid": 1, "name": 'a', "description": 'b'},
+ {"myid": 2, "name": 'c', "description": 'd'},
+ {"myid": 3, "name": 'e', "description": 'f'}
+ ])
+
result = "INSERT INTO mytable (myid, name, description) VALUES " \
- "(%(myid)s, %(name)s, %(description)s), " \
- "(%(myid0)s, %(name0)s, %(description0)s)"
+ "(:myid_0, :name_0, :description_0), " \
+ "(:myid_1, :name_1, :description_1), " \
+ "(:myid_2, :name_2, :description_2)"
- stmt = insert(table1, data, dialect='postgresql')
- self.assert_compile(stmt, result, dialect=postgresql.dialect())
+ dialect = default.DefaultDialect()
+ dialect.supports_multirow_insert = True
+ self.assert_compile(stmt, result,
+ checkparams={
+ 'description_2': 'f', 'name_2': 'e',
+ 'name_0': 'a', 'name_1': 'c', 'myid_2': 3,
+ 'description_0': 'b', 'myid_0': 1,
+ 'myid_1': 2, 'description_1': 'd'
+ },
+ dialect=dialect)
+
+ def test_multirow_insert_positional(self):
+ stmt = table1.insert().\
+ values([{"myid": 1, "name": 'a', "description": 'b'},
+ {"myid": 2, "name": 'c', "description": 'd'},
+ {"myid": 3, "name": 'e', "description": 'f'}
+ ])
+
+ result = "INSERT INTO mytable (myid, name, description) VALUES " \
+ "(%s, %s, %s), " \
+ "(%s, %s, %s), " \
+ "(%s, %s, %s)" \
- stmt = table1.insert(values=data, dialect='postgresql')
- self.assert_compile(stmt, result, dialect=postgresql.dialect())
+ dialect = default.DefaultDialect()
+ dialect.supports_multirow_insert = True
+ dialect.paramstyle = "format"
+ dialect.positional = True
+ self.assert_compile(stmt, result,
+ checkpositional=(1, 'a', 'b', 2, 'c', 'd', 3, 'e', 'f'),
+ dialect=dialect)
- stmt = table1.insert(dialect='postgresql').values(data)
- self.assert_compile(stmt, result, dialect=postgresql.dialect())
+ def test_multirow_inline_default_insert(self):
+ metadata = MetaData()
+ table = Table('sometable', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('data', String),
+ Column('foo', Integer, default=func.foobar()))
+
+ stmt = table.insert().\
+ values([
+ {"id": 1, "data": "data1"},
+ {"id": 2, "data": "data2", "foo": "plainfoo"},
+ {"id": 3, "data": "data3"},
+ ])
+ result = "INSERT INTO sometable (id, data, foo) VALUES "\
+ "(%(id_0)s, %(data_0)s, foobar()), "\
+ "(%(id_1)s, %(data_1)s, %(foo_1)s), "\
+ "(%(id_2)s, %(data_2)s, foobar())"
+
+ self.assert_compile(stmt, result,
+ checkparams={'data_2': 'data3', 'id_0': 1, 'id_2': 3,
+ 'foo_1': 'plainfoo', 'data_1': 'data2',
+ 'id_1': 2, 'data_0': 'data1'},
+ dialect=postgresql.dialect())
+
+ def test_multirow_server_default_insert(self):
+ metadata = MetaData()
+ table = Table('sometable', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('data', String),
+ Column('foo', Integer, server_default=func.foobar()))
+
+ stmt = table.insert().\
+ values([
+ {"id": 1, "data": "data1"},
+ {"id": 2, "data": "data2", "foo": "plainfoo"},
+ {"id": 3, "data": "data3"},
+ ])
+ result = "INSERT INTO sometable (id, data) VALUES "\
+ "(%(id_0)s, %(data_0)s), "\
+ "(%(id_1)s, %(data_1)s), "\
+ "(%(id_2)s, %(data_2)s)"
+
+ self.assert_compile(stmt, result,
+ checkparams={'data_2': 'data3', 'id_0': 1, 'id_2': 3,
+ 'data_1': 'data2',
+ 'id_1': 2, 'data_0': 'data1'},
+ dialect=postgresql.dialect())
+
+ stmt = table.insert().\
+ values([
+ {"id": 1, "data": "data1", "foo": "plainfoo"},
+ {"id": 2, "data": "data2"},
+ {"id": 3, "data": "data3", "foo": "otherfoo"},
+ ])
+
+ # note the effect here is that the first set of params
+ # takes effect for the rest of them, when one is absent
+ result = "INSERT INTO sometable (id, data, foo) VALUES "\
+ "(%(id_0)s, %(data_0)s, %(foo_0)s), "\
+ "(%(id_1)s, %(data_1)s, %(foo_0)s), "\
+ "(%(id_2)s, %(data_2)s, %(foo_2)s)"
+
+ self.assert_compile(stmt, result,
+ checkparams={'data_2': 'data3', 'id_0': 1, 'id_2': 3,
+ 'data_1': 'data2',
+ "foo_0": "plainfoo",
+ "foo_2": "otherfoo",
+ 'id_1': 2, 'data_0': 'data1'},
+ dialect=postgresql.dialect())
def test_update(self):
self.assert_compile(
from sqlalchemy import testing
from sqlalchemy.sql.visitors import ClauseVisitor, CloningVisitor, \
cloned_traverse, ReplacingCloningVisitor
-from sqlalchemy import util, exc
+from sqlalchemy import exc
from sqlalchemy.sql import util as sql_util
-from sqlalchemy.testing import eq_, ne_, assert_raises
+from sqlalchemy.testing import eq_, is_, assert_raises, assert_raises_message
class TraversalTest(fixtures.TestBase, AssertsExecutionResults):
"""test ClauseVisitor's traversal, particularly its
s = text('select 42', execution_options=dict(foo='bar'))
assert s._execution_options == dict(foo='bar')
-class InsertTest(fixtures.TestBase, AssertsCompiledSQL):
- """Tests the generative capability of Insert"""
+class ValuesBaseTest(fixtures.TestBase, AssertsCompiledSQL):
+ """Tests the generative capability of Insert, Update"""
__dialect__ = 'default'
"table1 (col1, col2, col3) "
"VALUES (:col1, :col2, :col3)")
+ def test_add_kwarg(self):
+ i = t1.insert()
+ eq_(i.parameters, None)
+ i = i.values(col1=5)
+ eq_(i.parameters, {"col1": 5})
+ i = i.values(col2=7)
+ eq_(i.parameters, {"col1": 5, "col2": 7})
+
+ def test_via_tuple_single(self):
+ i = t1.insert()
+ eq_(i.parameters, None)
+ i = i.values((5, 6, 7))
+ eq_(i.parameters, {"col1": 5, "col2": 6, "col3": 7})
+
+ def test_kw_and_dict_simulatenously_single(self):
+ i = t1.insert()
+ i = i.values({"col1": 5}, col2=7)
+ eq_(i.parameters, {"col1": 5, "col2": 7})
+
+ def test_via_tuple_multi(self):
+ i = t1.insert()
+ eq_(i.parameters, None)
+ i = i.values([(5, 6, 7), (8, 9, 10)])
+ eq_(i.parameters, [
+ {"col1": 5, "col2": 6, "col3": 7},
+ {"col1": 8, "col2": 9, "col3": 10},
+ ]
+ )
+
+ def test_inline_values_single(self):
+ i = t1.insert(values={"col1": 5})
+ eq_(i.parameters, {"col1": 5})
+ is_(i._has_multi_parameters, False)
+
+ def test_inline_values_multi(self):
+ i = t1.insert(values=[{"col1": 5}, {"col1": 6}])
+ eq_(i.parameters, [{"col1": 5}, {"col1": 6}])
+ is_(i._has_multi_parameters, True)
+
+ def test_add_dictionary(self):
+ i = t1.insert()
+ eq_(i.parameters, None)
+ i = i.values({"col1": 5})
+ eq_(i.parameters, {"col1": 5})
+ is_(i._has_multi_parameters, False)
+
+ i = i.values({"col1": 6})
+ # note replaces
+ eq_(i.parameters, {"col1": 6})
+ is_(i._has_multi_parameters, False)
+
+ i = i.values({"col2": 7})
+ eq_(i.parameters, {"col1": 6, "col2": 7})
+ is_(i._has_multi_parameters, False)
+
+ def test_add_kwarg_disallowed_multi(self):
+ i = t1.insert()
+ i = i.values([{"col1": 5}, {"col1": 7}])
+ assert_raises_message(
+ exc.InvalidRequestError,
+ "This construct already has multiple parameter sets.",
+ i.values, col2=7
+ )
+
+ def test_cant_mix_single_multi_formats_dict_to_list(self):
+ i = t1.insert().values(col1=5)
+ assert_raises_message(
+ exc.ArgumentError,
+ "Can't mix single-values and multiple values "
+ "formats in one statement",
+ i.values, [{"col1": 6}]
+ )
+
+ def test_cant_mix_single_multi_formats_list_to_dict(self):
+ i = t1.insert().values([{"col1": 6}])
+ assert_raises_message(
+ exc.ArgumentError,
+ "Can't mix single-values and multiple values "
+ "formats in one statement",
+ i.values, {"col1": 5}
+ )
+
+ def test_erroneous_multi_args_dicts(self):
+ i = t1.insert()
+ assert_raises_message(
+ exc.ArgumentError,
+ "Only a single dictionary/tuple or list of "
+ "dictionaries/tuples is accepted positionally.",
+ i.values, {"col1": 5}, {"col1": 7}
+ )
+
+ def test_erroneous_multi_args_tuples(self):
+ i = t1.insert()
+ assert_raises_message(
+ exc.ArgumentError,
+ "Only a single dictionary/tuple or list of "
+ "dictionaries/tuples is accepted positionally.",
+ i.values, (5, 6, 7), (8, 9, 10)
+ )
+
+ def test_erroneous_multi_args_plus_kw(self):
+ i = t1.insert()
+ assert_raises_message(
+ exc.ArgumentError,
+ "Can't pass kwargs and multiple parameter sets simultaenously",
+ i.values, [{"col1": 5}], col2=7
+ )
+
+ def test_update_no_support_multi_values(self):
+ u = t1.update()
+ assert_raises_message(
+ exc.InvalidRequestError,
+ "This construct does not support multiple parameter sets.",
+ u.values, [{"col1": 5}, {"col1": 7}]
+ )
+
+ def test_update_no_support_multi_constructor(self):
+ assert_raises_message(
+ exc.InvalidRequestError,
+ "This construct does not support multiple parameter sets.",
+ t1.update, values=[{"col1": 5}, {"col1": 7}]
+ )
def teardown_class(cls):
metadata.drop_all()
+ @testing.requires.multirow_inserts
def test_multirow_insert(self):
users.insert(values=[{'user_id':7, 'user_name':'jack'},
{'user_id':8, 'user_name':'ed'}]).execute()
eq_(result.fetchall(), [(1,)])
- @testing.fails_on('postgresql', '')
- @testing.fails_on('oracle+cx_oracle', '')
- @testing.crashes('mssql+mxodbc', 'Raises an error')
- def test_executemany():
- # return value is documented as failing with psycopg2/executemany
- result2 = table.insert().returning(table).execute(
- [{'persons': 2, 'full': False}, {'persons': 3, 'full': True}])
-
- if testing.against('mssql+zxjdbc'):
- # jtds apparently returns only the first row
- eq_(result2.fetchall(), [(2, 2, False, None)])
- elif testing.against('firebird', 'mssql', 'oracle'):
- # Multiple inserts only return the last row
- eq_(result2.fetchall(), [(3, 3, True, None)])
- else:
- # nobody does this as far as we know (pg8000?)
- eq_(result2.fetchall(), [(2, 2, False, None), (3, 3, True, None)])
-
- test_executemany()
+ @testing.fails_on('postgresql', 'undefined behavior')
+ @testing.fails_on('oracle+cx_oracle', 'undefined behavior')
+ @testing.crashes('mssql+mxodbc', 'Raises an error')
+ def test_insert_returning_execmany(self):
+
+ # return value is documented as failing with psycopg2/executemany
+ result2 = table.insert().returning(table).execute(
+ [{'persons': 2, 'full': False}, {'persons': 3, 'full': True}])
+
+ if testing.against('mssql+zxjdbc'):
+ # jtds apparently returns only the first row
+ eq_(result2.fetchall(), [(2, 2, False, None)])
+ elif testing.against('firebird', 'mssql', 'oracle'):
+ # Multiple inserts only return the last row
+ eq_(result2.fetchall(), [(3, 3, True, None)])
+ else:
+ # nobody does this as far as we know (pg8000?)
+ eq_(result2.fetchall(), [(2, 2, False, None), (3, 3, True, None)])
+
+
+ @testing.requires.multirow_inserts
+ def test_multirow_returning(self):
+ ins = table.insert().returning(table.c.id, table.c.persons).values(
+ [
+ {'persons': 1, 'full': False},
+ {'persons': 2, 'full': True},
+ {'persons': 3, 'full': False},
+ ]
+ )
+ result = testing.db.execute(ins)
+ eq_(
+ result.fetchall(),
+ [(1, 1), (2, 2), (3, 3)]
+ )
def test_no_ipk_on_returning(self):
result = testing.db.execute(
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
