#! coding:utf-8

"""

compiler tests.

These tests are among the very first that were written when SQLAlchemy

began in 2005. As a result the testing style here is very dense;

it's an ongoing job to break these into much smaller tests with correct pep8

styling and coherent test organization.

"""

import datetime

import decimal

from sqlalchemy import alias

from sqlalchemy import and_

from sqlalchemy import asc

from sqlalchemy import bindparam

from sqlalchemy import Boolean

from sqlalchemy import case

from sqlalchemy import cast

from sqlalchemy import CheckConstraint

from sqlalchemy import Column

from sqlalchemy import Date

from sqlalchemy import desc

from sqlalchemy import distinct

from sqlalchemy import exc

from sqlalchemy import except_

from sqlalchemy import exists

from sqlalchemy import Float

from sqlalchemy import ForeignKey

from sqlalchemy import func

from sqlalchemy import Integer

from sqlalchemy import intersect

from sqlalchemy import join

from sqlalchemy import literal

from sqlalchemy import literal_column

from sqlalchemy import MetaData

from sqlalchemy import not_

from sqlalchemy import null

from sqlalchemy import Numeric

from sqlalchemy import or_

from sqlalchemy import outerjoin

from sqlalchemy import over

from sqlalchemy import schema

from sqlalchemy import select

from sqlalchemy import Sequence

from sqlalchemy import sql

from sqlalchemy import String

from sqlalchemy import Table

from sqlalchemy import testing

from sqlalchemy import Text

from sqlalchemy import text

from sqlalchemy import TIMESTAMP

from sqlalchemy import true

from sqlalchemy import tuple_

from sqlalchemy import type_coerce

from sqlalchemy import types

from sqlalchemy import union

from sqlalchemy import union_all

from sqlalchemy import util

from sqlalchemy.dialects import mysql

from sqlalchemy.dialects import oracle

from sqlalchemy.dialects import postgresql

from sqlalchemy.dialects import sqlite

from sqlalchemy.dialects import sybase

from sqlalchemy.dialects.postgresql.base import PGCompiler

from sqlalchemy.dialects.postgresql.base import PGDialect

from sqlalchemy.engine import default

from sqlalchemy.ext.compiler import compiles

from sqlalchemy.sql import column

from sqlalchemy.sql import compiler

from sqlalchemy.sql import label

from sqlalchemy.sql import operators

from sqlalchemy.sql import table

from sqlalchemy.sql.elements import BooleanClauseList

from sqlalchemy.sql.expression import ClauseList

from sqlalchemy.sql.expression import HasPrefixes

from sqlalchemy.testing import assert_raises

from sqlalchemy.testing import assert_raises_message

from sqlalchemy.testing import AssertsCompiledSQL

from sqlalchemy.testing import eq_

from sqlalchemy.testing import eq_ignore_whitespace

from sqlalchemy.testing import fixtures

from sqlalchemy.testing import is_

from sqlalchemy.testing import mock

from sqlalchemy.util import u

table1 = table(

"mytable",

column("myid", Integer),

column("name", String),

column("description", String),

)

table2 = table(

"myothertable", column("otherid", Integer), column("othername", String)

)

table3 = table(

"thirdtable", column("userid", Integer), column("otherstuff", String)

)

metadata = MetaData()

# table with a schema

table4 = Table(

"remotetable",

metadata,

Column("rem_id", Integer, primary_key=True),

Column("datatype_id", Integer),

Column("value", String(20)),

schema="remote_owner",

)

# table with a 'multipart' schema

table5 = Table(

"remotetable",

metadata,

Column("rem_id", Integer, primary_key=True),

Column("datatype_id", Integer),

Column("value", String(20)),

schema="dbo.remote_owner",

)

parent = Table("parent", metadata, Column("id", Integer, primary_key=True))

child = Table(

"child",

metadata,

Column("id", Integer, primary_key=True),

Column("parent_id", ForeignKey("parent.id")),

)

users = table(

"users", column("user_id"), column("user_name"), column("password")

)

addresses = table(

"addresses",

column("address_id"),

column("user_id"),

column("street"),

column("city"),

column("state"),

column("zip"),

)

