summaryrefslogtreecommitdiff
path: root/test
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2014-09-01 20:19:54 -0400
committerMike Bayer <mike_mp@zzzcomputing.com>2014-09-01 20:19:54 -0400
commit7c6a45c480a865ac9580eb33fcca2dae5b19dd11 (patch)
tree870c078707cde0af769a940b1fc1a15ce7966691 /test
parent382f82538b5484b1c384c71fbf84438312cbe34f (diff)
downloadsqlalchemy-7c6a45c480a865ac9580eb33fcca2dae5b19dd11.tar.gz
- The :func:`~.expression.column` and :func:`~.expression.table`
constructs are now importable from the "from sqlalchemy" namespace, just like every other Core construct. - The implicit conversion of strings to :func:`.text` constructs when passed to most builder methods of :func:`.select` as well as :class:`.Query` now emits a warning with just the plain string sent. The textual conversion still proceeds normally, however. The only method that accepts a string without a warning are the "label reference" methods like order_by(), group_by(); these functions will now at compile time attempt to resolve a single string argument to a column or label expression present in the selectable; if none is located, the expression still renders, but you get the warning again. The rationale here is that the implicit conversion from string to text is more unexpected than not these days, and it is better that the user send more direction to the Core / ORM when passing a raw string as to what direction should be taken. Core/ORM tutorials have been updated to go more in depth as to how text is handled. fixes #2992
Diffstat (limited to 'test')
-rw-r--r--test/aaa_profiling/test_memusage.py2
-rw-r--r--test/dialect/mysql/test_compiler.py2
-rw-r--r--test/dialect/postgresql/test_types.py4
-rw-r--r--test/dialect/test_oracle.py2
-rw-r--r--test/orm/test_assorted_eager.py4
-rw-r--r--test/orm/test_deprecations.py9
-rw-r--r--test/orm/test_eager_relations.py6
-rw-r--r--test/orm/test_froms.py26
-rw-r--r--test/orm/test_query.py124
-rw-r--r--test/orm/test_selectable.py9
-rw-r--r--test/orm/test_update_delete.py6
-rw-r--r--test/sql/test_compiler.py33
-rw-r--r--test/sql/test_functions.py2
-rw-r--r--test/sql/test_generative.py44
-rw-r--r--test/sql/test_insert.py8
-rw-r--r--test/sql/test_query.py10
-rw-r--r--test/sql/test_text.py254
17 files changed, 404 insertions, 141 deletions
diff --git a/test/aaa_profiling/test_memusage.py b/test/aaa_profiling/test_memusage.py
index d4ed1fe2e..f4bce6b01 100644
--- a/test/aaa_profiling/test_memusage.py
+++ b/test/aaa_profiling/test_memusage.py
@@ -387,7 +387,7 @@ class MemUsageTest(EnsureZeroed):
def go():
util.warn_limited(
"memusage warning, param1: %s, param2: %s",
- next(counter), next(counter))
+ (next(counter), next(counter)))
go()
def test_mapper_reset(self):
diff --git a/test/dialect/mysql/test_compiler.py b/test/dialect/mysql/test_compiler.py
index e8171fad5..1f9a39055 100644
--- a/test/dialect/mysql/test_compiler.py
+++ b/test/dialect/mysql/test_compiler.py
@@ -193,7 +193,7 @@ class SQLTest(fixtures.TestBase, AssertsCompiledSQL):
kw['distinct'] = distinct
if prefixes is not None:
kw['prefixes'] = prefixes
- return str(select(['q'], **kw).compile(dialect=dialect))
+ return str(select([column('q')], **kw).compile(dialect=dialect))
eq_(gen(None), 'SELECT q')
eq_(gen(True), 'SELECT DISTINCT q')
diff --git a/test/dialect/postgresql/test_types.py b/test/dialect/postgresql/test_types.py
index c594c5974..5c5da59b1 100644
--- a/test/dialect/postgresql/test_types.py
+++ b/test/dialect/postgresql/test_types.py
@@ -7,7 +7,7 @@ from sqlalchemy import testing
import datetime
from sqlalchemy import Table, MetaData, Column, Integer, Enum, Float, select, \
func, DateTime, Numeric, exc, String, cast, REAL, TypeDecorator, Unicode, \
- Text, null
+ Text, null, text
from sqlalchemy.sql import operators
from sqlalchemy import types
from sqlalchemy.dialects.postgresql import base as postgresql
@@ -906,7 +906,7 @@ class TimestampTest(fixtures.TestBase, AssertsExecutionResults):
engine = testing.db
connection = engine.connect()
- s = select(["timestamp '2007-12-25'"])
+ s = select([text("timestamp '2007-12-25'")])
result = connection.execute(s).first()
eq_(result[0], datetime.datetime(2007, 12, 25, 0, 0))
diff --git a/test/dialect/test_oracle.py b/test/dialect/test_oracle.py
index 597a5dc40..187042036 100644
--- a/test/dialect/test_oracle.py
+++ b/test/dialect/test_oracle.py
@@ -397,7 +397,7 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
query = select([table1, table2], or_(table1.c.name == 'fred',
table1.c.myid == 10, table2.c.othername != 'jack',
- 'EXISTS (select yay from foo where boo = lar)'
+ text('EXISTS (select yay from foo where boo = lar)')
), from_obj=[outerjoin(table1, table2,
table1.c.myid == table2.c.otherid)])
self.assert_compile(query,
diff --git a/test/orm/test_assorted_eager.py b/test/orm/test_assorted_eager.py
index d00da77f5..2bee3cbd6 100644
--- a/test/orm/test_assorted_eager.py
+++ b/test/orm/test_assorted_eager.py
@@ -10,7 +10,7 @@ import datetime
import sqlalchemy as sa
from sqlalchemy import testing
-from sqlalchemy import Integer, String, ForeignKey
+from sqlalchemy import Integer, String, ForeignKey, table, text
from sqlalchemy.testing.schema import Table, Column
from sqlalchemy.orm import mapper, relationship, backref, create_session
from sqlalchemy.testing import eq_
@@ -215,7 +215,7 @@ class EagerTest(fixtures.MappedTest):
q = s.query(Thing).options(sa.orm.joinedload('category'))
l = (q.filter(
(tests.c.owner_id==1) &
- ('options.someoption is null or options.someoption=%s' % false)).
+ text('options.someoption is null or options.someoption=%s' % false)).
join('owner_option'))
result = ["%d %s" % ( t.id,t.category.name ) for t in l]
diff --git a/test/orm/test_deprecations.py b/test/orm/test_deprecations.py
index a8be34971..75ed5c726 100644
--- a/test/orm/test_deprecations.py
+++ b/test/orm/test_deprecations.py
@@ -5,11 +5,10 @@ modern (i.e. not deprecated) alternative to them. The tests snippets here can
be migrated directly to the wiki, docs, etc.
"""
-from sqlalchemy import testing
-from sqlalchemy import Integer, String, ForeignKey, func
+from sqlalchemy import Integer, String, ForeignKey, func, text
from sqlalchemy.testing.schema import Table
from sqlalchemy.testing.schema import Column
-from sqlalchemy.orm import mapper, relationship, relation, create_session, sessionmaker
+from sqlalchemy.orm import mapper, relationship, create_session, sessionmaker
from sqlalchemy.testing import fixtures
@@ -513,7 +512,7 @@ class QueryAlternativesTest(fixtures.MappedTest):
session = create_session()
users = (session.query(User).
- from_statement('SELECT * FROM users_table')).all()
+ from_statement(text('SELECT * FROM users_table'))).all()
assert len(users) == 4
def test_select_whereclause(self):
@@ -532,6 +531,6 @@ class QueryAlternativesTest(fixtures.MappedTest):
users = session.query(User).filter(User.name=='ed').all()
assert len(users) == 1 and users[0].name == 'ed'
- users = session.query(User).filter("name='ed'").all()
+ users = session.query(User).filter(text("name='ed'")).all()
assert len(users) == 1 and users[0].name == 'ed'
diff --git a/test/orm/test_eager_relations.py b/test/orm/test_eager_relations.py
index a8f44e146..b0c203bf1 100644
--- a/test/orm/test_eager_relations.py
+++ b/test/orm/test_eager_relations.py
@@ -7,7 +7,7 @@ from sqlalchemy.orm import joinedload, deferred, undefer, \
joinedload_all, backref, Session,\
defaultload, Load
from sqlalchemy import Integer, String, Date, ForeignKey, and_, select, \
- func
+ func, text
from sqlalchemy.testing.schema import Table, Column
from sqlalchemy.orm import mapper, relationship, create_session, \
lazyload, aliased, column_property
@@ -632,7 +632,7 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
# they should be required to locate only their aliased/fully table
# qualified column name.
noeagers = create_session().query(User).\
- from_statement("select * from users").all()
+ from_statement(text("select * from users")).all()
assert 'orders' not in noeagers[0].__dict__
assert 'addresses' not in noeagers[0].__dict__
@@ -1109,7 +1109,7 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
q = create_session().query(User)
- l = q.filter("users.id in (7, 8, 9)").order_by("users.id")
+ l = q.filter(text("users.id in (7, 8, 9)")).order_by(text("users.id"))
def go():
eq_(self.static.user_order_result[0:3], l.all())
diff --git a/test/orm/test_froms.py b/test/orm/test_froms.py
index 7259132fe..4246df1f6 100644
--- a/test/orm/test_froms.py
+++ b/test/orm/test_froms.py
@@ -3,7 +3,7 @@ from sqlalchemy.testing import (
fixtures, eq_, assert_raises, assert_raises_message, AssertsCompiledSQL)
from sqlalchemy import (
exc as sa_exc, util, Integer, Table, String, ForeignKey, select, func,
- and_, asc, desc, inspect, literal_column, cast, exists)
+ and_, asc, desc, inspect, literal_column, cast, exists, text)
from sqlalchemy.orm import (
configure_mappers, Session, mapper, create_session, relationship,
column_property, joinedload_all, contains_eager, contains_alias,
@@ -438,8 +438,8 @@ class ColumnAccessTest(QueryTest, AssertsCompiledSQL):
self.assert_compile(
q3.order_by(c1),
"SELECT anon_1.c1 AS anon_1_c1, anon_1.c2 "
- "AS anon_1_c2 FROM (SELECT c1 AS c1, c2 AS c2 WHERE "
- "c1 = :c1_1 UNION SELECT c1 AS c1, c2 AS c2 "
+ "AS anon_1_c2 FROM (SELECT c1, c2 WHERE "
+ "c1 = :c1_1 UNION SELECT c1, c2 "
"WHERE c1 = :c1_2) AS anon_1 ORDER BY anon_1.c1"
)
@@ -455,7 +455,7 @@ class ColumnAccessTest(QueryTest, AssertsCompiledSQL):
"SELECT anon_1.anon_2_c1 AS anon_1_anon_2_c1, anon_1.anon_2_c2 AS "
"anon_1_anon_2_c2 FROM (SELECT anon_2.c1 AS anon_2_c1, anon_2.c2 "
"AS anon_2_c2 "
- "FROM (SELECT c1 AS c1, c2 AS c2 WHERE c1 = :c1_1) AS "
+ "FROM (SELECT c1, c2 WHERE c1 = :c1_1) AS "
"anon_2) AS anon_1 ORDER BY anon_1.anon_2_c1"
)
@@ -470,8 +470,8 @@ class ColumnAccessTest(QueryTest, AssertsCompiledSQL):
self.assert_compile(
q3.order_by(c1),
"SELECT anon_1.c1 AS anon_1_c1, anon_1.c2 "
- "AS anon_1_c2 FROM (SELECT c1 AS c1, c2 AS c2 WHERE "
- "c1 = :c1_1 UNION SELECT c1 AS c1, c2 AS c2 "
+ "AS anon_1_c2 FROM (SELECT c1, c2 WHERE "
+ "c1 = :c1_1 UNION SELECT c1, c2 "
"WHERE c1 = :c1_2) AS anon_1 ORDER BY anon_1.c1"
)
@@ -643,7 +643,9 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
query = users.select(users.c.id == 7).\
union(users.select(users.c.id > 7)).alias('ulist').\
outerjoin(addresses).\
- select(use_labels=True, order_by=['ulist.id', addresses.c.id])
+ select(
+ use_labels=True,
+ order_by=[text('ulist.id'), addresses.c.id])
sess = create_session()
q = sess.query(User)
@@ -663,7 +665,9 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
query = users.select(users.c.id == 7).\
union(users.select(users.c.id > 7)).alias('ulist').\
outerjoin(addresses). \
- select(use_labels=True, order_by=['ulist.id', addresses.c.id])
+ select(
+ use_labels=True,
+ order_by=[text('ulist.id'), addresses.c.id])
sess = create_session()
q = sess.query(User)
@@ -682,7 +686,9 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
query = users.select(users.c.id == 7).\
union(users.select(users.c.id > 7)).alias('ulist').\
outerjoin(addresses). \
- select(use_labels=True, order_by=['ulist.id', addresses.c.id])
+ select(
+ use_labels=True,
+ order_by=[text('ulist.id'), addresses.c.id])
sess = create_session()
# better way. use select_entity_from()
@@ -706,7 +712,7 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
query = users.select(users.c.id == 7).\
union(users.select(users.c.id > 7)).\
alias('ulist').outerjoin(adalias).\
- select(use_labels=True, order_by=['ulist.id', adalias.c.id])
+ select(use_labels=True, order_by=[text('ulist.id'), adalias.c.id])
def go():
l = sess.query(User).select_entity_from(query).\
diff --git a/test/orm/test_query.py b/test/orm/test_query.py
index 8f83b07aa..c0e9f9e1c 100644
--- a/test/orm/test_query.py
+++ b/test/orm/test_query.py
@@ -2,7 +2,8 @@ from sqlalchemy import (
testing, null, exists, text, union, literal, literal_column, func, between,
Unicode, desc, and_, bindparam, select, distinct, or_, collate, insert,
Integer, String, Boolean, exc as sa_exc, util, cast)
-from sqlalchemy.sql import operators, column, expression
+from sqlalchemy.sql import operators, expression
+from sqlalchemy import column, table
from sqlalchemy.engine import default
from sqlalchemy.orm import (
attributes, mapper, relationship, create_session, synonym, Session,
@@ -12,7 +13,7 @@ from sqlalchemy.testing.assertsql import CompiledSQL
from sqlalchemy.testing.schema import Table, Column
import sqlalchemy as sa
from sqlalchemy.testing.assertions import (
- eq_, assert_raises, assert_raises_message)
+ eq_, assert_raises, assert_raises_message, expect_warnings)
from sqlalchemy.testing import fixtures, AssertsCompiledSQL
from test.orm import _fixtures
from sqlalchemy.orm.util import join, with_parent
@@ -746,11 +747,11 @@ class InvalidGenerationsTest(QueryTest, AssertsCompiledSQL):
q = meth(q, *arg, **kw)
assert_raises(
sa_exc.InvalidRequestError,
- q.from_statement, "x"
+ q.from_statement, text("x")
)
q = s.query(User)
- q = q.from_statement("x")
+ q = q.from_statement(text("x"))
assert_raises(
sa_exc.InvalidRequestError,
meth, q, *arg, **kw
@@ -1670,7 +1671,7 @@ class FilterTest(QueryTest, AssertsCompiledSQL):
User = self.classes.User
s = create_session()
self.assert_compile(
- s.query(User).filter("name='ed'"),
+ s.query(User).filter(text("name='ed'")),
"SELECT users.id AS users_id, users.name "
"AS users_name FROM users WHERE name='ed'"
)
@@ -1759,7 +1760,7 @@ class SetOpsTest(QueryTest, AssertsCompiledSQL):
)
for q in (
- q3.order_by(User.id, "anon_1_param_1"),
+ q3.order_by(User.id, text("anon_1_param_1")),
q6.order_by(User.id, "foo")):
eq_(
q.all(),
@@ -2139,7 +2140,7 @@ class YieldTest(_fixtures.FixtureTest):
sess = create_session()
q = iter(
sess.query(User).yield_per(1).from_statement(
- "select * from users"))
+ text("select * from users")))
ret = []
eq_(len(sess.identity_map), 0)
@@ -2279,52 +2280,57 @@ class TextTest(QueryTest):
def test_fulltext(self):
User = self.classes.User
- eq_(
- create_session().query(User).
- from_statement("select * from users order by id").all(),
- [User(id=7), User(id=8), User(id=9), User(id=10)]
- )
+ with expect_warnings("Textual SQL"):
+ eq_(
+ create_session().query(User).
+ from_statement("select * from users order by id").all(),
+ [User(id=7), User(id=8), User(id=9), User(id=10)]
+ )
eq_(
create_session().query(User).from_statement(
- "select * from users order by id").first(), User(id=7)
+ text("select * from users order by id")).first(), User(id=7)
)
eq_(
create_session().query(User).from_statement(
- "select * from users where name='nonexistent'").first(), None)
+ text("select * from users where name='nonexistent'")).first(),
+ None)
def test_fragment(self):
User = self.classes.User
- eq_(
- create_session().query(User).filter("id in (8, 9)").all(),
- [User(id=8), User(id=9)]
+ with expect_warnings("Textual SQL expression"):
+ eq_(
+ create_session().query(User).filter("id in (8, 9)").all(),
+ [User(id=8), User(id=9)]
- )
+ )
- eq_(
- create_session().query(User).filter("name='fred'").
- filter("id=9").all(), [User(id=9)]
- )
- eq_(
- create_session().query(User).filter("name='fred'").
- filter(User.id == 9).all(), [User(id=9)]
- )
+ eq_(
+ create_session().query(User).filter("name='fred'").
+ filter("id=9").all(), [User(id=9)]
+ )
+ eq_(
+ create_session().query(User).filter("name='fred'").
+ filter(User.id == 9).all(), [User(id=9)]
+ )
- def test_binds(self):
+ def test_binds_coerce(self):
User = self.classes.User
- eq_(
- create_session().query(User).filter("id in (:id1, :id2)").
- params(id1=8, id2=9).all(), [User(id=8), User(id=9)]
- )
+ with expect_warnings("Textual SQL expression"):
+ eq_(
+ create_session().query(User).filter("id in (:id1, :id2)").
+ params(id1=8, id2=9).all(), [User(id=8), User(id=9)]
+ )
def test_as_column(self):
User = self.classes.User
s = create_session()
- assert_raises(sa_exc.InvalidRequestError, s.query,
- User.id, text("users.name"))
+ assert_raises(
+ sa_exc.InvalidRequestError, s.query,
+ User.id, text("users.name"))
eq_(
s.query(User.id, "name").order_by(User.id).all(),
@@ -2335,7 +2341,8 @@ class TextTest(QueryTest):
s = create_session()
eq_(
s.query(User).from_statement(
- select(['id', 'name']).select_from('users').order_by('id'),
+ select([column('id'), column('name')]).
+ select_from(table('users')).order_by('id'),
).all(),
[User(id=7), User(id=8), User(id=9), User(id=10)]
)
@@ -2374,6 +2381,45 @@ class TextTest(QueryTest):
)
+class TextWarningTest(QueryTest, AssertsCompiledSQL):
+ def _test(self, fn, arg, offending_clause, expected):
+ assert_raises_message(
+ sa.exc.SAWarning,
+ r"Textual (?:SQL|column|SQL FROM) expression %(stmt)r should be "
+ r"explicitly declared (?:with|as) text\(%(stmt)r\)" % {
+ "stmt": util.ellipses_string(offending_clause),
+ },
+ fn, arg
+ )
+
+ with expect_warnings("Textual "):
+ stmt = fn(arg)
+ self.assert_compile(stmt, expected)
+
+ def test_filter(self):
+ User = self.classes.User
+ self._test(
+ Session().query(User.id).filter, "myid == 5", "myid == 5",
+ "SELECT users.id AS users_id FROM users WHERE myid == 5"
+ )
+
+ def test_having(self):
+ User = self.classes.User
+ self._test(
+ Session().query(User.id).having, "myid == 5", "myid == 5",
+ "SELECT users.id AS users_id FROM users HAVING myid == 5"
+ )
+
+ def test_from_statement(self):
+ User = self.classes.User
+ self._test(
+ Session().query(User.id).from_statement,
+ "select id from user",
+ "select id from user",
+ "select id from user",
+ )
+
+
class ParentTest(QueryTest, AssertsCompiledSQL):
__dialect__ = 'default'
@@ -2821,7 +2867,7 @@ class BooleanEvalTest(fixtures.TestBase, testing.AssertsCompiledSQL):
c = column('x', Boolean)
self.assert_compile(
s.query(c).filter(c),
- "SELECT x AS x WHERE x",
+ "SELECT x WHERE x",
dialect=self._dialect(True)
)
@@ -2830,7 +2876,7 @@ class BooleanEvalTest(fixtures.TestBase, testing.AssertsCompiledSQL):
c = column('x', Boolean)
self.assert_compile(
s.query(c).filter(c),
- "SELECT x AS x WHERE x = 1",
+ "SELECT x WHERE x = 1",
dialect=self._dialect(False)
)
@@ -2839,7 +2885,7 @@ class BooleanEvalTest(fixtures.TestBase, testing.AssertsCompiledSQL):
c = column('x', Boolean)
self.assert_compile(
s.query(c).filter(~c),
- "SELECT x AS x WHERE x = 0",
+ "SELECT x WHERE x = 0",
dialect=self._dialect(False)
)
@@ -2848,7 +2894,7 @@ class BooleanEvalTest(fixtures.TestBase, testing.AssertsCompiledSQL):
c = column('x', Boolean)
self.assert_compile(
s.query(c).filter(~c),
- "SELECT x AS x WHERE NOT x",
+ "SELECT x WHERE NOT x",
dialect=self._dialect(True)
)
@@ -2857,6 +2903,6 @@ class BooleanEvalTest(fixtures.TestBase, testing.AssertsCompiledSQL):
c = column('x', Boolean)
self.assert_compile(
s.query(c).having(c),
- "SELECT x AS x HAVING x = 1",
+ "SELECT x HAVING x = 1",
dialect=self._dialect(False)
)
diff --git a/test/orm/test_selectable.py b/test/orm/test_selectable.py
index 42347213e..3f2aaba45 100644
--- a/test/orm/test_selectable.py
+++ b/test/orm/test_selectable.py
@@ -2,7 +2,7 @@
from sqlalchemy.testing import assert_raises, assert_raises_message
import sqlalchemy as sa
from sqlalchemy import testing
-from sqlalchemy import String, Integer, select
+from sqlalchemy import String, Integer, select, column
from sqlalchemy.testing.schema import Table, Column
from sqlalchemy.orm import mapper, Session
from sqlalchemy.testing import eq_, AssertsCompiledSQL
@@ -29,12 +29,13 @@ class SelectableNoFromsTest(fixtures.MappedTest, AssertsCompiledSQL):
Subset = self.classes.Subset
- selectable = select(["x", "y", "z"]).alias()
+ selectable = select([column("x"), column("y"), column("z")]).alias()
mapper(Subset, selectable, primary_key=[selectable.c.x])
self.assert_compile(
Session().query(Subset),
- "SELECT anon_1.x, anon_1.y, anon_1.z FROM (SELECT x, y, z) AS anon_1",
+ "SELECT anon_1.x AS anon_1_x, anon_1.y AS anon_1_y, "
+ "anon_1.z AS anon_1_z FROM (SELECT x, y, z) AS anon_1",
use_default_dialect=True
)
@@ -42,7 +43,7 @@ class SelectableNoFromsTest(fixtures.MappedTest, AssertsCompiledSQL):
Subset = self.classes.Subset
- selectable = select(["x", "y", "z"]).alias()
+ selectable = select([column("x"), column("y"), column("z")]).alias()
assert_raises_message(
sa.exc.ArgumentError,
"could not assemble any primary key columns",
diff --git a/test/orm/test_update_delete.py b/test/orm/test_update_delete.py
index 4f23b1159..35d527ca8 100644
--- a/test/orm/test_update_delete.py
+++ b/test/orm/test_update_delete.py
@@ -1,7 +1,7 @@
from sqlalchemy.testing import eq_, assert_raises, assert_raises_message
from sqlalchemy.testing import fixtures
from sqlalchemy import Integer, String, ForeignKey, or_, and_, exc, \
- select, func, Boolean, case
+ select, func, Boolean, case, text
from sqlalchemy.orm import mapper, relationship, backref, Session, \
joinedload, aliased
from sqlalchemy import testing
@@ -105,7 +105,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
sess = Session()
john, jack, jill, jane = sess.query(User).order_by(User.id).all()
- sess.query(User).filter('name = :name').params(
+ sess.query(User).filter(text('name = :name')).params(
name='john').delete('fetch')
assert john not in sess
@@ -229,7 +229,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
john, jack, jill, jane = sess.query(User).order_by(User.id).all()
- sess.query(User).filter('age > :x').params(x=29).\
+ sess.query(User).filter(text('age > :x')).params(x=29).\
update({'age': User.age - 10}, synchronize_session='fetch')
eq_([john.age, jack.age, jill.age, jane.age], [25, 37, 29, 27])
diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index 2b2083bf7..4977611c5 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -239,7 +239,7 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
)
def test_select_precol_compile_ordering(self):
- s1 = select([column('x')]).select_from('a').limit(5).as_scalar()
+ s1 = select([column('x')]).select_from(text('a')).limit(5).as_scalar()
s2 = select([s1]).limit(10)
class MyCompiler(compiler.SQLCompiler):
@@ -346,7 +346,7 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
def test_select_from_clauselist(self):
self.assert_compile(
select([ClauseList(column('a'), column('b'))]
- ).select_from('sometable'),
+ ).select_from(text('sometable')),
'SELECT a, b FROM sometable'
)
@@ -462,7 +462,7 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
)
self.assert_compile(
- select(["a", "a", "a"]),
+ select([column("a"), column("a"), column("a")]),
"SELECT a, a, a"
)
@@ -933,7 +933,7 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
)
def test_conjunctions(self):
- a, b, c = 'a', 'b', 'c'
+ a, b, c = text('a'), text('b'), text('c')
x = and_(a, b, c)
assert isinstance(x.type, Boolean)
assert str(x) == 'a AND b AND c'
@@ -944,7 +944,7 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
self.assert_compile(
and_(table1.c.myid == 12, table1.c.name == 'asdf',
- table2.c.othername == 'foo', "sysdate() = today()"),
+ table2.c.othername == 'foo', text("sysdate() = today()")),
"mytable.myid = :myid_1 AND mytable.name = :name_1 "
"AND myothertable.othername = "
":othername_1 AND sysdate() = today()"
@@ -955,7 +955,7 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
table1.c.myid == 12,
or_(table2.c.othername == 'asdf',
table2.c.othername == 'foo', table2.c.otherid == 9),
- "sysdate() = today()",
+ text("sysdate() = today()"),
),
'mytable.myid = :myid_1 AND (myothertable.othername = '
':othername_1 OR myothertable.othername = :othername_2 OR '
@@ -1067,8 +1067,12 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
def test_multiple_col_binds(self):
self.assert_compile(
- select(["*"], or_(table1.c.myid == 12, table1.c.myid == 'asdf',
- table1.c.myid == 'foo')),
+ select(
+ [literal_column("*")],
+ or_(
+ table1.c.myid == 12, table1.c.myid == 'asdf',
+ table1.c.myid == 'foo')
+ ),
"SELECT * FROM mytable WHERE mytable.myid = :myid_1 "
"OR mytable.myid = :myid_2 OR mytable.myid = :myid_3"
)
@@ -1478,7 +1482,7 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
table1.c.name == 'fred',
table1.c.myid == 10,
table2.c.othername != 'jack',
- "EXISTS (select yay from foo where boo = lar)"
+ text("EXISTS (select yay from foo where boo = lar)")
),
from_obj=[outerjoin(table1, table2,
table1.c.myid == table2.c.otherid)]
@@ -1551,7 +1555,8 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
"SELECT mytable.myid, mytable.name "
"FROM mytable UNION SELECT myothertable.otherid, "
"myothertable.othername "
- "FROM myothertable ORDER BY myid LIMIT :param_1 OFFSET :param_2",
+ "FROM myothertable ORDER BY myid " # note table name is omitted
+ "LIMIT :param_1 OFFSET :param_2",
{'param_1': 5, 'param_2': 10}
)
@@ -1614,7 +1619,7 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
)
def test_compound_grouping(self):
- s = select([column('foo'), column('bar')]).select_from('bat')
+ s = select([column('foo'), column('bar')]).select_from(text('bat'))
self.assert_compile(
union(union(union(s, s), s), s),
@@ -2130,10 +2135,10 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
self.assert_compile(
select([
func.max(table1.c.name).over(
- partition_by=['foo']
+ partition_by=['description']
)
]),
- "SELECT max(mytable.name) OVER (PARTITION BY foo) "
+ "SELECT max(mytable.name) OVER (PARTITION BY mytable.description) "
"AS anon_1 FROM mytable"
)
# from partition_by
@@ -2396,7 +2401,7 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
def test_literal_as_text_fromstring(self):
self.assert_compile(
- and_("a", "b"),
+ and_(text("a"), text("b")),
"a AND b"
)
diff --git a/test/sql/test_functions.py b/test/sql/test_functions.py
index d3b718645..9b7649e63 100644
--- a/test/sql/test_functions.py
+++ b/test/sql/test_functions.py
@@ -340,7 +340,7 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
def test_select_method_three(self):
expr = func.rows("foo")
self.assert_compile(
- select(['foo']).select_from(expr),
+ select([column('foo')]).select_from(expr),
"SELECT foo FROM rows(:rows_1)"
)
diff --git a/test/sql/test_generative.py b/test/sql/test_generative.py
index 51a8a77cc..2e3c4b1e8 100644
--- a/test/sql/test_generative.py
+++ b/test/sql/test_generative.py
@@ -2,7 +2,7 @@ from sqlalchemy.sql import table, column, ClauseElement, operators
from sqlalchemy.sql.expression import _clone, _from_objects
from sqlalchemy import func, select, Integer, Table, \
Column, MetaData, extract, String, bindparam, tuple_, and_, union, text,\
- case, ForeignKey
+ case, ForeignKey, literal_column
from sqlalchemy.testing import fixtures, AssertsExecutionResults, \
AssertsCompiledSQL
from sqlalchemy import testing
@@ -620,7 +620,7 @@ class ClauseTest(fixtures.TestBase, AssertsCompiledSQL):
assert orig == str(s) == str(s5)
def test_correlated_select(self):
- s = select(['*'], t1.c.col1 == t2.c.col1,
+ s = select([literal_column('*')], t1.c.col1 == t2.c.col1,
from_obj=[t1, t2]).correlate(t2)
class Vis(CloningVisitor):
@@ -718,11 +718,11 @@ class ClauseAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
t2alias = t2.alias('t2alias')
vis = sql_util.ClauseAdapter(t1alias)
- s = select(['*'], from_obj=[t1alias, t2alias]).as_scalar()
+ s = select([literal_column('*')], from_obj=[t1alias, t2alias]).as_scalar()
assert t2alias in s._froms
assert t1alias in s._froms
- self.assert_compile(select(['*'], t2alias.c.col1 == s),
+ self.assert_compile(select([literal_column('*')], t2alias.c.col1 == s),
'SELECT * FROM table2 AS t2alias WHERE '
't2alias.col1 = (SELECT * FROM table1 AS '
't1alias)')
@@ -736,28 +736,28 @@ class ClauseAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
# correlate list on "s" needs to take into account the full
# _cloned_set for each element in _froms when correlating
- self.assert_compile(select(['*'], t2alias.c.col1 == s),
+ self.assert_compile(select([literal_column('*')], t2alias.c.col1 == s),
'SELECT * FROM table2 AS t2alias WHERE '
't2alias.col1 = (SELECT * FROM table1 AS '
't1alias)')
- s = select(['*'], from_obj=[t1alias,
+ s = select([literal_column('*')], from_obj=[t1alias,
t2alias]).correlate(t2alias).as_scalar()
- self.assert_compile(select(['*'], t2alias.c.col1 == s),
+ self.assert_compile(select([literal_column('*')], t2alias.c.col1 == s),
'SELECT * FROM table2 AS t2alias WHERE '
't2alias.col1 = (SELECT * FROM table1 AS '
't1alias)')
s = vis.traverse(s)
- self.assert_compile(select(['*'], t2alias.c.col1 == s),
+ self.assert_compile(select([literal_column('*')], t2alias.c.col1 == s),
'SELECT * FROM table2 AS t2alias WHERE '
't2alias.col1 = (SELECT * FROM table1 AS '
't1alias)')
s = CloningVisitor().traverse(s)
- self.assert_compile(select(['*'], t2alias.c.col1 == s),
+ self.assert_compile(select([literal_column('*')], t2alias.c.col1 == s),
'SELECT * FROM table2 AS t2alias WHERE '
't2alias.col1 = (SELECT * FROM table1 AS '
't1alias)')
- s = select(['*']).where(t1.c.col1 == t2.c.col1).as_scalar()
+ s = select([literal_column('*')]).where(t1.c.col1 == t2.c.col1).as_scalar()
self.assert_compile(select([t1.c.col1, s]),
'SELECT table1.col1, (SELECT * FROM table2 '
'WHERE table1.col1 = table2.col1) AS '
@@ -773,7 +773,7 @@ class ClauseAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
'SELECT t1alias.col1, (SELECT * FROM '
'table2 WHERE t1alias.col1 = table2.col1) '
'AS anon_1 FROM table1 AS t1alias')
- s = select(['*']).where(t1.c.col1
+ s = select([literal_column('*')]).where(t1.c.col1
== t2.c.col1).correlate(t1).as_scalar()
self.assert_compile(select([t1.c.col1, s]),
'SELECT table1.col1, (SELECT * FROM table2 '
@@ -823,20 +823,20 @@ class ClauseAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
def test_table_to_alias_2(self):
t1alias = t1.alias('t1alias')
vis = sql_util.ClauseAdapter(t1alias)
- self.assert_compile(vis.traverse(select(['*'], from_obj=[t1])),
+ self.assert_compile(vis.traverse(select([literal_column('*')], from_obj=[t1])),
'SELECT * FROM table1 AS t1alias')
def test_table_to_alias_3(self):
t1alias = t1.alias('t1alias')
vis = sql_util.ClauseAdapter(t1alias)
- self.assert_compile(select(['*'], t1.c.col1 == t2.c.col2),
+ self.assert_compile(select([literal_column('*')], t1.c.col1 == t2.c.col2),
'SELECT * FROM table1, table2 WHERE '
'table1.col1 = table2.col2')
def test_table_to_alias_4(self):
t1alias = t1.alias('t1alias')
vis = sql_util.ClauseAdapter(t1alias)
- self.assert_compile(vis.traverse(select(['*'], t1.c.col1
+ self.assert_compile(vis.traverse(select([literal_column('*')], t1.c.col1
== t2.c.col2)),
'SELECT * FROM table1 AS t1alias, table2 '
'WHERE t1alias.col1 = table2.col2')
@@ -847,7 +847,7 @@ class ClauseAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
self.assert_compile(
vis.traverse(
select(
- ['*'],
+ [literal_column('*')],
t1.c.col1 == t2.c.col2,
from_obj=[
t1,
@@ -861,7 +861,7 @@ class ClauseAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
self.assert_compile(
select([t1alias, t2]).where(
t1alias.c.col1 == vis.traverse(
- select(['*'], t1.c.col1 == t2.c.col2, from_obj=[t1, t2]).
+ select([literal_column('*')], t1.c.col1 == t2.c.col2, from_obj=[t1, t2]).
correlate(t1)
)
),
@@ -877,7 +877,7 @@ class ClauseAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
self.assert_compile(
select([t1alias, t2]).
where(t1alias.c.col1 == vis.traverse(
- select(['*'], t1.c.col1 == t2.c.col2, from_obj=[t1, t2]).
+ select([literal_column('*')], t1.c.col1 == t2.c.col2, from_obj=[t1, t2]).
correlate(t2))),
"SELECT t1alias.col1, t1alias.col2, t1alias.col3, "
"table2.col1, table2.col2, table2.col3 "
@@ -909,13 +909,13 @@ class ClauseAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
't1alias.col2 ELSE t1alias.col1 END')
def test_table_to_alias_10(self):
- s = select(['*'], from_obj=[t1]).alias('foo')
+ s = select([literal_column('*')], from_obj=[t1]).alias('foo')
self.assert_compile(s.select(),
'SELECT foo.* FROM (SELECT * FROM table1) '
'AS foo')
def test_table_to_alias_11(self):
- s = select(['*'], from_obj=[t1]).alias('foo')
+ s = select([literal_column('*')], from_obj=[t1]).alias('foo')
t1alias = t1.alias('t1alias')
vis = sql_util.ClauseAdapter(t1alias)
self.assert_compile(vis.traverse(s.select()),
@@ -923,7 +923,7 @@ class ClauseAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
'AS t1alias) AS foo')
def test_table_to_alias_12(self):
- s = select(['*'], from_obj=[t1]).alias('foo')
+ s = select([literal_column('*')], from_obj=[t1]).alias('foo')
self.assert_compile(s.select(),
'SELECT foo.* FROM (SELECT * FROM table1) '
'AS foo')
@@ -947,7 +947,7 @@ class ClauseAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
vis = sql_util.ClauseAdapter(t1alias)
t2alias = t2.alias('t2alias')
vis.chain(sql_util.ClauseAdapter(t2alias))
- self.assert_compile(vis.traverse(select(['*'], t1.c.col1
+ self.assert_compile(vis.traverse(select([literal_column('*')], t1.c.col1
== t2.c.col2)),
'SELECT * FROM table1 AS t1alias, table2 '
'AS t2alias WHERE t1alias.col1 = '
@@ -1317,7 +1317,7 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
self.assert_compile(s,
'SELECT table1.col1, table1.col2, '
'table1.col3 FROM table1')
- select_copy = s.column('yyy')
+ select_copy = s.column(column('yyy'))
self.assert_compile(select_copy,
'SELECT table1.col1, table1.col2, '
'table1.col3, yyy FROM table1')
diff --git a/test/sql/test_insert.py b/test/sql/test_insert.py
index d2fba5862..232c5758b 100644
--- a/test/sql/test_insert.py
+++ b/test/sql/test_insert.py
@@ -1,7 +1,7 @@
#! coding:utf-8
from sqlalchemy import Column, Integer, MetaData, String, Table,\
- bindparam, exc, func, insert, select
+ bindparam, exc, func, insert, select, column
from sqlalchemy.dialects import mysql, postgresql
from sqlalchemy.engine import default
from sqlalchemy.testing import AssertsCompiledSQL,\
@@ -238,8 +238,8 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
def test_insert_from_select_union(self):
mytable = self.tables.mytable
- name = 'name'
- description = 'desc'
+ name = column('name')
+ description = column('desc')
sel = select(
[name, mytable.c.description],
).union(
@@ -252,7 +252,7 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
ins,
"INSERT INTO mytable (name, description) "
"SELECT name, mytable.description FROM mytable "
- "UNION SELECT name, desc"
+ 'UNION SELECT name, "desc"'
)
def test_insert_from_select_col_values(self):
diff --git a/test/sql/test_query.py b/test/sql/test_query.py
index 2075bcecf..430c3fe7c 100644
--- a/test/sql/test_query.py
+++ b/test/sql/test_query.py
@@ -6,7 +6,7 @@ from sqlalchemy import (
exc, sql, func, select, String, Integer, MetaData, and_, ForeignKey,
union, intersect, except_, union_all, VARCHAR, INT, CHAR, text, Sequence,
bindparam, literal, not_, type_coerce, literal_column, desc, asc,
- TypeDecorator, or_, cast)
+ TypeDecorator, or_, cast, table, column)
from sqlalchemy.engine import default, result as _result
from sqlalchemy.testing.schema import Table, Column
@@ -864,8 +864,10 @@ class QueryTest(fixtures.TestBase):
# this will create column() objects inside
# the select(), these need to match on name anyway
r = testing.db.execute(
- select(['user_id', 'user_name']).select_from('query_users').
- where('user_id=2')
+ select([
+ column('user_id'), column('user_name')
+ ]).select_from(table('query_users')).
+ where(text('user_id=2'))
).first()
self.assert_(r.user_id == r['user_id'] == r[users.c.user_id] == 2)
self.assert_(
@@ -1764,7 +1766,7 @@ class KeyTargetingTest(fixtures.TablesTest):
# columns which the statement is against to be lightweight
# cols, which results in a more liberal comparison scheme
a, b = sql.column('a'), sql.column('b')
- stmt = select([a, b]).select_from("keyed2")
+ stmt = select([a, b]).select_from(table("keyed2"))
row = testing.db.execute(stmt).first()
assert keyed2.c.a in row
diff --git a/test/sql/test_text.py b/test/sql/test_text.py
index 6e9be9eea..182c63624 100644
--- a/test/sql/test_text.py
+++ b/test/sql/test_text.py
@@ -1,10 +1,13 @@
"""Test the TextClause and related constructs."""
-from sqlalchemy.testing import fixtures, AssertsCompiledSQL, eq_, assert_raises_message
+from sqlalchemy.testing import fixtures, AssertsCompiledSQL, eq_, \
+ assert_raises_message, expect_warnings
from sqlalchemy import text, select, Integer, String, Float, \
- bindparam, and_, func, literal_column, exc, MetaData, Table, Column
+ bindparam, and_, func, literal_column, exc, MetaData, Table, Column,\
+ asc, func, desc, union
from sqlalchemy.types import NullType
from sqlalchemy.sql import table, column
+from sqlalchemy import util
table1 = table('mytable',
column('myid', Integer),
@@ -38,9 +41,14 @@ class SelectCompositionTest(fixtures.TestBase, AssertsCompiledSQL):
def test_select_composition_one(self):
self.assert_compile(select(
- ["foobar(a)", "pk_foo_bar(syslaal)"],
- "a = 12",
- from_obj=["foobar left outer join lala on foobar.foo = lala.foo"]
+ [
+ literal_column("foobar(a)"),
+ literal_column("pk_foo_bar(syslaal)")
+ ],
+ text("a = 12"),
+ from_obj=[
+ text("foobar left outer join lala on foobar.foo = lala.foo")
+ ]
),
"SELECT foobar(a), pk_foo_bar(syslaal) FROM foobar "
"left outer join lala on foobar.foo = lala.foo WHERE a = 12"
@@ -48,18 +56,18 @@ class SelectCompositionTest(fixtures.TestBase, AssertsCompiledSQL):
def test_select_composition_two(self):
s = select()
- s.append_column("column1")
- s.append_column("column2")
- s.append_whereclause("column1=12")
- s.append_whereclause("column2=19")
+ s.append_column(column("column1"))
+ s.append_column(column("column2"))
+ s.append_whereclause(text("column1=12"))
+ s.append_whereclause(text("column2=19"))
s = s.order_by("column1")
- s.append_from("table1")
+ s.append_from(text("table1"))
self.assert_compile(s, "SELECT column1, column2 FROM table1 WHERE "
"column1=12 AND column2=19 ORDER BY column1")
def test_select_composition_three(self):
self.assert_compile(
- select(["column1", "column2"],
+ select([column("column1"), column("column2")],
from_obj=table1).alias('somealias').select(),
"SELECT somealias.column1, somealias.column2 FROM "
"(SELECT column1, column2 FROM mytable) AS somealias"
@@ -68,9 +76,13 @@ class SelectCompositionTest(fixtures.TestBase, AssertsCompiledSQL):
def test_select_composition_four(self):
# test that use_labels doesn't interfere with literal columns
self.assert_compile(
- select(["column1", "column2", table1.c.myid], from_obj=table1,
- use_labels=True),
- "SELECT column1, column2, mytable.myid AS mytable_myid "
+ select([
+ text("column1"), column("column2"),
+ column("column3").label("bar"), table1.c.myid],
+ from_obj=table1,
+ use_labels=True),
+ "SELECT column1, column2, column3 AS bar, "
+ "mytable.myid AS mytable_myid "
"FROM mytable"
)
@@ -78,8 +90,10 @@ class SelectCompositionTest(fixtures.TestBase, AssertsCompiledSQL):
# test that use_labels doesn't interfere
# with literal columns that have textual labels
self.assert_compile(
- select(["column1 AS foobar", "column2 AS hoho", table1.c.myid],
- from_obj=table1, use_labels=True),
+ select([
+ text("column1 AS foobar"), text("column2 AS hoho"),
+ table1.c.myid],
+ from_obj=table1, use_labels=True),
"SELECT column1 AS foobar, column2 AS hoho, "
"mytable.myid AS mytable_myid FROM mytable"
)
@@ -89,8 +103,10 @@ class SelectCompositionTest(fixtures.TestBase, AssertsCompiledSQL):
# doesn't interfere with literal columns,
# exported columns don't get quoted
self.assert_compile(
- select(["column1 AS foobar", "column2 AS hoho", table1.c.myid],
- from_obj=[table1]).select(),
+ select([
+ literal_column("column1 AS foobar"),
+ literal_column("column2 AS hoho"), table1.c.myid],
+ from_obj=[table1]).select(),
"SELECT column1 AS foobar, column2 AS hoho, myid FROM "
"(SELECT column1 AS foobar, column2 AS hoho, "
"mytable.myid AS myid FROM mytable)"
@@ -98,25 +114,29 @@ class SelectCompositionTest(fixtures.TestBase, AssertsCompiledSQL):
def test_select_composition_seven(self):
self.assert_compile(
- select(['col1', 'col2'], from_obj='tablename').alias('myalias'),
+ select([
+ literal_column('col1'),
+ literal_column('col2')
+ ], from_obj=table('tablename')).alias('myalias'),
"SELECT col1, col2 FROM tablename"
)
def test_select_composition_eight(self):
self.assert_compile(select(
- [table1.alias('t'), "foo.f"],
- "foo.f = t.id",
- from_obj=["(select f from bar where lala=heyhey) foo"]
+ [table1.alias('t'), text("foo.f")],
+ text("foo.f = t.id"),
+ from_obj=[text("(select f from bar where lala=heyhey) foo")]
),
"SELECT t.myid, t.name, t.description, foo.f FROM mytable AS t, "
"(select f from bar where lala=heyhey) foo WHERE foo.f = t.id")
def test_select_bundle_columns(self):
self.assert_compile(select(
- [table1, table2.c.otherid, "sysdate()", "foo, bar, lala"],
+ [table1, table2.c.otherid,
+ text("sysdate()"), text("foo, bar, lala")],
and_(
- "foo.id = foofoo(lala)",
- "datetime(foo) = Today",
+ text("foo.id = foofoo(lala)"),
+ text("datetime(foo) = Today"),
table1.c.myid == table2.c.otherid,
)
),
@@ -470,3 +490,187 @@ class AsFromTest(fixtures.TestBase, AssertsCompiledSQL):
set(t.element._bindparams),
set(["bat", "foo", "bar"])
)
+
+
+class TextWarningsTest(fixtures.TestBase, AssertsCompiledSQL):
+ __dialect__ = 'default'
+
+ def _test(self, fn, arg, offending_clause, expected):
+ assert_raises_message(
+ exc.SAWarning,
+ r"Textual (?:SQL|column|SQL FROM) expression %(stmt)r should be "
+ r"explicitly declared (?:with|as) text\(%(stmt)r\)" % {
+ "stmt": util.ellipses_string(offending_clause),
+ },
+ fn, arg
+ )
+
+ with expect_warnings("Textual "):
+ stmt = fn(arg)
+ self.assert_compile(stmt, expected)
+
+ def test_where(self):
+ self._test(
+ select([table1.c.myid]).where, "myid == 5", "myid == 5",
+ "SELECT mytable.myid FROM mytable WHERE myid == 5"
+ )
+
+ def test_column(self):
+ self._test(
+ select, ["myid"], "myid",
+ "SELECT myid"
+ )
+
+ def test_having(self):
+ self._test(
+ select([table1.c.myid]).having, "myid == 5", "myid == 5",
+ "SELECT mytable.myid FROM mytable HAVING myid == 5"
+ )
+
+ def test_from(self):
+ self._test(
+ select([table1.c.myid]).select_from, "mytable", "mytable",
+ "SELECT mytable.myid FROM mytable, mytable" # two FROMs
+ )
+
+
+class OrderByLabelResolutionTest(fixtures.TestBase, AssertsCompiledSQL):
+ __dialect__ = 'default'
+
+ def _test_warning(self, stmt, offending_clause, expected):
+ with expect_warnings(
+ "Can't resolve label reference %r;" % offending_clause):
+ self.assert_compile(
+ stmt,
+ expected
+ )
+ assert_raises_message(
+ exc.SAWarning,
+ "Can't resolve label reference %r; converting to text" %
+ offending_clause,
+ stmt.compile
+ )
+
+ def test_order_by_label(self):
+ stmt = select([table1.c.myid.label('foo')]).order_by('foo')
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid AS foo FROM mytable ORDER BY foo"
+ )
+
+ def test_order_by_colname(self):
+ stmt = select([table1.c.myid]).order_by('name')
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid FROM mytable ORDER BY mytable.name"
+ )
+
+ def test_order_by_alias_colname(self):
+ t1 = table1.alias()
+ stmt = select([t1.c.myid]).apply_labels().order_by('name')
+ self.assert_compile(
+ stmt,
+ "SELECT mytable_1.myid AS mytable_1_myid "
+ "FROM mytable AS mytable_1 ORDER BY mytable_1.name"
+ )
+
+ def test_unresolvable_warning_order_by(self):
+ stmt = select([table1.c.myid]).order_by('foobar')
+ self._test_warning(
+ stmt, "foobar",
+ "SELECT mytable.myid FROM mytable ORDER BY foobar"
+ )
+
+ def test_group_by_label(self):
+ stmt = select([table1.c.myid.label('foo')]).group_by('foo')
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid AS foo FROM mytable GROUP BY foo"
+ )
+
+ def test_group_by_colname(self):
+ stmt = select([table1.c.myid]).group_by('name')
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid FROM mytable GROUP BY mytable.name"
+ )
+
+ def test_unresolvable_warning_group_by(self):
+ stmt = select([table1.c.myid]).group_by('foobar')
+ self._test_warning(
+ stmt, "foobar",
+ "SELECT mytable.myid FROM mytable GROUP BY foobar"
+ )
+
+ def test_asc(self):
+ stmt = select([table1.c.myid]).order_by(asc('name'), 'description')
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid FROM mytable "
+ "ORDER BY mytable.name ASC, mytable.description"
+ )
+
+ def test_group_by_subquery(self):
+ stmt = select([table1]).alias()
+ stmt = select([stmt]).apply_labels().group_by("myid")
+ self.assert_compile(
+ stmt,
+ "SELECT anon_1.myid AS anon_1_myid, anon_1.name AS anon_1_name, "
+ "anon_1.description AS anon_1_description FROM "
+ "(SELECT mytable.myid AS myid, mytable.name AS name, "
+ "mytable.description AS description FROM mytable) AS anon_1 "
+ "GROUP BY anon_1.myid"
+ )
+
+ def test_order_by_func_label_desc(self):
+ stmt = select([func.foo('bar').label('fb'), table1]).\
+ order_by(desc('fb'))
+
+ self.assert_compile(
+ stmt,
+ "SELECT foo(:foo_1) AS fb, mytable.myid, mytable.name, "
+ "mytable.description FROM mytable ORDER BY fb DESC"
+ )
+
+ def test_pg_distinct(self):
+ stmt = select([table1]).distinct('name')
+ self.assert_compile(
+ stmt,
+ "SELECT DISTINCT ON (mytable.name) mytable.myid, "
+ "mytable.name, mytable.description FROM mytable",
+ dialect="postgresql"
+ )
+
+ def test_over(self):
+ stmt = select([column("foo"), column("bar")])
+ stmt = select(
+ [func.row_number().
+ over(order_by='foo', partition_by='bar')]
+ ).select_from(stmt)
+
+ self.assert_compile(
+ stmt,
+ "SELECT row_number() OVER (PARTITION BY bar ORDER BY foo) "
+ "AS anon_1 FROM (SELECT foo, bar)"
+ )
+
+ def test_union_column(self):
+ s1 = select([table1])
+ s2 = select([table1])
+ stmt = union(s1, s2).order_by("name")
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid, mytable.name, mytable.description FROM "
+ "mytable UNION SELECT mytable.myid, mytable.name, "
+ "mytable.description FROM mytable ORDER BY name"
+ )
+
+ def test_union_label(self):
+ s1 = select([func.foo("hoho").label('x')])
+ s2 = select([func.foo("Bar").label('y')])
+ stmt = union(s1, s2).order_by("x")
+ self.assert_compile(
+ stmt,
+ "SELECT foo(:foo_1) AS x UNION SELECT foo(:foo_2) AS y ORDER BY x"
+ )
+