.. changelog::
:version: 0.8.3
+ .. change::
+ :tags: feature, sql
+ :tickets: 722
+
+ Added new method to the :func:`.insert` construct
+ :meth:`.Insert.from_select`. Given a list of columns and
+ a selectable, renders ``INSERT INTO (table) (columns) SELECT ..``.
+
+ .. change::
+ :tags: feature, sql
+
+ The :func:`.update`, :func:`.insert`, and :func:`.delete` constructs
+ will now interpret ORM entities as FROM clauses to be operated upon,
+ e.g.::
+
+ from sqlalchemy import insert, update, delete
+
+ ins = insert(SomeMappedClass).values(x=5)
+
+ del_ = delete(SomeMappedClass).where(SomeMappedClass.id == 5)
+
+ upd = update(SomeMappedClass).where(SomeMappedClass.id == 5).values(name='ed')
+
.. change::
:tags: bug, orm
:tickets: 2773
"INSERT INTO mytable (SELECT mytable.x, mytable.y, mytable.z FROM mytable WHERE mytable.x > :x_1)"
+.. note::
+
+ The above ``InsertFromSelect`` construct is only an example, this actual
+ functionality is already available using the
+ :meth:`.Insert.from_select` method.
+
.. note::
The above ``InsertFromSelect`` construct probably wants to have "autocommit"
if self.returning_precedes_values:
text += " " + returning_clause
- if not colparams and supports_default_values:
+ if insert_stmt.select is not None:
+ text += " %s" % self.process(insert_stmt.select, **kw)
+ elif not colparams and supports_default_values:
text += " DEFAULT VALUES"
elif insert_stmt._has_multi_parameters:
text += " VALUES %s" % (
return insp.selectable
raise exc.ArgumentError("FROM expression expected")
+def _interpret_as_select(element):
+ element = _interpret_as_from(element)
+ if isinstance(element, Alias):
+ element = element.original
+ if not isinstance(element, Select):
+ element = element.select()
+ return element
+
def _const_expr(element):
if isinstance(element, (Null, False_, True_)):
_supports_multi_parameters = False
_has_multi_parameters = False
+ select = None
def __init__(self, table, values, prefixes):
- self.table = table
+ self.table = _interpret_as_from(table)
self.parameters, self._has_multi_parameters = \
self._process_colparams(values)
if prefixes:
:func:`~.expression.update` - produce an ``UPDATE`` statement
"""
+ if self.select is not None:
+ raise exc.InvalidRequestError(
+ "This construct already inserts from a SELECT")
if self._has_multi_parameters and kwargs:
raise exc.InvalidRequestError(
"This construct already has multiple parameter sets.")
else:
return ()
+ @_generative
+ def from_select(self, names, select):
+ """Return a new :class:`.Insert` construct which represents
+ an ``INSERT...FROM SELECT`` statement.
+
+ e.g.::
+
+ sel = select([table1.c.a, table1.c.b]).where(table1.c.c > 5)
+ ins = table2.insert().from_select(['a', 'b'], sel)
+
+ :param names: a sequence of string column names or :class:`.Column`
+ objects representing the target columns.
+ :param select: a :func:`.select` construct, :class:`.FromClause`
+ or other construct which resolves into a :class:`.FromClause`,
+ such as an ORM :class:`.Query` object, etc. The order of
+ columns returned from this FROM clause should correspond to the
+ order of columns sent as the ``names`` parameter; while this
+ is not checked before passing along to the database, the database
+ would normally raise an exception if these column lists don't
+ correspond.
+
+ .. versionadded:: 0.8.3
+
+ """
+ if self.parameters:
+ raise exc.InvalidRequestError(
+ "This construct already inserts value expressions")
+
+ self.parameters, self._has_multi_parameters = \
+ self._process_colparams(dict((n, null()) for n in names))
+
+ self.select = _interpret_as_select(select)
+
def _copy_internals(self, clone=_clone, **kw):
# TODO: coverage
self.parameters = self.parameters.copy()
+ if self.select is not None:
+ self.select = _clone(self.select)
class Update(ValuesBase):
prefixes=None,
**kwargs):
self._bind = bind
- self.table = table
+ self.table = _interpret_as_from(table)
self._returning = returning
if prefixes:
"empty inserts not supported"
)
+ @property
+ def insert_from_select(self):
+ """target platform supports INSERT from a SELECT."""
+
+ return exclusions.open()
+
@property
def returning(self):
"""target platform supports RETURNING."""
assert len(r.fetchall())
+ @requirements.insert_from_select
+ def test_insert_from_select(self):
+ table = self.tables.autoinc_pk
+ config.db.execute(
+ table.insert(),
+ [
+ dict(data="data1"),
+ dict(data="data2"),
+ dict(data="data3"),
+ ]
+ )
+
+
+ config.db.execute(
+ table.insert().
+ from_select(
+ ("data",), select([table.c.data]).where(
+ table.c.data.in_(["data2", "data3"]))
+ ),
+ )
+
+ eq_(
+ config.db.execute(
+ select([table.c.data]).order_by(table.c.data)
+ ).fetchall(),
+ [("data1", ), ("data2", ), ("data2", ),
+ ("data3", ), ("data3", )]
+ )
class ReturningTest(fixtures.TablesTest):
run_deletes = 'each'
import operator
from sqlalchemy import MetaData, null, exists, text, union, literal, \
literal_column, func, between, Unicode, desc, and_, bindparam, \
- select, distinct, or_, collate
+ select, distinct, or_, collate, insert
from sqlalchemy import inspect
from sqlalchemy import exc as sa_exc, util
from sqlalchemy.sql import compiler, table, column
"JOIN addresses ON users.id = addresses.user_id"
)
+ def test_insert_from_query(self):
+ User = self.classes.User
+ Address = self.classes.Address
+
+ s = Session()
+ q = s.query(User.id, User.name).filter_by(name='ed')
+ self.assert_compile(
+ insert(Address).from_select(('id', 'email_address'), q),
+ "INSERT INTO addresses (id, email_address) "
+ "SELECT users.id AS users_id, users.name AS users_name "
+ "FROM users WHERE users.name = :name_1"
+ )
+
+ def test_insert_from_query_col_attr(self):
+ User = self.classes.User
+ Address = self.classes.Address
+
+ s = Session()
+ q = s.query(User.id, User.name).filter_by(name='ed')
+ self.assert_compile(
+ insert(Address).from_select(
+ (Address.id, Address.email_address), q),
+ "INSERT INTO addresses (id, email_address) "
+ "SELECT users.id AS users_id, users.name AS users_name "
+ "FROM users WHERE users.name = :name_1"
+ )
+
+ def test_update_from_entity(self):
+ from sqlalchemy.sql import update
+ User = self.classes.User
+ self.assert_compile(
+ update(User),
+ "UPDATE users SET id=:id, name=:name"
+ )
+
+ self.assert_compile(
+ update(User).values(name='ed').where(User.id == 5),
+ "UPDATE users SET name=:name WHERE users.id = :id_1",
+ checkparams={"id_1": 5, "name": "ed"}
+ )
+
+ def test_delete_from_entity(self):
+ from sqlalchemy.sql import delete
+ User = self.classes.User
+ self.assert_compile(
+ delete(User),
+ "DELETE FROM users"
+ )
+
+ self.assert_compile(
+ delete(User).where(User.id == 5),
+ "DELETE FROM users WHERE users.id = :id_1",
+ checkparams={"id_1": 5}
+ )
+
+ def test_insert_from_entity(self):
+ from sqlalchemy.sql import insert
+ User = self.classes.User
+ self.assert_compile(
+ insert(User),
+ "INSERT INTO users (id, name) VALUES (:id, :name)"
+ )
+
+ self.assert_compile(
+ insert(User).values(name="ed"),
+ "INSERT INTO users (name) VALUES (:name)",
+ checkparams={"name": "ed"}
+ )
+
class GetTest(QueryTest):
def test_get(self):
User = self.classes.User
#! coding:utf-8
from sqlalchemy import Column, Integer, MetaData, String, Table,\
- bindparam, exc, func, insert
+ bindparam, exc, func, insert, select
from sqlalchemy.dialects import mysql, postgresql
from sqlalchemy.engine import default
from sqlalchemy.testing import AssertsCompiledSQL,\
dialect=default.DefaultDialect()
)
+ def test_insert_from_select_select(self):
+ table1 = self.tables.mytable
+ sel = select([table1.c.myid, table1.c.name]).where(table1.c.name == 'foo')
+ ins = self.tables.myothertable.insert().\
+ from_select(("otherid", "othername"), sel)
+ self.assert_compile(
+ ins,
+ "INSERT INTO myothertable (otherid, othername) "
+ "SELECT mytable.myid, mytable.name FROM mytable "
+ "WHERE mytable.name = :name_1",
+ checkparams={"name_1": "foo"}
+ )
+
+ def test_insert_mix_select_values_exception(self):
+ table1 = self.tables.mytable
+ sel = select([table1.c.myid, table1.c.name]).where(table1.c.name == 'foo')
+ ins = self.tables.myothertable.insert().\
+ from_select(("otherid", "othername"), sel)
+ assert_raises_message(
+ exc.InvalidRequestError,
+ "This construct already inserts from a SELECT",
+ ins.values, othername="5"
+ )
+
+ def test_insert_mix_values_select_exception(self):
+ table1 = self.tables.mytable
+ sel = select([table1.c.myid, table1.c.name]).where(table1.c.name == 'foo')
+ ins = self.tables.myothertable.insert().values(othername="5")
+ assert_raises_message(
+ exc.InvalidRequestError,
+ "This construct already inserts value expressions",
+ ins.from_select, ("otherid", "othername"), sel
+ )
+
+ def test_insert_from_select_table(self):
+ table1 = self.tables.mytable
+ ins = self.tables.myothertable.insert().\
+ from_select(("otherid", "othername"), table1)
+ # note we aren't checking the number of columns right now
+ self.assert_compile(
+ ins,
+ "INSERT INTO myothertable (otherid, othername) "
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable",
+ checkparams={}
+ )
+
+
+ def test_insert_from_select_col_values(self):
+ table1 = self.tables.mytable
+ table2 = self.tables.myothertable
+ sel = select([table1.c.myid, table1.c.name]).where(table1.c.name == 'foo')
+ ins = table2.insert().\
+ from_select((table2.c.otherid, table2.c.othername), sel)
+ self.assert_compile(
+ ins,
+ "INSERT INTO myothertable (otherid, othername) "
+ "SELECT mytable.myid, mytable.name FROM mytable "
+ "WHERE mytable.name = :name_1",
+ checkparams={"name_1": "foo"}
+ )
+
+
class EmptyTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
__dialect__ = 'default'
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
