"""Test the TextClause and related constructs.""" from sqlalchemy.testing import fixtures, AssertsCompiledSQL, eq_, assert_raises_message from sqlalchemy import text, select, Integer, String, Float, \ bindparam, and_, func, literal_column, exc, MetaData, Table, Column from sqlalchemy.types import NullType from sqlalchemy.sql import table, column table1 = table('mytable', column('myid', Integer), column('name', String), column('description', String), ) table2 = table( 'myothertable', column('otherid', Integer), column('othername', String), ) class CompileTest(fixtures.TestBase, AssertsCompiledSQL): __dialect__ = 'default' def test_basic(self): self.assert_compile( text("select * from foo where lala = bar"), "select * from foo where lala = bar" ) class SelectCompositionTest(fixtures.TestBase, AssertsCompiledSQL): """test the usage of text() implicit within the select() construct when strings are passed.""" __dialect__ = 'default' 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"] ), "SELECT foobar(a), pk_foo_bar(syslaal) FROM foobar " "left outer join lala on foobar.foo = lala.foo WHERE a = 12" ) 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 = s.order_by("column1") s.append_from("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"], from_obj=table1).alias('somealias').select(), "SELECT somealias.column1, somealias.column2 FROM " "(SELECT column1, column2 FROM mytable) AS somealias" ) 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 " "FROM mytable" ) def test_select_composition_five(self): # 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 column1 AS foobar, column2 AS hoho, " "mytable.myid AS mytable_myid FROM mytable" ) def test_select_composition_six(self): # test that "auto-labeling of subquery columns" # 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 column1 AS foobar, column2 AS hoho, myid FROM " "(SELECT column1 AS foobar, column2 AS hoho, " "mytable.myid AS myid FROM mytable)" ) def test_select_composition_seven(self): self.assert_compile( select(['col1', 'col2'], from_obj='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"] ), "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"], and_( "foo.id = foofoo(lala)", "datetime(foo) = Today", table1.c.myid == table2.c.otherid, ) ), "SELECT mytable.myid, mytable.name, mytable.description, " "myothertable.otherid, sysdate(), foo, bar, lala " "FROM mytable, myothertable WHERE foo.id = foofoo(lala) AND " "datetime(foo) = Today AND mytable.myid = myothertable.otherid") class BindParamTest(fixtures.TestBase, AssertsCompiledSQL): __dialect__ = 'default' def test_legacy(self): t = text("select * from foo where lala=:bar and hoho=:whee", bindparams=[bindparam('bar', 4), bindparam('whee', 7)]) self.assert_compile( t, "select * from foo where lala=:bar and hoho=:whee", checkparams={'bar': 4, 'whee': 7}, ) def test_positional(self): t = text("select * from foo where lala=:bar and hoho=:whee") t = t.bindparams(bindparam('bar', 4), bindparam('whee', 7)) self.assert_compile( t, "select * from foo where lala=:bar and hoho=:whee", checkparams={'bar': 4, 'whee': 7}, ) def test_kw(self): t = text("select * from foo where lala=:bar and hoho=:whee") t = t.bindparams(bar=4, whee=7) self.assert_compile( t, "select * from foo where lala=:bar and hoho=:whee", checkparams={'bar': 4, 'whee': 7}, ) def test_positional_plus_kw(self): t = text("select * from foo where lala=:bar and hoho=:whee") t = t.bindparams(bindparam('bar', 4), whee=7) self.assert_compile( t, "select * from foo where lala=:bar and hoho=:whee", checkparams={'bar': 4, 'whee': 7}, ) def test_literal_binds(self): t = text("select * from foo where lala=:bar and hoho=:whee") t = t.bindparams(bindparam('bar', 4), whee='whee') self.assert_compile( t, "select * from foo where lala=4 and hoho='whee'", checkparams={}, literal_binds=True ) def _assert_type_map(self, t, compare): map_ = dict( (b.key, b.type) for b in t._bindparams.values() ) for k in compare: assert compare[k]._type_affinity is map_[k]._type_affinity def test_typing_construction(self): t = text("select * from table :foo :bar :bat") self._assert_type_map(t, {"foo": NullType(), "bar": NullType(), "bat": NullType()}) t = t.bindparams(bindparam('foo', type_=String)) self._assert_type_map(t, {"foo": String(), "bar": NullType(), "bat": NullType()}) t = t.bindparams(bindparam('bar', type_=Integer)) self._assert_type_map(t, {"foo": String(), "bar": Integer(), "bat": NullType()}) t = t.bindparams(bat=45.564) self._assert_type_map(t, {"foo": String(), "bar": Integer(), "bat": Float()}) def test_binds_compiled_named(self): self.assert_compile( text("select * from foo where lala=:bar and hoho=:whee"). bindparams(bar=4, whee=7), "select * from foo where lala=%(bar)s and hoho=%(whee)s", checkparams={'bar': 4, 'whee': 7}, dialect="postgresql" ) def test_binds_compiled_positional(self): self.assert_compile( text("select * from foo where lala=:bar and hoho=:whee"). bindparams(bar=4, whee=7), "select * from foo where lala=? and hoho=?", checkparams={'bar': 4, 'whee': 7}, dialect="sqlite" ) def test_missing_bind_kw(self): assert_raises_message( exc.ArgumentError, "This text\(\) construct doesn't define a bound parameter named 'bar'", text(":foo").bindparams, foo=5, bar=7) def test_missing_bind_posn(self): assert_raises_message( exc.ArgumentError, "This text\(\) construct doesn't define a bound parameter named 'bar'", text(":foo").bindparams, bindparam( 'foo', value=5), bindparam( 'bar', value=7)) def test_escaping_colons(self): # test escaping out text() params with a backslash self.assert_compile( text("select * from foo where clock='05:06:07' " "and mork='\:mindy'"), "select * from foo where clock='05:06:07' and mork=':mindy'", checkparams={}, params={}, dialect="postgresql" ) def test_text_in_select_nonfrom(self): generate_series = text("generate_series(:x, :y, :z) as s(a)").\ bindparams(x=None, y=None, z=None) s = select([ (func.current_date() + literal_column("s.a")).label("dates") ]).select_from(generate_series) self.assert_compile( s, "SELECT CURRENT_DATE + s.a AS dates FROM " "generate_series(:x, :y, :z) as s(a)", checkparams={'y': None, 'x': None, 'z': None} ) self.assert_compile( s.params(x=5, y=6, z=7), "SELECT CURRENT_DATE + s.a AS dates FROM " "generate_series(:x, :y, :z) as s(a)", checkparams={'y': 6, 'x': 5, 'z': 7} ) class AsFromTest(fixtures.TestBase, AssertsCompiledSQL): __dialect__ = 'default' def test_basic_toplevel_resultmap_positional(self): t = text("select id, name from user").columns( column('id', Integer), column('name') ) compiled = t.compile() eq_(compiled.result_map, {'id': ('id', (t.c.id._proxies[0], 'id', 'id'), t.c.id.type), 'name': ('name', (t.c.name._proxies[0], 'name', 'name'), t.c.name.type)}) def test_basic_toplevel_resultmap(self): t = text("select id, name from user").columns(id=Integer, name=String) compiled = t.compile() eq_(compiled.result_map, {'id': ('id', (t.c.id._proxies[0], 'id', 'id'), t.c.id.type), 'name': ('name', (t.c.name._proxies[0], 'name', 'name'), t.c.name.type)}) def test_basic_subquery_resultmap(self): t = text("select id, name from user").columns(id=Integer, name=String) stmt = select([table1.c.myid]).select_from( table1.join(t, table1.c.myid == t.c.id)) compiled = stmt.compile() eq_( compiled.result_map, { "myid": ("myid", (table1.c.myid, "myid", "myid"), table1.c.myid.type), } ) def test_column_collection_ordered(self): t = text("select a, b, c from foo").columns(column('a'), column('b'), column('c')) eq_(t.c.keys(), ['a', 'b', 'c']) def test_column_collection_pos_plus_bykey(self): # overlapping positional names + type names t = text("select a, b, c from foo").columns( column('a'), column('b'), b=Integer, c=String) eq_(t.c.keys(), ['a', 'b', 'c']) eq_(t.c.b.type._type_affinity, Integer) eq_(t.c.c.type._type_affinity, String) def _xy_table_fixture(self): m = MetaData() t = Table('t', m, Column('x', Integer), Column('y', Integer)) return t def _mapping(self, stmt): compiled = stmt.compile() return dict( (elem, key) for key, elements in compiled.result_map.items() for elem in elements[1] ) def test_select_label_alt_name(self): t = self._xy_table_fixture() l1, l2 = t.c.x.label('a'), t.c.y.label('b') s = text("select x AS a, y AS b FROM t").columns(l1, l2) mapping = self._mapping(s) assert l1 in mapping assert t.c.x not in mapping def test_select_alias_label_alt_name(self): t = self._xy_table_fixture() l1, l2 = t.c.x.label('a'), t.c.y.label('b') s = text("select x AS a, y AS b FROM t").columns(l1, l2).alias() mapping = self._mapping(s) assert l1 in mapping assert t.c.x not in mapping def test_select_column(self): t = self._xy_table_fixture() x, y = t.c.x, t.c.y s = text("select x, y FROM t").columns(x, y) mapping = self._mapping(s) assert t.c.x in mapping def test_select_alias_column(self): t = self._xy_table_fixture() x, y = t.c.x, t.c.y s = text("select x, y FROM t").columns(x, y).alias() mapping = self._mapping(s) assert t.c.x in mapping def test_select_table_alias_column(self): t = self._xy_table_fixture() x, y = t.c.x, t.c.y ta = t.alias() s = text("select ta.x, ta.y FROM t AS ta").columns(ta.c.x, ta.c.y) mapping = self._mapping(s) assert x not in mapping def test_select_label_alt_name_table_alias_column(self): t = self._xy_table_fixture() x, y = t.c.x, t.c.y ta = t.alias() l1, l2 = ta.c.x.label('a'), ta.c.y.label('b') s = text("SELECT ta.x AS a, ta.y AS b FROM t AS ta").columns(l1, l2) mapping = self._mapping(s) assert x not in mapping assert l1 in mapping assert ta.c.x not in mapping def test_cte(self): t = text("select id, name from user").columns( id=Integer, name=String).cte('t') s = select([table1]).where(table1.c.myid == t.c.id) self.assert_compile( s, "WITH t AS (select id, name from user) " "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable, t WHERE mytable.myid = t.id" ) def test_alias(self): t = text("select id, name from user").columns( id=Integer, name=String).alias('t') s = select([table1]).where(table1.c.myid == t.c.id) self.assert_compile( s, "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable, (select id, name from user) AS t " "WHERE mytable.myid = t.id" ) def test_scalar_subquery(self): t = text("select id from user").columns(id=Integer) subq = t.as_scalar() assert subq.type._type_affinity is Integer()._type_affinity s = select([table1.c.myid, subq]).where(table1.c.myid == subq) self.assert_compile( s, "SELECT mytable.myid, (select id from user) AS anon_1 " "FROM mytable WHERE mytable.myid = (select id from user)" ) def test_build_bindparams(self): t = text("select id from user :foo :bar :bat") t = t.bindparams(bindparam("foo", type_=Integer)) t = t.columns(id=Integer) t = t.bindparams(bar=String) t = t.bindparams(bindparam('bat', value='bat')) eq_( set(t.element._bindparams), set(["bat", "foo", "bar"]) )