keyed = Table(

"keyed",

metadata,

Column("x", Integer, key="colx"),

Column("y", Integer, key="coly"),

Column("z", Integer),

)

class SelectTest(fixtures.TestBase, AssertsCompiledSQL):

__dialect__ = "default"

def test_attribute_sanity(self):

assert hasattr(table1, "c")

assert hasattr(table1.select().subquery(), "c")

assert not hasattr(table1.c.myid.self_group(), "columns")

assert not hasattr(table1.c.myid, "columns")

assert not hasattr(table1.c.myid, "c")

assert not hasattr(table1.select().subquery().c.myid, "c")

assert not hasattr(table1.select().subquery().c.myid, "columns")

assert not hasattr(table1.alias().c.myid, "columns")

assert not hasattr(table1.alias().c.myid, "c")

with testing.expect_deprecated(

"The SelectBase.c and SelectBase.columns attributes are "

"deprecated"

):

assert hasattr(table1.select(), "c")

assert_raises_message(

exc.InvalidRequestError,

"Scalar Select expression has no "

"columns; use this object directly within a "

"column-level expression.",

getattr,

select([table1.c.myid]).scalar_subquery().self_group(),

"columns",

)

assert_raises_message(

exc.InvalidRequestError,

"Scalar Select expression has no "

"columns; use this object directly within a "

"column-level expression.",

getattr,

select([table1.c.myid]).scalar_subquery(),

"columns",

)

def test_prefix_constructor(self):

class Pref(HasPrefixes):

def _generate(self):

return self

assert_raises(

exc.ArgumentError,

Pref().prefix_with,

"some prefix",

not_a_dialect=True,

)

def test_table_select(self):

self.assert_compile(

table1.select(),

"SELECT mytable.myid, mytable.name, "

"mytable.description FROM mytable",

)

self.assert_compile(

select([table1, table2]),

"SELECT mytable.myid, mytable.name, mytable.description, "

"myothertable.otherid, myothertable.othername FROM mytable, "

"myothertable",

)

def test_invalid_col_argument(self):

assert_raises(exc.ArgumentError, select, table1)

assert_raises(exc.ArgumentError, select, table1.c.myid)

def test_int_limit_offset_coercion(self):

for given, exp in [

("5", 5),

(5, 5),

(5.2, 5),

(decimal.Decimal("5"), 5),

(None, None),

]:

eq_(select().limit(given)._limit, exp)

eq_(select().offset(given)._offset, exp)

eq_(select(limit=given)._limit, exp)

eq_(select(offset=given)._offset, exp)

assert_raises(ValueError, select().limit, "foo")

assert_raises(ValueError, select().offset, "foo")

assert_raises(ValueError, select, offset="foo")

assert_raises(ValueError, select, limit="foo")

def test_limit_offset_no_int_coercion_one(self):

exp1 = literal_column("Q")

exp2 = literal_column("Y")

self.assert_compile(

select([1]).limit(exp1).offset(exp2), "SELECT 1 LIMIT Q OFFSET Y"

)

self.assert_compile(

select([1]).limit(bindparam("x")).offset(bindparam("y")),

"SELECT 1 LIMIT :x OFFSET :y",

)

def test_limit_offset_no_int_coercion_two(self):

exp1 = literal_column("Q")

exp2 = literal_column("Y")

sel = select([1]).limit(exp1).offset(exp2)

assert_raises_message(

exc.CompileError,

"This SELECT structure does not use a simple integer "

"value for limit",

getattr,

sel,

"_limit",

)

assert_raises_message(

exc.CompileError,

"This SELECT structure does not use a simple integer "

"value for offset",

getattr,

sel,

"_offset",

)

def test_limit_offset_no_int_coercion_three(self):

exp1 = bindparam("Q")

exp2 = bindparam("Y")

sel = select([1]).limit(exp1).offset(exp2)

assert_raises_message(

exc.CompileError,

"This SELECT structure does not use a simple integer "

"value for limit",

getattr,

sel,

"_limit",

)

assert_raises_message(

exc.CompileError,

"This SELECT structure does not use a simple integer "

"value for offset",

getattr,

sel,

"_offset",

)

