of statements have more consistent behavior w.r.t.
parenthesizing. Each compound element embedded within
another will now be grouped with parenthesis - previously,
the first compound element in the list would not be grouped,
as SQLite doesn't like a statement to start with
parenthesis. However, Postgresql in particular has
precedence rules regarding INTERSECT, and it is
more consistent for parenthesis to be applied equally
to all sub-elements. So now, the workaround for SQLite
is also what the workaround for PG was previously -
when nesting compound elements, the first one usually needs
".alias().select()" called on it to wrap it inside
of a subquery. [ticket:1665]
version in use supports it (a version number check is
performed). This occurs if no end-user returning() was
specified.
-
+
+ - union(), intersect(), except() and other "compound" types
+ of statements have more consistent behavior w.r.t.
+ parenthesizing. Each compound element embedded within
+ another will now be grouped with parenthesis - previously,
+ the first compound element in the list would not be grouped,
+ as SQLite doesn't like a statement to start with
+ parenthesis. However, Postgresql in particular has
+ precedence rules regarding INTERSECT, and it is
+ more consistent for parenthesis to be applied equally
+ to all sub-elements. So now, the workaround for SQLite
+ is also what the workaround for PG was previously -
+ when nesting compound elements, the first one usually needs
+ ".alias().select()" called on it to wrap it inside
+ of a subquery. [ticket:1665]
+
- insert() and update() constructs can now embed bindparam()
objects using names that match the keys of columns. These
bind parameters will circumvent the usual route to those
{sql}>>> metadata.create_all(engine) #doctest: +NORMALIZE_WHITESPACE
PRAGMA table_info("users")
- {}
+ ()
PRAGMA table_info("addresses")
- {}
+ ()
CREATE TABLE users (
id INTEGER NOT NULL,
name VARCHAR,
fullname VARCHAR,
PRIMARY KEY (id)
)
- {}
+ ()
COMMIT
CREATE TABLE addresses (
id INTEGER NOT NULL,
user_id INTEGER,
email_address VARCHAR NOT NULL,
PRIMARY KEY (id),
- FOREIGN KEY(user_id) REFERENCES users (id)
+ FOREIGN KEY(user_id) REFERENCES users (id)
)
- {}
+ ()
COMMIT
Users familiar with the syntax of CREATE TABLE may notice that the VARCHAR columns were generated without a length; on SQLite, this is a valid datatype, but on most databases it's not allowed. So if running this tutorial on a database such as PostgreSQL or MySQL, and you wish to use SQLAlchemy to generate the tables, a "length" may be provided to the ``String`` type as below::
By "generates", we mean that **any** SQL function is created based on the word you choose::
- >>> print func.xyz_my_goofy_function()
+ >>> print func.xyz_my_goofy_function() # doctest: +NORMALIZE_WHITESPACE
xyz_my_goofy_function()
Certain function names are known by SQLAlchemy, allowing special behavioral rules to be applied. Some for example are "ANSI" functions, which mean they don't get the parenthesis added after them, such as CURRENT_TIMESTAMP:
Unions and Other Set Operations
-------------------------------
-
Unions come in two flavors, UNION and UNION ALL, which are available via module level functions:
.. sourcecode:: pycon+sql
['%@%.com', '%@msn.com']
{stop}[(1, 1, u'jack@yahoo.com'), (4, 2, u'wendy@aol.com')]
+A common issue with so-called "compound" selectables arises due to the fact that they nest with parenthesis. SQLite in particular doesn't like a statement that starts with parenthesis. So when nesting a "compound" inside a "compound", it's often necessary to apply
+``.alias().select()`` to the first element of the outermost compound, if that element is also a compount. For example, to nest a "union" and a "select" inside of "except\_", SQLite will want
+the "union" to be stated as a subquery:
+
+.. sourcecode:: pycon+sql
+
+ >>> u = except_(
+ ... union(
+ ... addresses.select(addresses.c.email_address.like('%@yahoo.com')),
+ ... addresses.select(addresses.c.email_address.like('%@msn.com'))
+ ... ).alias().select(), # apply subquery here
+ ... addresses.select(addresses.c.email_address.like('%@msn.com'))
+ ... )
+ {sql}>>> print conn.execute(u).fetchall() # doctest: +NORMALIZE_WHITESPACE
+ SELECT anon_1.id, anon_1.user_id, anon_1.email_address
+ FROM (SELECT addresses.id AS id, addresses.user_id AS user_id,
+ addresses.email_address AS email_address FROM addresses
+ WHERE addresses.email_address LIKE ? UNION SELECT addresses.id AS id,
+ addresses.user_id AS user_id, addresses.email_address AS email_address
+ FROM addresses WHERE addresses.email_address LIKE ?) AS anon_1 EXCEPT
+ SELECT addresses.id, addresses.user_id, addresses.email_address
+ FROM addresses
+ WHERE addresses.email_address LIKE ?
+ ['%@yahoo.com', '%@msn.com', '%@msn.com']
+ {stop}[(1, 1, u'jack@yahoo.com')]
+
+
Scalar Selects
--------------
self.isinsert = compiled.isinsert
self.isupdate = compiled.isupdate
self.isdelete = compiled.isdelete
- self.execution_options =\
- compiled.statement._execution_options.union(self.execution_options)
+ if compiled.statement._execution_options:
+ self.execution_options =\
+ compiled.statement._execution_options.union(self.execution_options)
if not parameters:
self.compiled_parameters = [compiled.construct_params()]
(1, len(self.selects[0].c), n+1, len(s.c))
)
- # unions group from left to right, so don't group first select
- if n:
- self.selects.append(s.self_group(self))
- else:
- self.selects.append(s)
+ self.selects.append(s.self_group(self))
_SelectBaseMixin.__init__(self, **kwargs)
@testing.crashes('oracle', 'FIXME: unknown, verify not fails_on')
@testing.crashes('sybase', 'FIXME: unknown, verify not fails_on')
@testing.fails_on('mysql', 'FIXME: unknown')
+ @testing.fails_on('sqlite', "Can't handle this style of nesting")
def test_except_style1(self):
e = except_(union(
select([t1.c.col3, t1.c.col4]),
wanted = [('aaa', 'aaa'), ('aaa', 'ccc'), ('bbb', 'aaa'),
('bbb', 'bbb'), ('ccc', 'bbb'), ('ccc', 'ccc')]
- found = self._fetchall_sorted(e.alias('bar').select().execute())
+ found = self._fetchall_sorted(e.alias().select().execute())
eq_(found, wanted)
@testing.crashes('firebird', 'Does not support except')
@testing.crashes('sybase', 'FIXME: unknown, verify not fails_on')
@testing.fails_on('mysql', 'FIXME: unknown')
def test_except_style2(self):
+ # same as style1, but add alias().select() to the except_().
+ # sqlite can handle it now.
+
e = except_(union(
select([t1.c.col3, t1.c.col4]),
select([t2.c.col3, t2.c.col4]),
select([t3.c.col3, t3.c.col4]),
- ).alias('foo').select(), select([t2.c.col3, t2.c.col4]))
+ ).alias().select(), select([t2.c.col3, t2.c.col4]))
wanted = [('aaa', 'aaa'), ('aaa', 'ccc'), ('bbb', 'aaa'),
('bbb', 'bbb'), ('ccc', 'bbb'), ('ccc', 'ccc')]
found1 = self._fetchall_sorted(e.execute())
eq_(found1, wanted)
- found2 = self._fetchall_sorted(e.alias('bar').select().execute())
+ found2 = self._fetchall_sorted(e.alias().select().execute())
eq_(found2, wanted)
@testing.crashes('firebird', 'Does not support except')
@testing.crashes('oracle', 'FIXME: unknown, verify not fails_on')
@testing.crashes('sybase', 'FIXME: unknown, verify not fails_on')
@testing.fails_on('mysql', 'FIXME: unknown')
- @testing.fails_on('sqlite', 'FIXME: unknown')
+ @testing.fails_on('sqlite', "Can't handle this style of nesting")
def test_except_style3(self):
# aaa, bbb, ccc - (aaa, bbb, ccc - (ccc)) = ccc
e = except_(
eq_(e.alias('foo').select().execute().fetchall(),
[('ccc',)])
+ @testing.crashes('firebird', 'Does not support except')
+ @testing.crashes('oracle', 'FIXME: unknown, verify not fails_on')
+ @testing.crashes('sybase', 'FIXME: unknown, verify not fails_on')
+ @testing.fails_on('mysql', 'FIXME: unknown')
+ def test_except_style4(self):
+ # aaa, bbb, ccc - (aaa, bbb, ccc - (ccc)) = ccc
+ e = except_(
+ select([t1.c.col3]), # aaa, bbb, ccc
+ except_(
+ select([t2.c.col3]), # aaa, bbb, ccc
+ select([t3.c.col3], t3.c.col3 == 'ccc'), #ccc
+ ).alias().select()
+ )
+
+ eq_(e.execute().fetchall(), [('ccc',)])
+ eq_(
+ e.alias().select().execute().fetchall(),
+ [('ccc',)]
+ )
+
+ @testing.crashes('firebird', 'Does not support intersect')
+ @testing.fails_on('mysql', 'FIXME: unknown')
+ @testing.fails_on('sqlite', "sqlite can't handle leading parenthesis")
+ def test_intersect_unions(self):
+ u = intersect(
+ union(
+ select([t1.c.col3, t1.c.col4]),
+ select([t3.c.col3, t3.c.col4]),
+ ),
+ union(
+ select([t2.c.col3, t2.c.col4]),
+ select([t3.c.col3, t3.c.col4]),
+ ).alias().select()
+ )
+ wanted = [('aaa', 'ccc'), ('bbb', 'aaa'), ('ccc', 'bbb')]
+ found = self._fetchall_sorted(u.execute())
+
+ eq_(found, wanted)
+
+ @testing.crashes('firebird', 'Does not support intersect')
+ @testing.fails_on('mysql', 'FIXME: unknown')
+ def test_intersect_unions_2(self):
+ u = intersect(
+ union(
+ select([t1.c.col3, t1.c.col4]),
+ select([t3.c.col3, t3.c.col4]),
+ ).alias().select(),
+ union(
+ select([t2.c.col3, t2.c.col4]),
+ select([t3.c.col3, t3.c.col4]),
+ ).alias().select()
+ )
+ wanted = [('aaa', 'ccc'), ('bbb', 'aaa'), ('ccc', 'bbb')]
+ found = self._fetchall_sorted(u.execute())
+
+ eq_(found, wanted)
+
@testing.crashes('firebird', 'Does not support intersect')
@testing.fails_on('mysql', 'FIXME: unknown')
- def test_composite(self):
+ def test_intersect(self):
u = intersect(
select([t2.c.col3, t2.c.col4]),
union(
select([t1.c.col3, t1.c.col4]),
select([t2.c.col3, t2.c.col4]),
select([t3.c.col3, t3.c.col4]),
- ).alias('foo').select()
+ ).alias().select()
)
wanted = [('aaa', 'bbb'), ('bbb', 'ccc'), ('ccc', 'aaa')]
found = self._fetchall_sorted(u.execute())
select([t1.c.col3, t1.c.col4]),
select([t2.c.col3, t2.c.col4]),
select([t3.c.col3, t3.c.col4]),
- ).alias('foo').select()
- ).alias('bar')
+ ).alias().select()
+ ).alias()
wanted = [('aaa', 'bbb'), ('bbb', 'ccc'), ('ccc', 'aaa')]
found = self._fetchall_sorted(ua.select().execute())
order_by = [table1.c.myid],
)
- self.assert_compile(x, "SELECT mytable.myid, mytable.name, mytable.description \
-FROM mytable WHERE mytable.myid = :myid_1 UNION \
-SELECT mytable.myid, mytable.name, mytable.description \
-FROM mytable WHERE mytable.myid = :myid_2 ORDER BY mytable.myid")
+ self.assert_compile(x, "SELECT mytable.myid, mytable.name, mytable.description "\
+ "FROM mytable WHERE mytable.myid = :myid_1 UNION "\
+ "SELECT mytable.myid, mytable.name, mytable.description "\
+ "FROM mytable WHERE mytable.myid = :myid_2 ORDER BY mytable.myid")
+ x = union(
+ select([table1]),
+ select([table1])
+ )
+ x = union(x, select([table1]))
+ self.assert_compile(x, "(SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable UNION SELECT mytable.myid, mytable.name, "
+ "mytable.description FROM mytable) UNION SELECT mytable.myid,"
+ " mytable.name, mytable.description FROM mytable")
+
u1 = union(
select([table1.c.myid, table1.c.name]),
select([table2]),
select([table3])
)
- self.assert_compile(u1,
- "SELECT mytable.myid, mytable.name \
-FROM mytable UNION SELECT myothertable.otherid, myothertable.othername \
-FROM myothertable UNION SELECT thirdtable.userid, thirdtable.otherstuff FROM thirdtable")
+ self.assert_compile(u1, "SELECT mytable.myid, mytable.name "
+ "FROM mytable UNION SELECT myothertable.otherid, "
+ "myothertable.othername FROM myothertable "
+ "UNION SELECT thirdtable.userid, thirdtable.otherstuff "
+ "FROM thirdtable")
assert u1.corresponding_column(table2.c.otherid) is u1.c.myid
order_by=['myid'],
offset=10,
limit=5
- )
- , "SELECT mytable.myid, mytable.name \
-FROM mytable UNION SELECT myothertable.otherid, myothertable.othername \
-FROM myothertable ORDER BY myid LIMIT 5 OFFSET 10"
+ ),
+ "SELECT mytable.myid, mytable.name "
+ "FROM mytable UNION SELECT myothertable.otherid, myothertable.othername "
+ "FROM myothertable ORDER BY myid LIMIT 5 OFFSET 10"
)
self.assert_compile(
union(
- select([table1.c.myid, table1.c.name, func.max(table1.c.description)], table1.c.name=='name2', group_by=[table1.c.myid, table1.c.name]),
+ select([table1.c.myid, table1.c.name, func.max(table1.c.description)],
+ table1.c.name=='name2',
+ group_by=[table1.c.myid, table1.c.name]),
table1.select(table1.c.name=='name1')
- )
- ,
- "SELECT mytable.myid, mytable.name, max(mytable.description) AS max_1 FROM mytable \
-WHERE mytable.name = :name_1 GROUP BY mytable.myid, mytable.name UNION SELECT mytable.myid, mytable.name, mytable.description \
-FROM mytable WHERE mytable.name = :name_2"
+ ),
+ "SELECT mytable.myid, mytable.name, max(mytable.description) AS max_1 "
+ "FROM mytable WHERE mytable.name = :name_1 GROUP BY mytable.myid, "
+ "mytable.name UNION SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.name = :name_2"
)
self.assert_compile(
)
)
,
- "SELECT mytable.myid FROM mytable UNION ALL (SELECT myothertable.otherid FROM myothertable UNION \
-SELECT thirdtable.userid FROM thirdtable)"
- )
- # This doesn't need grouping, so don't group to not give sqlite unnecessarily hard time
- self.assert_compile(
- union(
- except_(
- select([table2.c.otherid]),
- select([table3.c.userid]),
- ),
- select([table1.c.myid])
- )
- ,
- "SELECT myothertable.otherid FROM myothertable EXCEPT SELECT thirdtable.userid FROM thirdtable \
-UNION SELECT mytable.myid FROM mytable"
+ "SELECT mytable.myid FROM mytable UNION ALL "
+ "(SELECT myothertable.otherid FROM myothertable UNION "
+ "SELECT thirdtable.userid FROM thirdtable)"
)
+
s = select([column('foo'), column('bar')])
- s = union(s, s)
- s = union(s, s)
- self.assert_compile(s, "SELECT foo, bar UNION SELECT foo, bar UNION (SELECT foo, bar UNION SELECT foo, bar)")
-
- s = select([column('foo'), column('bar')])
+
# ORDER BY's even though not supported by all DB's, are rendered if requested
self.assert_compile(union(s.order_by("foo"), s.order_by("bar")),
"SELECT foo, bar ORDER BY foo UNION SELECT foo, bar ORDER BY bar"
)
# self_group() is honored
- self.assert_compile(union(s.order_by("foo").self_group(), s.order_by("bar").limit(10).self_group()),
+ self.assert_compile(
+ union(s.order_by("foo").self_group(), s.order_by("bar").limit(10).self_group()),
"(SELECT foo, bar ORDER BY foo) UNION (SELECT foo, bar ORDER BY bar LIMIT 10)"
)
+ def test_compound_grouping(self):
+ s = select([column('foo'), column('bar')]).select_from('bat')
+
+ self.assert_compile(
+ union(union(union(s, s), s), s),
+ "((SELECT foo, bar FROM bat UNION SELECT foo, bar FROM bat) "
+ "UNION SELECT foo, bar FROM bat) UNION SELECT foo, bar FROM bat"
+ )
+
+ self.assert_compile(
+ union(s, s, s, s),
+ "SELECT foo, bar FROM bat UNION SELECT foo, bar "
+ "FROM bat UNION SELECT foo, bar FROM bat UNION SELECT foo, bar FROM bat"
+ )
+
+ self.assert_compile(
+ union(s, union(s, union(s, s))),
+ "SELECT foo, bar FROM bat UNION (SELECT foo, bar FROM bat "
+ "UNION (SELECT foo, bar FROM bat UNION SELECT foo, bar FROM bat))"
+ )
+
+ self.assert_compile(
+ select([s.alias()]),
+ 'SELECT anon_1.foo, anon_1.bar FROM (SELECT foo, bar FROM bat) AS anon_1'
+ )
+
+ self.assert_compile(
+ select([union(s, s).alias()]),
+ 'SELECT anon_1.foo, anon_1.bar FROM '
+ '(SELECT foo, bar FROM bat UNION SELECT foo, bar FROM bat) AS anon_1'
+ )
+
+ self.assert_compile(
+ select([except_(s, s).alias()]),
+ 'SELECT anon_1.foo, anon_1.bar FROM '
+ '(SELECT foo, bar FROM bat EXCEPT SELECT foo, bar FROM bat) AS anon_1'
+ )
+
+ # this query sqlite specifically chokes on
+ self.assert_compile(
+ union(
+ except_(s, s),
+ s
+ ),
+ "(SELECT foo, bar FROM bat EXCEPT SELECT foo, bar FROM bat) "
+ "UNION SELECT foo, bar FROM bat"
+ )
+
+ self.assert_compile(
+ union(
+ s,
+ except_(s, s),
+ ),
+ "SELECT foo, bar FROM bat "
+ "UNION (SELECT foo, bar FROM bat EXCEPT SELECT foo, bar FROM bat)"
+ )
+
+ # this solves it
+ self.assert_compile(
+ union(
+ except_(s, s).alias().select(),
+ s
+ ),
+ "SELECT anon_1.foo, anon_1.bar FROM "
+ "(SELECT foo, bar FROM bat EXCEPT SELECT foo, bar FROM bat) AS anon_1 "
+ "UNION SELECT foo, bar FROM bat"
+ )
+
+ self.assert_compile(
+ except_(
+ union(s, s),
+ union(s, s)
+ ),
+ "(SELECT foo, bar FROM bat UNION SELECT foo, bar FROM bat) "
+ "EXCEPT (SELECT foo, bar FROM bat UNION SELECT foo, bar FROM bat)"
+ )
+ s2 = union(s, s)
+ s3 = union(s2, s2)
+ self.assert_compile(s3, "(SELECT foo, bar FROM bat "
+ "UNION SELECT foo, bar FROM bat) "
+ "UNION (SELECT foo, bar FROM bat "
+ "UNION SELECT foo, bar FROM bat)")
+
+
+ self.assert_compile(
+ union(
+ intersect(s, s),
+ intersect(s, s)
+ ),
+ "(SELECT foo, bar FROM bat INTERSECT SELECT foo, bar FROM bat) "
+ "UNION (SELECT foo, bar FROM bat INTERSECT SELECT foo, bar FROM bat)"
+ )
@testing.uses_deprecated()
def test_binds(self):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