def test_limit_offset(self):

for lim, offset, exp, params in [

(

5,

10,

"LIMIT :param_1 OFFSET :param_2",

{"param_1": 5, "param_2": 10},

),

(None, 10, "LIMIT -1 OFFSET :param_1", {"param_1": 10}),

(5, None, "LIMIT :param_1", {"param_1": 5}),

(

0,

0,

"LIMIT :param_1 OFFSET :param_2",

{"param_1": 0, "param_2": 0},

),

]:

self.assert_compile(

select([1]).limit(lim).offset(offset),

"SELECT 1 " + exp,

checkparams=params,

)

def test_select_precol_compile_ordering(self):

s1 = (

select([column("x")])

.select_from(text("a"))

.limit(5)

.scalar_subquery()

)

s2 = select([s1]).limit(10)

class MyCompiler(compiler.SQLCompiler):

def get_select_precolumns(self, select, **kw):

result = ""

if select._limit:

result += "FIRST %s " % self.process(

literal(select._limit), **kw

)

if select._offset:

result += "SKIP %s " % self.process(

literal(select._offset), **kw

)

return result

def limit_clause(self, select, **kw):

return ""

dialect = default.DefaultDialect()

dialect.statement_compiler = MyCompiler

dialect.paramstyle = "qmark"

dialect.positional = True

self.assert_compile(

s2,

"SELECT FIRST ? (SELECT FIRST ? x FROM a) AS anon_1",

checkpositional=(10, 5),

dialect=dialect,

)

def test_from_subquery(self):

"""tests placing select statements in the column clause of

another select, for the

purposes of selecting from the exported columns of that select."""

s = select([table1], table1.c.name == "jack").subquery()

self.assert_compile(

select([s], s.c.myid == 7),

"SELECT anon_1.myid, anon_1.name, anon_1.description FROM "

"(SELECT mytable.myid AS myid, "

"mytable.name AS name, mytable.description AS description "

"FROM mytable "

"WHERE mytable.name = :name_1) AS anon_1 WHERE "

"anon_1.myid = :myid_1",

)

sq = select([table1])

self.assert_compile(

sq.subquery().select(),

"SELECT anon_1.myid, anon_1.name, anon_1.description FROM "

"(SELECT mytable.myid AS myid, "

"mytable.name AS name, mytable.description "

"AS description FROM mytable) AS anon_1",

)

sq = select([table1]).alias("sq")

self.assert_compile(

sq.select(sq.c.myid == 7),

"SELECT sq.myid, sq.name, sq.description FROM "

"(SELECT mytable.myid AS myid, mytable.name AS name, "

"mytable.description AS description FROM mytable) AS sq "

"WHERE sq.myid = :myid_1",

)

sq = select(

[table1, table2],

and_(table1.c.myid == 7, table2.c.otherid == table1.c.myid),

use_labels=True,

).alias("sq")

sqstring = (

"SELECT mytable.myid AS mytable_myid, mytable.name AS "

"mytable_name, mytable.description AS mytable_description, "

"myothertable.otherid AS myothertable_otherid, "

"myothertable.othername AS myothertable_othername FROM "

"mytable, myothertable WHERE mytable.myid = :myid_1 AND "

"myothertable.otherid = mytable.myid"

)

self.assert_compile(

sq.select(),

"SELECT sq.mytable_myid, sq.mytable_name, "

"sq.mytable_description, sq.myothertable_otherid, "

"sq.myothertable_othername FROM (%s) AS sq" % sqstring,

)

sq2 = select([sq], use_labels=True).alias("sq2")

self.assert_compile(

sq2.select(),

"SELECT sq2.sq_mytable_myid, sq2.sq_mytable_name, "

"sq2.sq_mytable_description, sq2.sq_myothertable_otherid, "

"sq2.sq_myothertable_othername FROM "

"(SELECT sq.mytable_myid AS "

"sq_mytable_myid, sq.mytable_name AS sq_mytable_name, "

"sq.mytable_description AS sq_mytable_description, "

"sq.myothertable_otherid AS sq_myothertable_otherid, "

"sq.myothertable_othername AS sq_myothertable_othername "

"FROM (%s) AS sq) AS sq2" % sqstring,

)

def test_select_from_clauselist(self):

self.assert_compile(

select([ClauseList(column("a"), column("b"))]).select_from(

text("sometable")

),

"SELECT a, b FROM sometable",

)

def test_use_labels(self):

self.assert_compile(

select([table1.c.myid == 5], use_labels=True),

"SELECT mytable.myid = :myid_1 AS anon_1 FROM mytable",

)

self.assert_compile(

select([func.foo()], use_labels=True), "SELECT foo() AS foo_1"

)

# this is native_boolean=False for default dialect

self.assert_compile(

select([not_(True)], use_labels=True),

"SELECT :param_1 = 0 AS anon_1",

)

self.assert_compile(

select([cast("data", Integer)], use_labels=True),

"SELECT CAST(:param_1 AS INTEGER) AS anon_1",

)

self.assert_compile(

select(

[func.sum(func.lala(table1.c.myid).label("foo")).label("bar")]

),

"SELECT sum(lala(mytable.myid)) AS bar FROM mytable",

)

self.assert_compile(

select([keyed]), "SELECT keyed.x, keyed.y" ", keyed.z FROM keyed"

)

self.assert_compile(

select([keyed]).apply_labels(),

"SELECT keyed.x AS keyed_x, keyed.y AS "

"keyed_y, keyed.z AS keyed_z FROM keyed",

)

self.assert_compile(

select([select([keyed]).apply_labels().subquery()]).apply_labels(),

"SELECT anon_1.keyed_x AS anon_1_keyed_x, "

"anon_1.keyed_y AS anon_1_keyed_y, "

"anon_1.keyed_z AS anon_1_keyed_z "

"FROM (SELECT keyed.x AS keyed_x, keyed.y AS keyed_y, "

"keyed.z AS keyed_z FROM keyed) AS anon_1",

)

def test_paramstyles(self):

stmt = text("select :foo, :bar, :bat from sometable")

self.assert_compile(

stmt,

"select ?, ?, ? from sometable",

dialect=default.DefaultDialect(paramstyle="qmark"),

)

self.assert_compile(

stmt,

"select :foo, :bar, :bat from sometable",

dialect=default.DefaultDialect(paramstyle="named"),

)

self.assert_compile(

stmt,

"select %s, %s, %s from sometable",

dialect=default.DefaultDialect(paramstyle="format"),

)

self.assert_compile(

stmt,

"select :1, :2, :3 from sometable",

dialect=default.DefaultDialect(paramstyle="numeric"),

)

self.assert_compile(

stmt,

"select %(foo)s, %(bar)s, %(bat)s from sometable",

dialect=default.DefaultDialect(paramstyle="pyformat"),

)

def test_anon_param_name_on_keys(self):

self.assert_compile(

keyed.insert(),

"INSERT INTO keyed (x, y, z) VALUES (%(colx)s, %(coly)s, %(z)s)",

dialect=default.DefaultDialect(paramstyle="pyformat"),

)

self.assert_compile(

keyed.c.coly == 5,

"keyed.y = %(coly_1)s",

checkparams={"coly_1": 5},

dialect=default.DefaultDialect(paramstyle="pyformat"),

)

def test_dupe_columns(self):

"""as of 1.4, there's no deduping."""

self.assert_compile(

select([column("a"), column("a"), column("a")]),

"SELECT a, a, a",

dialect=default.DefaultDialect(),

)

c = column("a")

self.assert_compile(

select([c, c, c]),

"SELECT a, a, a",

dialect=default.DefaultDialect(),

)

a, b = column("a"), column("b")

self.assert_compile(

select([a, b, b, b, a, a]),

"SELECT a, b, b, b, a, a",

dialect=default.DefaultDialect(),

)

# using alternate keys.

a, b, c = (

Column("a", Integer, key="b"),

Column("b", Integer),

Column("c", Integer, key="a"),

)

self.assert_compile(

select([a, b, c, a, b, c]),

"SELECT a, b, c, a, b, c",

dialect=default.DefaultDialect(),

)

self.assert_compile(

select([bindparam("a"), bindparam("b"), bindparam("c")]),

"SELECT :a AS anon_1, :b AS anon_2, :c AS anon_3",

dialect=default.DefaultDialect(paramstyle="named"),

)

self.assert_compile(

select([bindparam("a"), bindparam("b"), bindparam("c")]),

"SELECT ? AS anon_1, ? AS anon_2, ? AS anon_3",

dialect=default.DefaultDialect(paramstyle="qmark"),

)

self.assert_compile(

select([column("a"), column("a"), column("a")]), "SELECT a, a, a"

)

s = select([bindparam("a"), bindparam("b"), bindparam("c")])

s = s.compile(dialect=default.DefaultDialect(paramstyle="qmark"))

eq_(s.positiontup, ["a", "b", "c"])

def test_overlapping_labels_use_labels(self):

foo = table("foo", column("id"), column("bar_id"))

foo_bar = table("foo_bar", column("id"))

stmt = select([foo, foo_bar]).apply_labels()

self.assert_compile(

stmt,

"SELECT foo.id AS foo_id, foo.bar_id AS foo_bar_id, "

"foo_bar.id AS foo_bar_id_1 "

"FROM foo, foo_bar",

)

def test_overlapping_labels_plus_dupes_use_labels(self):

foo = table("foo", column("id"), column("bar_id"))

foo_bar = table("foo_bar", column("id"))

# current approach is:

# 1. positional nature of columns is always maintained in all cases

# 2. two different columns that have the same label, second one

# is disambiguated

# 3. if the same column is repeated, it gets deduped using a special

# 'dedupe' label that will show two underscores

# 4. The disambiguating label generated in #2 also has to be deduped.

# 5. The derived columns, e.g. subquery().c etc. do not export the

# "dedupe" columns, at all. they are unreachable (because they

# are unreachable anyway in SQL unless you use "SELECT *")

#

# this is all new logic necessitated by #4753 since we allow columns

# to be repeated. We would still like the targeting of this column,

# both in a result set as well as in a derived selectable, to be

# unambiguous (DBs like postgresql won't let us reference an ambiguous

# label in a derived selectable even if its the same column repeated).

#

# this kind of thing happens of course because the ORM is in some

# more exotic cases writing in joins where columns may be duped.

# it might be nice to fix it on that side also, however SQLAlchemy

# has deduped columns in SELECT statements for 13 years so having a

# robust behavior when dupes are present is still very useful.

stmt = select(

[

foo.c.id,

foo.c.bar_id,

foo_bar.c.id,

foo.c.bar_id,

foo.c.id,

foo.c.bar_id,

foo_bar.c.id,

foo_bar.c.id,

]

).apply_labels()

self.assert_compile(

stmt,

"SELECT foo.id AS foo_id, "

"foo.bar_id AS foo_bar_id, " # 1. 1st foo.bar_id, as is

"foo_bar.id AS foo_bar_id_1, " # 2. 1st foo_bar.id, disamb from 1

"foo.bar_id AS foo_bar_id__1, " # 3. 2nd foo.bar_id, dedupe from 1

"foo.id AS foo_id__1, "

"foo.bar_id AS foo_bar_id__1, " # 4. 3rd foo.bar_id, same as 3

"foo_bar.id AS foo_bar_id__2, " # 5. 2nd foo_bar.id

"foo_bar.id AS foo_bar_id__2 " # 6. 3rd foo_bar.id, same as 5

"FROM foo, foo_bar",

)

# for the subquery, the labels created for repeated occurrences

# of the same column are not used. only the label applied to the

# first occurrence of each column is used

self.assert_compile(

select([stmt.subquery()]),

"SELECT "

"anon_1.foo_id, " # from 1st foo.id in derived (line 1)

"anon_1.foo_bar_id, " # from 1st foo.bar_id in derived (line 2)

"anon_1.foo_bar_id_1, " # from 1st foo_bar.id in derived (line 3)

"anon_1.foo_bar_id, " # from 1st foo.bar_id in derived (line 2)

"anon_1.foo_id, " # from 1st foo.id in derived (line 1)

"anon_1.foo_bar_id, " # from 1st foo.bar_id in derived (line 2)

"anon_1.foo_bar_id_1, " # from 1st foo_bar.id in derived (line 3)

"anon_1.foo_bar_id_1 " # from 1st foo_bar.id in derived (line 3)

"FROM ("

"SELECT foo.id AS foo_id, "

"foo.bar_id AS foo_bar_id, " # 1. 1st foo.bar_id, as is

"foo_bar.id AS foo_bar_id_1, " # 2. 1st foo_bar.id, disamb from 1

"foo.bar_id AS foo_bar_id__1, " # 3. 2nd foo.bar_id, dedupe from 1

"foo.id AS foo_id__1, "

"foo.bar_id AS foo_bar_id__1, " # 4. 3rd foo.bar_id, same as 3

"foo_bar.id AS foo_bar_id__2, " # 5. 2nd foo_bar.id

"foo_bar.id AS foo_bar_id__2 " # 6. 3rd foo_bar.id, same as 5

"FROM foo, foo_bar"

") AS anon_1",

)

def test_dupe_columns_use_labels(self):

t = table("t", column("a"), column("b"))

self.assert_compile(

select([t.c.a, t.c.a, t.c.b, t.c.a]).apply_labels(),

"SELECT t.a AS t_a, t.a AS t_a__1, t.b AS t_b, "

"t.a AS t_a__1 FROM t",

)

def test_dupe_columns_use_labels_derived_selectable(self):

t = table("t", column("a"), column("b"))

stmt = select([t.c.a, t.c.a, t.c.b, t.c.a]).apply_labels().subquery()

self.assert_compile(

select([stmt]),

"SELECT anon_1.t_a, anon_1.t_a, anon_1.t_b, anon_1.t_a FROM "

"(SELECT t.a AS t_a, t.a AS t_a__1, t.b AS t_b, t.a AS t_a__1 "

"FROM t) AS anon_1",

)

def test_dupe_columns_use_labels_mix_annotations(self):

t = table("t", column("a"), column("b"))

a, b, a_a = t.c.a, t.c.b, t.c.a._annotate({"some_orm_thing": True})

self.assert_compile(

select([a, a_a, b, a_a]).apply_labels(),

"SELECT t.a AS t_a, t.a AS t_a__1, t.b AS t_b, "

"t.a AS t_a__1 FROM t",

)

self.assert_compile(

select([a_a, a, b, a_a]).apply_labels(),

"SELECT t.a AS t_a, t.a AS t_a__1, t.b AS t_b, "

"t.a AS t_a__1 FROM t",

)

self.assert_compile(

select([a_a, a_a, b, a]).apply_labels(),

"SELECT t.a AS t_a, t.a AS t_a__1, t.b AS t_b, "

"t.a AS t_a__1 FROM t",

)

def test_dupe_columns_use_labels_derived_selectable_mix_annotations(self):

t = table("t", column("a"), column("b"))

a, b, a_a = t.c.a, t.c.b, t.c.a._annotate({"some_orm_thing": True})

stmt = select([a, a_a, b, a_a]).apply_labels().subquery()

self.assert_compile(

select([stmt]),

"SELECT anon_1.t_a, anon_1.t_a, anon_1.t_b, anon_1.t_a FROM "

"(SELECT t.a AS t_a, t.a AS t_a__1, t.b AS t_b, t.a AS t_a__1 "

"FROM t) AS anon_1",

)

def test_overlapping_labels_plus_dupes_use_labels_mix_annotations(self):

foo = table("foo", column("id"), column("bar_id"))

foo_bar = table("foo_bar", column("id"))

foo_bar__id = foo_bar.c.id._annotate({"some_orm_thing": True})

stmt = select(

[

foo.c.bar_id,

foo_bar.c.id,

foo_bar.c.id,

foo_bar__id,

foo_bar__id,

]

).apply_labels()

self.assert_compile(

stmt,

"SELECT foo.bar_id AS foo_bar_id, foo_bar.id AS foo_bar_id_1, "

"foo_bar.id AS foo_bar_id__1, foo_bar.id AS foo_bar_id__1, "

"foo_bar.id AS foo_bar_id__1 FROM foo, foo_bar",

)

def test_dupe_columns_use_labels_from_anon(self):

t = table("t", column("a"), column("b"))

a = t.alias()

# second and third occurrences of a.c.a are labeled, but are

# dupes of each other.

self.assert_compile(

select([a.c.a, a.c.a, a.c.b, a.c.a]).apply_labels(),

"SELECT t_1.a AS t_1_a, t_1.a AS t_1_a__1, t_1.b AS t_1_b, "

"t_1.a AS t_1_a__1 "

"FROM t AS t_1",

)

def test_nested_label_targeting(self):

"""test nested anonymous label generation.

"""

s1 = table1.select()

s2 = s1.alias()

s3 = select([s2], use_labels=True)

s4 = s3.alias()

s5 = select([s4], use_labels=True)

self.assert_compile(

s5,

"SELECT anon_1.anon_2_myid AS "

"anon_1_anon_2_myid, anon_1.anon_2_name AS "

"anon_1_anon_2_name, anon_1.anon_2_descript"

"ion AS anon_1_anon_2_description FROM "

"(SELECT anon_2.myid AS anon_2_myid, "

"anon_2.name AS anon_2_name, "

"anon_2.description AS anon_2_description "

"FROM (SELECT mytable.myid AS myid, "

"mytable.name AS name, mytable.description "

"AS description FROM mytable) AS anon_2) "

"AS anon_1",

)

def test_nested_label_targeting_keyed(self):

s1 = keyed.select()

s2 = s1.alias()

s3 = select([s2], use_labels=True)

self.assert_compile(

s3,

"SELECT anon_1.x AS anon_1_x, "

"anon_1.y AS anon_1_y, "

"anon_1.z AS anon_1_z FROM "

"(SELECT keyed.x AS x, keyed.y "

"AS y, keyed.z AS z FROM keyed) AS anon_1",

)

s4 = s3.alias()

s5 = select([s4], use_labels=True)

self.assert_compile(

s5,

"SELECT anon_1.anon_2_x AS anon_1_anon_2_x, "

"anon_1.anon_2_y AS anon_1_anon_2_y, "

"anon_1.anon_2_z AS anon_1_anon_2_z "

"FROM (SELECT anon_2.x AS anon_2_x, "

"anon_2.y AS anon_2_y, "

"anon_2.z AS anon_2_z FROM "

"(SELECT keyed.x AS x, keyed.y AS y, keyed.z "

"AS z FROM keyed) AS anon_2) AS anon_1",

)

def test_exists(self):

s = select([table1.c.myid]).where(table1.c.myid == 5)

self.assert_compile(

exists(s),

"EXISTS (SELECT mytable.myid FROM mytable "

"WHERE mytable.myid = :myid_1)",

)

self.assert_compile(

exists(s.scalar_subquery()),

"EXISTS (SELECT mytable.myid FROM mytable "

"WHERE mytable.myid = :myid_1)",

)

self.assert_compile(

exists([table1.c.myid], table1.c.myid == 5).select(),

"SELECT EXISTS (SELECT mytable.myid FROM "

"mytable WHERE mytable.myid = :myid_1) AS anon_1",

params={"mytable_myid": 5},

)

self.assert_compile(

select([table1, exists([1], from_obj=table2)]),

"SELECT mytable.myid, mytable.name, "

"mytable.description, EXISTS (SELECT 1 "

"FROM myothertable) AS anon_1 FROM mytable",

params={},

)

self.assert_compile(

select([table1, exists([1], from_obj=table2).label("foo")]),

"SELECT mytable.myid, mytable.name, "

"mytable.description, EXISTS (SELECT 1 "

"FROM myothertable) AS foo FROM mytable",

params={},

)

self.assert_compile(

table1.select(

exists()

.where(table2.c.otherid == table1.c.myid)

.correlate(table1)

),

"SELECT mytable.myid, mytable.name, "

"mytable.description FROM mytable WHERE "

"EXISTS (SELECT * FROM myothertable WHERE "

"myothertable.otherid = mytable.myid)",

)

self.assert_compile(

table1.select(

exists()

.where(table2.c.otherid == table1.c.myid)

.correlate(table1)

),

"SELECT mytable.myid, mytable.name, "

"mytable.description FROM mytable WHERE "

"EXISTS (SELECT * FROM myothertable WHERE "

"myothertable.otherid = mytable.myid)",

)

self.assert_compile(

select(

[

or_(

exists().where(table2.c.otherid == "foo"),

exists().where(table2.c.otherid == "bar"),

)

]

),

"SELECT (EXISTS (SELECT * FROM myothertable "

"WHERE myothertable.otherid = :otherid_1)) "

"OR (EXISTS (SELECT * FROM myothertable WHERE "

"myothertable.otherid = :otherid_2)) AS anon_1",

)

self.assert_compile(

select([exists([1])]), "SELECT EXISTS (SELECT 1) AS anon_1"

)

self.assert_compile(

select([~exists([1])]), "SELECT NOT (EXISTS (SELECT 1)) AS anon_1"

)

self.assert_compile(

select([~(~exists([1]))]),

"SELECT NOT (NOT (EXISTS (SELECT 1))) AS anon_1",

)

def test_where_subquery(self):

s = select(

[addresses.c.street],

addresses.c.user_id == users.c.user_id,

correlate=True,

).alias("s")

# don't correlate in a FROM list

self.assert_compile(

select([users, s.c.street], from_obj=s),

"SELECT users.user_id, users.user_name, "

"users.password, s.street FROM users, "

"(SELECT addresses.street AS street FROM "

"addresses, users WHERE addresses.user_id = "

"users.user_id) AS s",

)

self.assert_compile(

table1.select(

table1.c.myid

== select(

[table1.c.myid], table1.c.name == "jack"

).scalar_subquery()

),

"SELECT mytable.myid, mytable.name, "

"mytable.description FROM mytable WHERE "

"mytable.myid = (SELECT mytable.myid FROM "

"mytable WHERE mytable.name = :name_1)",

)

self.assert_compile(

table1.select(

table1.c.myid

== select(

[table2.c.otherid], table1.c.name == table2.c.othername

).scalar_subquery()

),

"SELECT mytable.myid, mytable.name, "

"mytable.description FROM mytable WHERE "

"mytable.myid = (SELECT "

"myothertable.otherid FROM myothertable "

"WHERE mytable.name = myothertable.othernam"

"e)",

)

self.assert_compile(

table1.select(exists([1], table2.c.otherid == table1.c.myid)),

"SELECT mytable.myid, mytable.name, "

"mytable.description FROM mytable WHERE "

"EXISTS (SELECT 1 FROM myothertable WHERE "

"myothertable.otherid = mytable.myid)",

)

talias = table1.alias("ta")

s = select(

[talias], exists([1], table2.c.otherid == talias.c.myid)

).subquery("sq2")

self.assert_compile(

select([s, table1]),

"SELECT sq2.myid, sq2.name, "

"sq2.description, mytable.myid, "

"mytable.name, mytable.description FROM "

"(SELECT ta.myid AS myid, ta.name AS name, "

"ta.description AS description FROM "

"mytable AS ta WHERE EXISTS (SELECT 1 FROM "

"myothertable WHERE myothertable.otherid = "

"ta.myid)) AS sq2, mytable",

)

# test constructing the outer query via append_column(), which

# occurs in the ORM's Query object

s = select(

[], exists([1], table2.c.otherid == table1.c.myid), from_obj=table1

)

s.add_columns.non_generative(s, table1)

self.assert_compile(

s,

"SELECT mytable.myid, mytable.name, "

"mytable.description FROM mytable WHERE "

"EXISTS (SELECT 1 FROM myothertable WHERE "

"myothertable.otherid = mytable.myid)",

)

def test_orderby_subquery(self):

self.assert_compile(

table1.select().order_by(

select(

[table2.c.otherid], table1.c.myid == table2.c.otherid

).scalar_subquery()

),

"SELECT mytable.myid, mytable.name, "

"mytable.description FROM mytable ORDER BY "

"(SELECT myothertable.otherid FROM "

"myothertable WHERE mytable.myid = "

"myothertable.otherid)",

)

self.assert_compile(

table1.select().order_by(

desc(

select(