from sqlalchemy.testing import eq_, assert_raises_message, assert_raises, is_ from sqlalchemy import testing from sqlalchemy.testing import fixtures, engines from sqlalchemy import util 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) from sqlalchemy.engine import default, result as _result from sqlalchemy.testing.schema import Table, Column # ongoing - these are old tests. those which are of general use # to test a dialect are being slowly migrated to # sqlalhcemy.testing.suite users = users2 = addresses = metadata = None class QueryTest(fixtures.TestBase): __backend__ = True @classmethod def setup_class(cls): global users, users2, addresses, metadata metadata = MetaData(testing.db) users = Table( 'query_users', metadata, Column( 'user_id', INT, primary_key=True, test_needs_autoincrement=True), Column('user_name', VARCHAR(20)), test_needs_acid=True ) addresses = Table( 'query_addresses', metadata, Column( 'address_id', Integer, primary_key=True, test_needs_autoincrement=True), Column('user_id', Integer, ForeignKey('query_users.user_id')), Column('address', String(30)), test_needs_acid=True ) users2 = Table( 'u2', metadata, Column('user_id', INT, primary_key=True), Column('user_name', VARCHAR(20)), test_needs_acid=True ) metadata.create_all() @engines.close_first def teardown(self): addresses.delete().execute() users.delete().execute() users2.delete().execute() @classmethod def teardown_class(cls): metadata.drop_all() @testing.requires.multivalues_inserts def test_multivalues_insert(self): users.insert( values=[ {'user_id': 7, 'user_name': 'jack'}, {'user_id': 8, 'user_name': 'ed'}]).execute() rows = users.select().order_by(users.c.user_id).execute().fetchall() self.assert_(rows[0] == (7, 'jack')) self.assert_(rows[1] == (8, 'ed')) users.insert(values=[(9, 'jack'), (10, 'ed')]).execute() rows = users.select().order_by(users.c.user_id).execute().fetchall() self.assert_(rows[2] == (9, 'jack')) self.assert_(rows[3] == (10, 'ed')) def test_insert_heterogeneous_params(self): """test that executemany parameters are asserted to match the parameter set of the first.""" assert_raises_message( exc.StatementError, r"A value is required for bind parameter 'user_name', in " "parameter group 2 " "\(original cause: (sqlalchemy.exc.)?InvalidRequestError: A " "value is required for bind parameter 'user_name', in " "parameter group 2\) u?'INSERT INTO query_users", users.insert().execute, {'user_id': 7, 'user_name': 'jack'}, {'user_id': 8, 'user_name': 'ed'}, {'user_id': 9} ) # this succeeds however. We aren't yet doing # a length check on all subsequent parameters. users.insert().execute( {'user_id': 7}, {'user_id': 8, 'user_name': 'ed'}, {'user_id': 9} ) def test_lastrow_accessor(self): """Tests the inserted_primary_key and lastrow_has_id() functions.""" def insert_values(engine, table, values): """ Inserts a row into a table, returns the full list of values INSERTed including defaults that fired off on the DB side and detects rows that had defaults and post-fetches. """ # verify implicit_returning is working if engine.dialect.implicit_returning: ins = table.insert() comp = ins.compile(engine, column_keys=list(values)) if not set(values).issuperset( c.key for c in table.primary_key): assert comp.returning result = engine.execute(table.insert(), **values) ret = values.copy() for col, id in zip(table.primary_key, result.inserted_primary_key): ret[col.key] = id if result.lastrow_has_defaults(): criterion = and_( *[ col == id for col, id in zip(table.primary_key, result.inserted_primary_key)]) row = engine.execute(table.select(criterion)).first() for c in table.c: ret[c.key] = row[c] return ret if testing.against('firebird', 'postgresql', 'oracle', 'mssql'): assert testing.db.dialect.implicit_returning if testing.db.dialect.implicit_returning: test_engines = [ engines.testing_engine(options={'implicit_returning': False}), engines.testing_engine(options={'implicit_returning': True}), ] else: test_engines = [testing.db] for engine in test_engines: metadata = MetaData() for supported, table, values, assertvalues in [ ( {'unsupported': ['sqlite']}, Table( "t1", metadata, Column( 'id', Integer, primary_key=True, test_needs_autoincrement=True), Column('foo', String(30), primary_key=True)), {'foo': 'hi'}, {'id': 1, 'foo': 'hi'} ), ( {'unsupported': ['sqlite']}, Table( "t2", metadata, Column( 'id', Integer, primary_key=True, test_needs_autoincrement=True), Column('foo', String(30), primary_key=True), Column('bar', String(30), server_default='hi') ), {'foo': 'hi'}, {'id': 1, 'foo': 'hi', 'bar': 'hi'} ), ( {'unsupported': []}, Table( "t3", metadata, Column("id", String(40), primary_key=True), Column('foo', String(30), primary_key=True), Column("bar", String(30)) ), {'id': 'hi', 'foo': 'thisisfoo', 'bar': "thisisbar"}, {'id': 'hi', 'foo': 'thisisfoo', 'bar': "thisisbar"} ), ( {'unsupported': []}, Table( "t4", metadata, Column( 'id', Integer, Sequence('t4_id_seq', optional=True), primary_key=True), Column('foo', String(30), primary_key=True), Column('bar', String(30), server_default='hi') ), {'foo': 'hi', 'id': 1}, {'id': 1, 'foo': 'hi', 'bar': 'hi'} ), ( {'unsupported': []}, Table( "t5", metadata, Column('id', String(10), primary_key=True), Column('bar', String(30), server_default='hi') ), {'id': 'id1'}, {'id': 'id1', 'bar': 'hi'}, ), ( {'unsupported': ['sqlite']}, Table( "t6", metadata, Column( 'id', Integer, primary_key=True, test_needs_autoincrement=True), Column('bar', Integer, primary_key=True) ), {'bar': 0}, {'id': 1, 'bar': 0}, ), ]: if testing.db.name in supported['unsupported']: continue try: table.create(bind=engine, checkfirst=True) i = insert_values(engine, table, values) assert i == assertvalues, "tablename: %s %r %r" % \ (table.name, repr(i), repr(assertvalues)) finally: table.drop(bind=engine) # TODO: why not in the sqlite suite? @testing.only_on('sqlite+pysqlite') @testing.provide_metadata def test_lastrowid_zero(self): from sqlalchemy.dialects import sqlite eng = engines.testing_engine() class ExcCtx(sqlite.base.SQLiteExecutionContext): def get_lastrowid(self): return 0 eng.dialect.execution_ctx_cls = ExcCtx t = Table( 't', self.metadata, Column('x', Integer, primary_key=True), Column('y', Integer)) t.create(eng) r = eng.execute(t.insert().values(y=5)) eq_(r.inserted_primary_key, [0]) @testing.fails_on( 'sqlite', "sqlite autoincremnt doesn't work with composite pks") def test_misordered_lastrow(self): related = Table( 'related', metadata, Column('id', Integer, primary_key=True), mysql_engine='MyISAM' ) t6 = Table( "t6", metadata, Column( 'manual_id', Integer, ForeignKey('related.id'), primary_key=True), Column( 'auto_id', Integer, primary_key=True, test_needs_autoincrement=True), mysql_engine='MyISAM' ) metadata.create_all() r = related.insert().values(id=12).execute() id = r.inserted_primary_key[0] assert id == 12 r = t6.insert().values(manual_id=id).execute() eq_(r.inserted_primary_key, [12, 1]) def test_row_iteration(self): users.insert().execute( {'user_id': 7, 'user_name': 'jack'}, {'user_id': 8, 'user_name': 'ed'}, {'user_id': 9, 'user_name': 'fred'}, ) r = users.select().execute() l = [] for row in r: l.append(row) self.assert_(len(l) == 3) @testing.requires.subqueries def test_anonymous_rows(self): users.insert().execute( {'user_id': 7, 'user_name': 'jack'}, {'user_id': 8, 'user_name': 'ed'}, {'user_id': 9, 'user_name': 'fred'}, ) sel = select([users.c.user_id]).where(users.c.user_name == 'jack'). \ as_scalar() for row in select([sel + 1, sel + 3], bind=users.bind).execute(): assert row['anon_1'] == 8 assert row['anon_2'] == 10 @testing.fails_on( 'firebird', "kinterbasdb doesn't send full type information") def test_order_by_label(self): """test that a label within an ORDER BY works on each backend. This test should be modified to support [ticket:1068] when that ticket is implemented. For now, you need to put the actual string in the ORDER BY. """ users.insert().execute( {'user_id': 7, 'user_name': 'jack'}, {'user_id': 8, 'user_name': 'ed'}, {'user_id': 9, 'user_name': 'fred'}, ) concat = ("test: " + users.c.user_name).label('thedata') eq_( select([concat]).order_by("thedata").execute().fetchall(), [("test: ed",), ("test: fred",), ("test: jack",)] ) eq_( select([concat]).order_by("thedata").execute().fetchall(), [("test: ed",), ("test: fred",), ("test: jack",)] ) concat = ("test: " + users.c.user_name).label('thedata') eq_( select([concat]).order_by(desc('thedata')).execute().fetchall(), [("test: jack",), ("test: fred",), ("test: ed",)] ) @testing.requires.order_by_label_with_expression def test_order_by_label_compound(self): users.insert().execute( {'user_id': 7, 'user_name': 'jack'}, {'user_id': 8, 'user_name': 'ed'}, {'user_id': 9, 'user_name': 'fred'}, ) concat = ("test: " + users.c.user_name).label('thedata') eq_( select([concat]).order_by(literal_column('thedata') + "x"). execute().fetchall(), [("test: ed",), ("test: fred",), ("test: jack",)] ) def test_row_comparison(self): users.insert().execute(user_id=7, user_name='jack') rp = users.select().execute().first() self.assert_(rp == rp) self.assert_(not(rp != rp)) equal = (7, 'jack') self.assert_(rp == equal) self.assert_(equal == rp) self.assert_(not (rp != equal)) self.assert_(not (equal != equal)) def endless(): while True: yield 1 self.assert_(rp != endless()) self.assert_(endless() != rp) # test that everything compares the same # as it would against a tuple import operator for compare in [False, 8, endless(), 'xyz', (7, 'jack')]: for op in [ operator.eq, operator.ne, operator.gt, operator.lt, operator.ge, operator.le ]: try: control = op(equal, compare) except TypeError: # Py3K raises TypeError for some invalid comparisons assert_raises(TypeError, op, rp, compare) else: eq_(control, op(rp, compare)) try: control = op(compare, equal) except TypeError: # Py3K raises TypeError for some invalid comparisons assert_raises(TypeError, op, compare, rp) else: eq_(control, op(compare, rp)) @testing.provide_metadata def test_column_label_overlap_fallback(self): content = Table( 'content', self.metadata, Column('type', String(30)), ) bar = Table( 'bar', self.metadata, Column('content_type', String(30)) ) self.metadata.create_all(testing.db) testing.db.execute(content.insert().values(type="t1")) row = testing.db.execute(content.select(use_labels=True)).first() assert content.c.type in row assert bar.c.content_type not in row assert sql.column('content_type') in row row = testing.db.execute( select([content.c.type.label("content_type")])).first() assert content.c.type in row assert bar.c.content_type not in row assert sql.column('content_type') in row row = testing.db.execute(select([func.now().label("content_type")])). \ first() assert content.c.type not in row assert bar.c.content_type not in row assert sql.column('content_type') in row def test_pickled_rows(self): users.insert().execute( {'user_id': 7, 'user_name': 'jack'}, {'user_id': 8, 'user_name': 'ed'}, {'user_id': 9, 'user_name': 'fred'}, ) for pickle in False, True: for use_labels in False, True: result = users.select(use_labels=use_labels).order_by( users.c.user_id).execute().fetchall() if pickle: result = util.pickle.loads(util.pickle.dumps(result)) eq_( result, [(7, "jack"), (8, "ed"), (9, "fred")] ) if use_labels: eq_(result[0]['query_users_user_id'], 7) eq_( list(result[0].keys()), ["query_users_user_id", "query_users_user_name"]) else: eq_(result[0]['user_id'], 7) eq_(list(result[0].keys()), ["user_id", "user_name"]) eq_(result[0][0], 7) eq_(result[0][users.c.user_id], 7) eq_(result[0][users.c.user_name], 'jack') if not pickle or use_labels: assert_raises( exc.NoSuchColumnError, lambda: result[0][addresses.c.user_id]) else: # test with a different table. name resolution is # causing 'user_id' to match when use_labels wasn't used. eq_(result[0][addresses.c.user_id], 7) assert_raises( exc.NoSuchColumnError, lambda: result[0]['fake key']) assert_raises( exc.NoSuchColumnError, lambda: result[0][addresses.c.address_id]) def test_column_error_printing(self): row = testing.db.execute(select([1])).first() class unprintable(object): def __str__(self): raise ValueError("nope") msg = r"Could not locate column in row for column '%s'" for accessor, repl in [ ("x", "x"), (Column("q", Integer), "q"), (Column("q", Integer) + 12, r"q \+ :q_1"), (unprintable(), "unprintable element.*"), ]: assert_raises_message( exc.NoSuchColumnError, msg % repl, lambda: row[accessor] ) @testing.requires.boolean_col_expressions def test_or_and_as_columns(self): true, false = literal(True), literal(False) eq_(testing.db.execute(select([and_(true, false)])).scalar(), False) eq_(testing.db.execute(select([and_(true, true)])).scalar(), True) eq_(testing.db.execute(select([or_(true, false)])).scalar(), True) eq_(testing.db.execute(select([or_(false, false)])).scalar(), False) eq_( testing.db.execute(select([not_(or_(false, false))])).scalar(), True) row = testing.db.execute( select( [or_(false, false).label("x"), and_(true, false).label("y")])).first() assert row.x == False # noqa assert row.y == False # noqa row = testing.db.execute( select( [or_(true, false).label("x"), and_(true, false).label("y")])).first() assert row.x == True # noqa assert row.y == False # noqa def test_fetchmany(self): users.insert().execute(user_id=7, user_name='jack') users.insert().execute(user_id=8, user_name='ed') users.insert().execute(user_id=9, user_name='fred') r = users.select().execute() l = [] for row in r.fetchmany(size=2): l.append(row) self.assert_(len(l) == 2, "fetchmany(size=2) got %s rows" % len(l)) def test_like_ops(self): users.insert().execute( {'user_id': 1, 'user_name': 'apples'}, {'user_id': 2, 'user_name': 'oranges'}, {'user_id': 3, 'user_name': 'bananas'}, {'user_id': 4, 'user_name': 'legumes'}, {'user_id': 5, 'user_name': 'hi % there'}, ) for expr, result in ( (select([users.c.user_id]). where(users.c.user_name.startswith('apple')), [(1,)]), (select([users.c.user_id]). where(users.c.user_name.contains('i % t')), [(5,)]), (select([users.c.user_id]). where(users.c.user_name.endswith('anas')), [(3,)]), (select([users.c.user_id]). where(users.c.user_name.contains('i % t', escape='&')), [(5,)]), ): eq_(expr.execute().fetchall(), result) @testing.requires.mod_operator_as_percent_sign @testing.emits_warning('.*now automatically escapes.*') def test_percents_in_text(self): for expr, result in ( (text("select 6 % 10"), 6), (text("select 17 % 10"), 7), (text("select '%'"), '%'), (text("select '%%'"), '%%'), (text("select '%%%'"), '%%%'), (text("select 'hello % world'"), "hello % world") ): eq_(testing.db.scalar(expr), result) def test_ilike(self): users.insert().execute( {'user_id': 1, 'user_name': 'one'}, {'user_id': 2, 'user_name': 'TwO'}, {'user_id': 3, 'user_name': 'ONE'}, {'user_id': 4, 'user_name': 'OnE'}, ) eq_( select([users.c.user_id]).where(users.c.user_name.ilike('one')). execute().fetchall(), [(1, ), (3, ), (4, )]) eq_( select([users.c.user_id]).where(users.c.user_name.ilike('TWO')). execute().fetchall(), [(2, )]) if testing.against('postgresql'): eq_( select([users.c.user_id]). where(users.c.user_name.like('one')).execute().fetchall(), [(1, )]) eq_( select([users.c.user_id]). where(users.c.user_name.like('TWO')).execute().fetchall(), []) def test_compiled_execute(self): users.insert().execute(user_id=7, user_name='jack') s = select([users], users.c.user_id == bindparam('id')).compile() c = testing.db.connect() assert c.execute(s, id=7).fetchall()[0]['user_id'] == 7 def test_compiled_insert_execute(self): users.insert().compile().execute(user_id=7, user_name='jack') s = select([users], users.c.user_id == bindparam('id')).compile() c = testing.db.connect() assert c.execute(s, id=7).fetchall()[0]['user_id'] == 7 def test_repeated_bindparams(self): """Tests that a BindParam can be used more than once. This should be run for DB-APIs with both positional and named paramstyles. """ users.insert().execute(user_id=7, user_name='jack') users.insert().execute(user_id=8, user_name='fred') u = bindparam('userid') s = users.select(and_(users.c.user_name == u, users.c.user_name == u)) r = s.execute(userid='fred').fetchall() assert len(r) == 1 def test_bindparam_detection(self): dialect = default.DefaultDialect(paramstyle='qmark') prep = lambda q: str(sql.text(q).compile(dialect=dialect)) def a_eq(got, wanted): if got != wanted: print("Wanted %s" % wanted) print("Received %s" % got) self.assert_(got == wanted, got) a_eq(prep('select foo'), 'select foo') a_eq(prep("time='12:30:00'"), "time='12:30:00'") a_eq(prep("time='12:30:00'"), "time='12:30:00'") a_eq(prep(":this:that"), ":this:that") a_eq(prep(":this :that"), "? ?") a_eq(prep("(:this),(:that :other)"), "(?),(? ?)") a_eq(prep("(:this),(:that:other)"), "(?),(:that:other)") a_eq(prep("(:this),(:that,:other)"), "(?),(?,?)") a_eq(prep("(:that_:other)"), "(:that_:other)") a_eq(prep("(:that_ :other)"), "(? ?)") a_eq(prep("(:that_other)"), "(?)") a_eq(prep("(:that$other)"), "(?)") a_eq(prep("(:that$:other)"), "(:that$:other)") a_eq(prep(".:that$ :other."), ".? ?.") a_eq(prep(r'select \foo'), r'select \foo') a_eq(prep(r"time='12\:30:00'"), r"time='12\:30:00'") a_eq(prep(":this \:that"), "? :that") a_eq(prep(r"(\:that$other)"), "(:that$other)") a_eq(prep(r".\:that$ :other."), ".:that$ ?.") @testing.requires.standalone_binds def test_select_from_bindparam(self): """Test result row processing when selecting from a plain bind param.""" class MyInteger(TypeDecorator): impl = Integer def process_bind_param(self, value, dialect): return int(value[4:]) def process_result_value(self, value, dialect): return "INT_%d" % value eq_( testing.db.scalar(select([cast("INT_5", type_=MyInteger)])), "INT_5" ) eq_( testing.db.scalar( select([cast("INT_5", type_=MyInteger).label('foo')])), "INT_5" ) def test_order_by(self): """Exercises ORDER BY clause generation. Tests simple, compound, aliased and DESC clauses. """ users.insert().execute(user_id=1, user_name='c') users.insert().execute(user_id=2, user_name='b') users.insert().execute(user_id=3, user_name='a') def a_eq(executable, wanted): got = list(executable.execute()) eq_(got, wanted) for labels in False, True: a_eq(users.select(order_by=[users.c.user_id], use_labels=labels), [(1, 'c'), (2, 'b'), (3, 'a')]) a_eq(users.select(order_by=[users.c.user_name, users.c.user_id], use_labels=labels), [(3, 'a'), (2, 'b'), (1, 'c')]) a_eq(select([users.c.user_id.label('foo')], use_labels=labels, order_by=[users.c.user_id]), [(1,), (2,), (3,)]) a_eq(select([users.c.user_id.label('foo'), users.c.user_name], use_labels=labels, order_by=[users.c.user_name, users.c.user_id]), [(3, 'a'), (2, 'b'), (1, 'c')]) a_eq(users.select(distinct=True, use_labels=labels, order_by=[users.c.user_id]), [(1, 'c'), (2, 'b'), (3, 'a')]) a_eq(select([users.c.user_id.label('foo')], distinct=True, use_labels=labels, order_by=[users.c.user_id]), [(1,), (2,), (3,)]) a_eq(select([users.c.user_id.label('a'), users.c.user_id.label('b'), users.c.user_name], use_labels=labels, order_by=[users.c.user_id]), [(1, 1, 'c'), (2, 2, 'b'), (3, 3, 'a')]) a_eq(users.select(distinct=True, use_labels=labels, order_by=[desc(users.c.user_id)]), [(3, 'a'), (2, 'b'), (1, 'c')]) a_eq(select([users.c.user_id.label('foo')], distinct=True, use_labels=labels, order_by=[users.c.user_id.desc()]), [(3,), (2,), (1,)]) @testing.requires.nullsordering def test_order_by_nulls(self): """Exercises ORDER BY clause generation. Tests simple, compound, aliased and DESC clauses. """ users.insert().execute(user_id=1) users.insert().execute(user_id=2, user_name='b') users.insert().execute(user_id=3, user_name='a') def a_eq(executable, wanted): got = list(executable.execute()) eq_(got, wanted) for labels in False, True: a_eq(users.select(order_by=[users.c.user_name.nullsfirst()], use_labels=labels), [(1, None), (3, 'a'), (2, 'b')]) a_eq(users.select(order_by=[users.c.user_name.nullslast()], use_labels=labels), [(3, 'a'), (2, 'b'), (1, None)]) a_eq(users.select(order_by=[asc(users.c.user_name).nullsfirst()], use_labels=labels), [(1, None), (3, 'a'), (2, 'b')]) a_eq(users.select(order_by=[asc(users.c.user_name).nullslast()], use_labels=labels), [(3, 'a'), (2, 'b'), (1, None)]) a_eq(users.select(order_by=[users.c.user_name.desc().nullsfirst()], use_labels=labels), [(1, None), (2, 'b'), (3, 'a')]) a_eq( users.select( order_by=[users.c.user_name.desc().nullslast()], use_labels=labels), [(2, 'b'), (3, 'a'), (1, None)]) a_eq( users.select( order_by=[desc(users.c.user_name).nullsfirst()], use_labels=labels), [(1, None), (2, 'b'), (3, 'a')]) a_eq(users.select(order_by=[desc(users.c.user_name).nullslast()], use_labels=labels), [(2, 'b'), (3, 'a'), (1, None)]) a_eq( users.select( order_by=[users.c.user_name.nullsfirst(), users.c.user_id], use_labels=labels), [(1, None), (3, 'a'), (2, 'b')]) a_eq( users.select( order_by=[users.c.user_name.nullslast(), users.c.user_id], use_labels=labels), [(3, 'a'), (2, 'b'), (1, None)]) def test_column_slices(self): users.insert().execute(user_id=1, user_name='john') users.insert().execute(user_id=2, user_name='jack') addresses.insert().execute( address_id=1, user_id=2, address='foo@bar.com') r = text( "select * from query_addresses", bind=testing.db).execute().first() self.assert_(r[0:1] == (1,)) self.assert_(r[1:] == (2, 'foo@bar.com')) self.assert_(r[:-1] == (1, 2)) def test_column_accessor_basic_compiled(self): users.insert().execute( dict(user_id=1, user_name='john'), dict(user_id=2, user_name='jack') ) r = users.select(users.c.user_id == 2).execute().first() self.assert_(r.user_id == r['user_id'] == r[users.c.user_id] == 2) self.assert_( r.user_name == r['user_name'] == r[users.c.user_name] == 'jack') def test_column_accessor_basic_text(self): users.insert().execute( dict(user_id=1, user_name='john'), dict(user_id=2, user_name='jack') ) r = testing.db.execute( text("select * from query_users where user_id=2")).first() self.assert_(r.user_id == r['user_id'] == r[users.c.user_id] == 2) self.assert_( r.user_name == r['user_name'] == r[users.c.user_name] == 'jack') def test_column_accessor_textual_select(self): users.insert().execute( dict(user_id=1, user_name='john'), dict(user_id=2, user_name='jack') ) # 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') ).first() self.assert_(r.user_id == r['user_id'] == r[users.c.user_id] == 2) self.assert_( r.user_name == r['user_name'] == r[users.c.user_name] == 'jack') def test_column_accessor_dotted_union(self): users.insert().execute( dict(user_id=1, user_name='john'), ) # test a little sqlite weirdness - with the UNION, # cols come back as "query_users.user_id" in cursor.description r = testing.db.execute( text( "select query_users.user_id, query_users.user_name " "from query_users " "UNION select query_users.user_id, " "query_users.user_name from query_users" ) ).first() eq_(r['user_id'], 1) eq_(r['user_name'], "john") eq_(list(r.keys()), ["user_id", "user_name"]) @testing.only_on("sqlite", "sqlite specific feature") def test_column_accessor_sqlite_raw(self): users.insert().execute( dict(user_id=1, user_name='john'), ) r = text( "select query_users.user_id, query_users.user_name " "from query_users " "UNION select query_users.user_id, " "query_users.user_name from query_users", bind=testing.db).execution_options(sqlite_raw_colnames=True). \ execute().first() assert 'user_id' not in r assert 'user_name' not in r eq_(r['query_users.user_id'], 1) eq_(r['query_users.user_name'], "john") eq_(list(r.keys()), ["query_users.user_id", "query_users.user_name"]) @testing.only_on("sqlite", "sqlite specific feature") def test_column_accessor_sqlite_translated(self): users.insert().execute( dict(user_id=1, user_name='john'), ) r = text( "select query_users.user_id, query_users.user_name " "from query_users " "UNION select query_users.user_id, " "query_users.user_name from query_users", bind=testing.db).execute().first() eq_(r['user_id'], 1) eq_(r['user_name'], "john") eq_(r['query_users.user_id'], 1) eq_(r['query_users.user_name'], "john") eq_(list(r.keys()), ["user_id", "user_name"]) def test_column_accessor_labels_w_dots(self): users.insert().execute( dict(user_id=1, user_name='john'), ) # test using literal tablename.colname r = text( 'select query_users.user_id AS "query_users.user_id", ' 'query_users.user_name AS "query_users.user_name" ' 'from query_users', bind=testing.db).\ execution_options(sqlite_raw_colnames=True).execute().first() eq_(r['query_users.user_id'], 1) eq_(r['query_users.user_name'], "john") assert "user_name" not in r eq_(list(r.keys()), ["query_users.user_id", "query_users.user_name"]) def test_column_accessor_unary(self): users.insert().execute( dict(user_id=1, user_name='john'), ) # unary experssions r = select([users.c.user_name.distinct()]).order_by( users.c.user_name).execute().first() eq_(r[users.c.user_name], 'john') eq_(r.user_name, 'john') def test_column_accessor_err(self): r = testing.db.execute(select([1])).first() assert_raises_message( AttributeError, "Could not locate column in row for column 'foo'", getattr, r, "foo" ) assert_raises_message( KeyError, "Could not locate column in row for column 'foo'", lambda: r['foo'] ) def test_graceful_fetch_on_non_rows(self): """test that calling fetchone() etc. on a result that doesn't return rows fails gracefully. """ # these proxies don't work with no cursor.description present. # so they don't apply to this test at the moment. # result.FullyBufferedResultProxy, # result.BufferedRowResultProxy, # result.BufferedColumnResultProxy conn = testing.db.connect() for meth in ('fetchone', 'fetchall', 'first', 'scalar', 'fetchmany'): trans = conn.begin() result = conn.execute(users.insert(), user_id=1) assert_raises_message( exc.ResourceClosedError, "This result object does not return rows. " "It has been closed automatically.", getattr(result, meth), ) trans.rollback() @testing.requires.empty_inserts @testing.requires.returning def test_no_inserted_pk_on_returning(self): result = testing.db.execute(users.insert().returning( users.c.user_id, users.c.user_name)) assert_raises_message( exc.InvalidRequestError, r"Can't call inserted_primary_key when returning\(\) is used.", getattr, result, 'inserted_primary_key' ) def test_fetchone_til_end(self): result = testing.db.execute("select * from query_users") eq_(result.fetchone(), None) assert_raises_message( exc.ResourceClosedError, "This result object is closed.", result.fetchone ) def test_row_case_sensitive(self): row = testing.db.execute( select([ literal_column("1").label("case_insensitive"), literal_column("2").label("CaseSensitive") ]) ).first() eq_(list(row.keys()), ["case_insensitive", "CaseSensitive"]) eq_(row["case_insensitive"], 1) eq_(row["CaseSensitive"], 2) assert_raises( KeyError, lambda: row["Case_insensitive"] ) assert_raises( KeyError, lambda: row["casesensitive"] ) def test_row_case_insensitive(self): ins_db = engines.testing_engine(options={"case_sensitive": False}) row = ins_db.execute( select([ literal_column("1").label("case_insensitive"), literal_column("2").label("CaseSensitive") ]) ).first() eq_(list(row.keys()), ["case_insensitive", "CaseSensitive"]) eq_(row["case_insensitive"], 1) eq_(row["CaseSensitive"], 2) eq_(row["Case_insensitive"], 1) eq_(row["casesensitive"], 2) def test_row_as_args(self): users.insert().execute(user_id=1, user_name='john') r = users.select(users.c.user_id == 1).execute().first() users.delete().execute() users.insert().execute(r) eq_(users.select().execute().fetchall(), [(1, 'john')]) def test_result_as_args(self): users.insert().execute([ dict(user_id=1, user_name='john'), dict(user_id=2, user_name='ed')]) r = users.select().execute() users2.insert().execute(list(r)) eq_( users2.select().order_by(users2.c.user_id).execute().fetchall(), [(1, 'john'), (2, 'ed')] ) users2.delete().execute() r = users.select().execute() users2.insert().execute(*list(r)) eq_( users2.select().order_by(users2.c.user_id).execute().fetchall(), [(1, 'john'), (2, 'ed')] ) @testing.requires.duplicate_names_in_cursor_description def test_ambiguous_column(self): users.insert().execute(user_id=1, user_name='john') result = users.outerjoin(addresses).select().execute() r = result.first() assert_raises_message( exc.InvalidRequestError, "Ambiguous column name", lambda: r['user_id'] ) assert_raises_message( exc.InvalidRequestError, "Ambiguous column name", lambda: r[users.c.user_id] ) assert_raises_message( exc.InvalidRequestError, "Ambiguous column name", lambda: r[addresses.c.user_id] ) # try to trick it - fake_table isn't in the result! # we get the correct error fake_table = Table('fake', MetaData(), Column('user_id', Integer)) assert_raises_message( exc.InvalidRequestError, "Could not locate column in row for column 'fake.user_id'", lambda: r[fake_table.c.user_id] ) r = util.pickle.loads(util.pickle.dumps(r)) assert_raises_message( exc.InvalidRequestError, "Ambiguous column name", lambda: r['user_id'] ) result = users.outerjoin(addresses).select().execute() result = _result.BufferedColumnResultProxy(result.context) r = result.first() assert isinstance(r, _result.BufferedColumnRow) assert_raises_message( exc.InvalidRequestError, "Ambiguous column name", lambda: r['user_id'] ) @testing.requires.duplicate_names_in_cursor_description def test_ambiguous_column_by_col(self): users.insert().execute(user_id=1, user_name='john') ua = users.alias() u2 = users.alias() result = select([users.c.user_id, ua.c.user_id]).execute() row = result.first() assert_raises_message( exc.InvalidRequestError, "Ambiguous column name", lambda: row[users.c.user_id] ) assert_raises_message( exc.InvalidRequestError, "Ambiguous column name", lambda: row[ua.c.user_id] ) # Unfortunately, this fails - # we'd like # "Could not locate column in row" # to be raised here, but the check for # "common column" in _compare_name_for_result() # has other requirements to be more liberal. # Ultimately the # expression system would need a way to determine # if given two columns in a "proxy" relationship, if they # refer to a different parent table assert_raises_message( exc.InvalidRequestError, "Ambiguous column name", lambda: row[u2.c.user_id] ) @testing.requires.duplicate_names_in_cursor_description def test_ambiguous_column_contains(self): # ticket 2702. in 0.7 we'd get True, False. # in 0.8, both columns are present so it's True; # but when they're fetched you'll get the ambiguous error. users.insert().execute(user_id=1, user_name='john') result = select([users.c.user_id, addresses.c.user_id]).\ select_from(users.outerjoin(addresses)).execute() row = result.first() eq_( set([users.c.user_id in row, addresses.c.user_id in row]), set([True]) ) def test_ambiguous_column_by_col_plus_label(self): users.insert().execute(user_id=1, user_name='john') result = select( [users.c.user_id, type_coerce(users.c.user_id, Integer).label('foo')]).execute() row = result.first() eq_( row[users.c.user_id], 1 ) eq_( row[1], 1 ) @testing.requires.subqueries def test_column_label_targeting(self): users.insert().execute(user_id=7, user_name='ed') for s in ( users.select().alias('foo'), users.select().alias(users.name), ): row = s.select(use_labels=True).execute().first() assert row[s.c.user_id] == 7 assert row[s.c.user_name] == 'ed' def test_keys(self): users.insert().execute(user_id=1, user_name='foo') r = users.select().execute() eq_([x.lower() for x in list(r.keys())], ['user_id', 'user_name']) r = r.first() eq_([x.lower() for x in list(r.keys())], ['user_id', 'user_name']) def test_items(self): users.insert().execute(user_id=1, user_name='foo') r = users.select().execute().first() eq_( [(x[0].lower(), x[1]) for x in list(r.items())], [('user_id', 1), ('user_name', 'foo')]) def test_len(self): users.insert().execute(user_id=1, user_name='foo') r = users.select().execute().first() eq_(len(r), 2) r = testing.db.execute('select user_name, user_id from query_users'). \ first() eq_(len(r), 2) r = testing.db.execute('select user_name from query_users').first() eq_(len(r), 1) def test_sorting_in_python(self): users.insert().execute( dict(user_id=1, user_name='foo'), dict(user_id=2, user_name='bar'), dict(user_id=3, user_name='def'), ) rows = users.select().order_by(users.c.user_name).execute().fetchall() eq_(rows, [(2, 'bar'), (3, 'def'), (1, 'foo')]) eq_(sorted(rows), [(1, 'foo'), (2, 'bar'), (3, 'def')]) def test_column_order_with_simple_query(self): # should return values in column definition order users.insert().execute(user_id=1, user_name='foo') r = users.select(users.c.user_id == 1).execute().first() eq_(r[0], 1) eq_(r[1], 'foo') eq_([x.lower() for x in list(r.keys())], ['user_id', 'user_name']) eq_(list(r.values()), [1, 'foo']) def test_column_order_with_text_query(self): # should return values in query order users.insert().execute(user_id=1, user_name='foo') r = testing.db.execute('select user_name, user_id from query_users'). \ first() eq_(r[0], 'foo') eq_(r[1], 1) eq_([x.lower() for x in list(r.keys())], ['user_name', 'user_id']) eq_(list(r.values()), ['foo', 1]) @testing.crashes('oracle', 'FIXME: unknown, varify not fails_on()') @testing.crashes('firebird', 'An identifier must begin with a letter') def test_column_accessor_shadow(self): meta = MetaData(testing.db) shadowed = Table( 'test_shadowed', meta, Column('shadow_id', INT, primary_key=True), Column('shadow_name', VARCHAR(20)), Column('parent', VARCHAR(20)), Column('row', VARCHAR(40)), Column('_parent', VARCHAR(20)), Column('_row', VARCHAR(20)), ) shadowed.create(checkfirst=True) try: shadowed.insert().execute( shadow_id=1, shadow_name='The Shadow', parent='The Light', row='Without light there is no shadow', _parent='Hidden parent', _row='Hidden row') r = shadowed.select(shadowed.c.shadow_id == 1).execute().first() self.assert_( r.shadow_id == r['shadow_id'] == r[shadowed.c.shadow_id] == 1) self.assert_( r.shadow_name == r['shadow_name'] == r[shadowed.c.shadow_name] == 'The Shadow') self.assert_( r.parent == r['parent'] == r[shadowed.c.parent] == 'The Light') self.assert_( r.row == r['row'] == r[shadowed.c.row] == 'Without light there is no shadow') self.assert_(r['_parent'] == 'Hidden parent') self.assert_(r['_row'] == 'Hidden row') finally: shadowed.drop(checkfirst=True) @testing.emits_warning('.*empty sequence.*') def test_in_filtering(self): """test the behavior of the in_() function.""" users.insert().execute(user_id=7, user_name='jack') users.insert().execute(user_id=8, user_name='fred') users.insert().execute(user_id=9, user_name=None) s = users.select(users.c.user_name.in_([])) r = s.execute().fetchall() # No username is in empty set assert len(r) == 0 s = users.select(not_(users.c.user_name.in_([]))) r = s.execute().fetchall() # All usernames with a value are outside an empty set assert len(r) == 2 s = users.select(users.c.user_name.in_(['jack', 'fred'])) r = s.execute().fetchall() assert len(r) == 2 s = users.select(not_(users.c.user_name.in_(['jack', 'fred']))) r = s.execute().fetchall() # Null values are not outside any set assert len(r) == 0 @testing.emits_warning('.*empty sequence.*') @testing.fails_on('firebird', "uses sql-92 rules") @testing.fails_on('sybase', "uses sql-92 rules") @testing.fails_if( lambda: testing.against('mssql+pyodbc') and not testing.db.dialect.freetds, "uses sql-92 rules") def test_bind_in(self): """test calling IN against a bind parameter. this isn't allowed on several platforms since we generate ? = ?. """ users.insert().execute(user_id=7, user_name='jack') users.insert().execute(user_id=8, user_name='fred') users.insert().execute(user_id=9, user_name=None) u = bindparam('search_key') s = users.select(not_(u.in_([]))) r = s.execute(search_key='john').fetchall() assert len(r) == 3 r = s.execute(search_key=None).fetchall() assert len(r) == 0 @testing.emits_warning('.*empty sequence.*') def test_literal_in(self): """similar to test_bind_in but use a bind with a value.""" users.insert().execute(user_id=7, user_name='jack') users.insert().execute(user_id=8, user_name='fred') users.insert().execute(user_id=9, user_name=None) s = users.select(not_(literal("john").in_([]))) r = s.execute().fetchall() assert len(r) == 3 @testing.emits_warning('.*empty sequence.*') @testing.requires.boolean_col_expressions def test_in_filtering_advanced(self): """test the behavior of the in_() function when comparing against an empty collection, specifically that a proper boolean value is generated. """ users.insert().execute(user_id=7, user_name='jack') users.insert().execute(user_id=8, user_name='fred') users.insert().execute(user_id=9, user_name=None) s = users.select(users.c.user_name.in_([]) == True) # noqa r = s.execute().fetchall() assert len(r) == 0 s = users.select(users.c.user_name.in_([]) == False) # noqa r = s.execute().fetchall() assert len(r) == 2 s = users.select(users.c.user_name.in_([]) == None) # noqa r = s.execute().fetchall() assert len(r) == 1 class RequiredBindTest(fixtures.TablesTest): run_create_tables = None run_deletes = None @classmethod def define_tables(cls, metadata): Table( 'foo', metadata, Column('id', Integer, primary_key=True), Column('data', String(50)), Column('x', Integer) ) def _assert_raises(self, stmt, params): assert_raises_message( exc.StatementError, "A value is required for bind parameter 'x'", testing.db.execute, stmt, **params) assert_raises_message( exc.StatementError, "A value is required for bind parameter 'x'", testing.db.execute, stmt, params) def test_insert(self): stmt = self.tables.foo.insert().values( x=bindparam('x'), data=bindparam('data')) self._assert_raises(stmt, {'data': 'data'}) def test_select_where(self): stmt = select([self.tables.foo]). \ where(self.tables.foo.c.data == bindparam('data')). \ where(self.tables.foo.c.x == bindparam('x')) self._assert_raises(stmt, {'data': 'data'}) @testing.requires.standalone_binds def test_select_columns(self): stmt = select([bindparam('data'), bindparam('x')]) self._assert_raises( stmt, {'data': 'data'} ) def test_text(self): stmt = text("select * from foo where x=:x and data=:data1") self._assert_raises( stmt, {'data1': 'data'} ) def test_required_flag(self): is_(bindparam('foo').required, True) is_(bindparam('foo', required=False).required, False) is_(bindparam('foo', 'bar').required, False) is_(bindparam('foo', 'bar', required=True).required, True) c = lambda: None is_(bindparam('foo', callable_=c, required=True).required, True) is_(bindparam('foo', callable_=c).required, False) is_(bindparam('foo', callable_=c, required=False).required, False) class TableInsertTest(fixtures.TablesTest): """test for consistent insert behavior across dialects regarding the inline=True flag, lower-case 't' tables. """ run_create_tables = 'each' __backend__ = True @classmethod def define_tables(cls, metadata): Table( 'foo', metadata, Column('id', Integer, Sequence('t_id_seq'), primary_key=True), Column('data', String(50)), Column('x', Integer) ) def _fixture(self, types=True): if types: t = sql.table( 'foo', sql.column('id', Integer), sql.column('data', String), sql.column('x', Integer)) else: t = sql.table( 'foo', sql.column('id'), sql.column('data'), sql.column('x')) return t def _test(self, stmt, row, returning=None, inserted_primary_key=False): r = testing.db.execute(stmt) if returning: returned = r.first() eq_(returned, returning) elif inserted_primary_key is not False: eq_(r.inserted_primary_key, inserted_primary_key) eq_(testing.db.execute(self.tables.foo.select()).first(), row) def _test_multi(self, stmt, rows, data): testing.db.execute(stmt, rows) eq_( testing.db.execute( self.tables.foo.select(). order_by(self.tables.foo.c.id)).fetchall(), data) @testing.requires.sequences def test_expicit_sequence(self): t = self._fixture() self._test( t.insert().values( id=func.next_value(Sequence('t_id_seq')), data='data', x=5), (1, 'data', 5) ) def test_uppercase(self): t = self.tables.foo self._test( t.insert().values(id=1, data='data', x=5), (1, 'data', 5), inserted_primary_key=[1] ) def test_uppercase_inline(self): t = self.tables.foo self._test( t.insert(inline=True).values(id=1, data='data', x=5), (1, 'data', 5), inserted_primary_key=[1] ) @testing.crashes( "mssql+pyodbc", "Pyodbc + SQL Server + Py3K, some decimal handling issue") def test_uppercase_inline_implicit(self): t = self.tables.foo self._test( t.insert(inline=True).values(data='data', x=5), (1, 'data', 5), inserted_primary_key=[None] ) def test_uppercase_implicit(self): t = self.tables.foo self._test( t.insert().values(data='data', x=5), (1, 'data', 5), inserted_primary_key=[1] ) def test_uppercase_direct_params(self): t = self.tables.foo self._test( t.insert().values(id=1, data='data', x=5), (1, 'data', 5), inserted_primary_key=[1] ) @testing.requires.returning def test_uppercase_direct_params_returning(self): t = self.tables.foo self._test( t.insert().values(id=1, data='data', x=5).returning(t.c.id, t.c.x), (1, 'data', 5), returning=(1, 5) ) @testing.fails_on( 'mssql', "lowercase table doesn't support identity insert disable") def test_direct_params(self): t = self._fixture() self._test( t.insert().values(id=1, data='data', x=5), (1, 'data', 5), inserted_primary_key=[] ) @testing.fails_on( 'mssql', "lowercase table doesn't support identity insert disable") @testing.requires.returning def test_direct_params_returning(self): t = self._fixture() self._test( t.insert().values(id=1, data='data', x=5).returning(t.c.id, t.c.x), (1, 'data', 5), returning=(1, 5) ) @testing.requires.emulated_lastrowid def test_implicit_pk(self): t = self._fixture() self._test( t.insert().values(data='data', x=5), (1, 'data', 5), inserted_primary_key=[] ) @testing.requires.emulated_lastrowid def test_implicit_pk_multi_rows(self): t = self._fixture() self._test_multi( t.insert(), [ {'data': 'd1', 'x': 5}, {'data': 'd2', 'x': 6}, {'data': 'd3', 'x': 7}, ], [ (1, 'd1', 5), (2, 'd2', 6), (3, 'd3', 7) ], ) @testing.requires.emulated_lastrowid def test_implicit_pk_inline(self): t = self._fixture() self._test( t.insert(inline=True).values(data='data', x=5), (1, 'data', 5), inserted_primary_key=[] ) class KeyTargetingTest(fixtures.TablesTest): run_inserts = 'once' run_deletes = None __backend__ = True @classmethod def define_tables(cls, metadata): Table( 'keyed1', metadata, Column("a", CHAR(2), key="b"), Column("c", CHAR(2), key="q") ) Table('keyed2', metadata, Column("a", CHAR(2)), Column("b", CHAR(2))) Table('keyed3', metadata, Column("a", CHAR(2)), Column("d", CHAR(2))) Table('keyed4', metadata, Column("b", CHAR(2)), Column("q", CHAR(2))) Table('content', metadata, Column('t', String(30), key="type")) Table('bar', metadata, Column('ctype', String(30), key="content_type")) if testing.requires.schemas.enabled: Table( 'wschema', metadata, Column("a", CHAR(2), key="b"), Column("c", CHAR(2), key="q"), schema="test_schema" ) @classmethod def insert_data(cls): cls.tables.keyed1.insert().execute(dict(b="a1", q="c1")) cls.tables.keyed2.insert().execute(dict(a="a2", b="b2")) cls.tables.keyed3.insert().execute(dict(a="a3", d="d3")) cls.tables.keyed4.insert().execute(dict(b="b4", q="q4")) cls.tables.content.insert().execute(type="t1") if testing.requires.schemas.enabled: cls.tables['test_schema.wschema'].insert().execute( dict(b="a1", q="c1")) @testing.requires.schemas def test_keyed_accessor_wschema(self): keyed1 = self.tables['test_schema.wschema'] row = testing.db.execute(keyed1.select()).first() eq_(row.b, "a1") eq_(row.q, "c1") eq_(row.a, "a1") eq_(row.c, "c1") def test_keyed_accessor_single(self): keyed1 = self.tables.keyed1 row = testing.db.execute(keyed1.select()).first() eq_(row.b, "a1") eq_(row.q, "c1") eq_(row.a, "a1") eq_(row.c, "c1") def test_keyed_accessor_single_labeled(self): keyed1 = self.tables.keyed1 row = testing.db.execute(keyed1.select().apply_labels()).first() eq_(row.keyed1_b, "a1") eq_(row.keyed1_q, "c1") eq_(row.keyed1_a, "a1") eq_(row.keyed1_c, "c1") @testing.requires.duplicate_names_in_cursor_description def test_keyed_accessor_composite_conflict_2(self): keyed1 = self.tables.keyed1 keyed2 = self.tables.keyed2 row = testing.db.execute(select([keyed1, keyed2])).first() # row.b is unambiguous eq_(row.b, "b2") # row.a is ambiguous assert_raises_message( exc.InvalidRequestError, "Ambig", getattr, row, "a" ) def test_keyed_accessor_composite_names_precedent(self): keyed1 = self.tables.keyed1 keyed4 = self.tables.keyed4 row = testing.db.execute(select([keyed1, keyed4])).first() eq_(row.b, "b4") eq_(row.q, "q4") eq_(row.a, "a1") eq_(row.c, "c1") @testing.requires.duplicate_names_in_cursor_description def test_keyed_accessor_composite_keys_precedent(self): keyed1 = self.tables.keyed1 keyed3 = self.tables.keyed3 row = testing.db.execute(select([keyed1, keyed3])).first() eq_(row.q, "c1") assert_raises_message( exc.InvalidRequestError, "Ambiguous column name 'b'", getattr, row, "b" ) assert_raises_message( exc.InvalidRequestError, "Ambiguous column name 'a'", getattr, row, "a" ) eq_(row.d, "d3") def test_keyed_accessor_composite_labeled(self): keyed1 = self.tables.keyed1 keyed2 = self.tables.keyed2 row = testing.db.execute(select([keyed1, keyed2]).apply_labels()). \ first() eq_(row.keyed1_b, "a1") eq_(row.keyed1_a, "a1") eq_(row.keyed1_q, "c1") eq_(row.keyed1_c, "c1") eq_(row.keyed2_a, "a2") eq_(row.keyed2_b, "b2") assert_raises(KeyError, lambda: row['keyed2_c']) assert_raises(KeyError, lambda: row['keyed2_q']) def test_column_label_overlap_fallback(self): content, bar = self.tables.content, self.tables.bar row = testing.db.execute( select([content.c.type.label("content_type")])).first() assert content.c.type not in row assert bar.c.content_type not in row assert sql.column('content_type') in row row = testing.db.execute(select([func.now().label("content_type")])). \ first() assert content.c.type not in row assert bar.c.content_type not in row assert sql.column('content_type') in row def test_column_label_overlap_fallback_2(self): content, bar = self.tables.content, self.tables.bar row = testing.db.execute(content.select(use_labels=True)).first() assert content.c.type in row assert bar.c.content_type not in row assert sql.column('content_type') not in row def test_columnclause_schema_column_one(self): keyed2 = self.tables.keyed2 # this is addressed by [ticket:2932] # ColumnClause._compare_name_for_result allows the # 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") row = testing.db.execute(stmt).first() assert keyed2.c.a in row assert keyed2.c.b in row assert a in row assert b in row def test_columnclause_schema_column_two(self): keyed2 = self.tables.keyed2 a, b = sql.column('a'), sql.column('b') stmt = select([keyed2.c.a, keyed2.c.b]) row = testing.db.execute(stmt).first() assert keyed2.c.a in row assert keyed2.c.b in row assert a in row assert b in row def test_columnclause_schema_column_three(self): keyed2 = self.tables.keyed2 # this is also addressed by [ticket:2932] a, b = sql.column('a'), sql.column('b') stmt = text("select a, b from keyed2").columns(a=CHAR, b=CHAR) row = testing.db.execute(stmt).first() assert keyed2.c.a in row assert keyed2.c.b in row assert a in row assert b in row assert stmt.c.a in row assert stmt.c.b in row def test_columnclause_schema_column_four(self): keyed2 = self.tables.keyed2 # this is also addressed by [ticket:2932] a, b = sql.column('keyed2_a'), sql.column('keyed2_b') stmt = text("select a AS keyed2_a, b AS keyed2_b from keyed2").columns( a, b) row = testing.db.execute(stmt).first() assert keyed2.c.a in row assert keyed2.c.b in row assert a in row assert b in row assert stmt.c.keyed2_a in row assert stmt.c.keyed2_b in row def test_columnclause_schema_column_five(self): keyed2 = self.tables.keyed2 # this is also addressed by [ticket:2932] stmt = text("select a AS keyed2_a, b AS keyed2_b from keyed2").columns( keyed2_a=CHAR, keyed2_b=CHAR) row = testing.db.execute(stmt).first() assert keyed2.c.a in row assert keyed2.c.b in row assert stmt.c.keyed2_a in row assert stmt.c.keyed2_b in row class LimitTest(fixtures.TestBase): __backend__ = True @classmethod def setup_class(cls): global users, addresses, metadata metadata = MetaData(testing.db) users = Table( 'query_users', metadata, Column('user_id', INT, primary_key=True), Column('user_name', VARCHAR(20)), ) addresses = Table( 'query_addresses', metadata, Column('address_id', Integer, primary_key=True), Column('user_id', Integer, ForeignKey('query_users.user_id')), Column('address', String(30))) metadata.create_all() users.insert().execute(user_id=1, user_name='john') addresses.insert().execute(address_id=1, user_id=1, address='addr1') users.insert().execute(user_id=2, user_name='jack') addresses.insert().execute(address_id=2, user_id=2, address='addr1') users.insert().execute(user_id=3, user_name='ed') addresses.insert().execute(address_id=3, user_id=3, address='addr2') users.insert().execute(user_id=4, user_name='wendy') addresses.insert().execute(address_id=4, user_id=4, address='addr3') users.insert().execute(user_id=5, user_name='laura') addresses.insert().execute(address_id=5, user_id=5, address='addr4') users.insert().execute(user_id=6, user_name='ralph') addresses.insert().execute(address_id=6, user_id=6, address='addr5') users.insert().execute(user_id=7, user_name='fido') addresses.insert().execute(address_id=7, user_id=7, address='addr5') @classmethod def teardown_class(cls): metadata.drop_all() def test_select_limit(self): r = users.select(limit=3, order_by=[users.c.user_id]).execute(). \ fetchall() self.assert_(r == [(1, 'john'), (2, 'jack'), (3, 'ed')], repr(r)) @testing.requires.offset def test_select_limit_offset(self): """Test the interaction between limit and offset""" r = users.select(limit=3, offset=2, order_by=[users.c.user_id]). \ execute().fetchall() self.assert_(r == [(3, 'ed'), (4, 'wendy'), (5, 'laura')]) r = users.select(offset=5, order_by=[users.c.user_id]).execute(). \ fetchall() self.assert_(r == [(6, 'ralph'), (7, 'fido')]) def test_select_distinct_limit(self): """Test the interaction between limit and distinct""" r = sorted( [x[0] for x in select([addresses.c.address]).distinct(). limit(3).order_by(addresses.c.address).execute().fetchall()]) self.assert_(len(r) == 3, repr(r)) self.assert_(r[0] != r[1] and r[1] != r[2], repr(r)) @testing.requires.offset @testing.fails_on('mssql', 'FIXME: unknown') def test_select_distinct_offset(self): """Test the interaction between distinct and offset""" r = sorted( [x[0] for x in select([addresses.c.address]).distinct(). offset(1).order_by(addresses.c.address). execute().fetchall()]) eq_(len(r), 4) self.assert_(r[0] != r[1] and r[1] != r[2] and r[2] != [3], repr(r)) @testing.requires.offset def test_select_distinct_limit_offset(self): """Test the interaction between limit and limit/offset""" r = select([addresses.c.address]).order_by(addresses.c.address). \ distinct().offset(2).limit(3).execute().fetchall() self.assert_(len(r) == 3, repr(r)) self.assert_(r[0] != r[1] and r[1] != r[2], repr(r)) class CompoundTest(fixtures.TestBase): """test compound statements like UNION, INTERSECT, particularly their ability to nest on different databases.""" __backend__ = True @classmethod def setup_class(cls): global metadata, t1, t2, t3 metadata = MetaData(testing.db) t1 = Table( 't1', metadata, Column( 'col1', Integer, test_needs_autoincrement=True, primary_key=True), Column('col2', String(30)), Column('col3', String(40)), Column('col4', String(30))) t2 = Table( 't2', metadata, Column( 'col1', Integer, test_needs_autoincrement=True, primary_key=True), Column('col2', String(30)), Column('col3', String(40)), Column('col4', String(30))) t3 = Table( 't3', metadata, Column( 'col1', Integer, test_needs_autoincrement=True, primary_key=True), Column('col2', String(30)), Column('col3', String(40)), Column('col4', String(30))) metadata.create_all() t1.insert().execute([ dict(col2="t1col2r1", col3="aaa", col4="aaa"), dict(col2="t1col2r2", col3="bbb", col4="bbb"), dict(col2="t1col2r3", col3="ccc", col4="ccc"), ]) t2.insert().execute([ dict(col2="t2col2r1", col3="aaa", col4="bbb"), dict(col2="t2col2r2", col3="bbb", col4="ccc"), dict(col2="t2col2r3", col3="ccc", col4="aaa"), ]) t3.insert().execute([ dict(col2="t3col2r1", col3="aaa", col4="ccc"), dict(col2="t3col2r2", col3="bbb", col4="aaa"), dict(col2="t3col2r3", col3="ccc", col4="bbb"), ]) @engines.close_first def teardown(self): pass @classmethod def teardown_class(cls): metadata.drop_all() def _fetchall_sorted(self, executed): return sorted([tuple(row) for row in executed.fetchall()]) @testing.requires.subqueries def test_union(self): (s1, s2) = ( select([t1.c.col3.label('col3'), t1.c.col4.label('col4')], t1.c.col2.in_(["t1col2r1", "t1col2r2"])), select([t2.c.col3.label('col3'), t2.c.col4.label('col4')], t2.c.col2.in_(["t2col2r2", "t2col2r3"])) ) u = union(s1, s2) wanted = [('aaa', 'aaa'), ('bbb', 'bbb'), ('bbb', 'ccc'), ('ccc', 'aaa')] found1 = self._fetchall_sorted(u.execute()) eq_(found1, wanted) found2 = self._fetchall_sorted(u.alias('bar').select().execute()) eq_(found2, wanted) @testing.fails_on('firebird', "doesn't like ORDER BY with UNIONs") def test_union_ordered(self): (s1, s2) = ( select([t1.c.col3.label('col3'), t1.c.col4.label('col4')], t1.c.col2.in_(["t1col2r1", "t1col2r2"])), select([t2.c.col3.label('col3'), t2.c.col4.label('col4')], t2.c.col2.in_(["t2col2r2", "t2col2r3"])) ) u = union(s1, s2, order_by=['col3', 'col4']) wanted = [('aaa', 'aaa'), ('bbb', 'bbb'), ('bbb', 'ccc'), ('ccc', 'aaa')] eq_(u.execute().fetchall(), wanted) @testing.fails_on('firebird', "doesn't like ORDER BY with UNIONs") @testing.requires.subqueries def test_union_ordered_alias(self): (s1, s2) = ( select([t1.c.col3.label('col3'), t1.c.col4.label('col4')], t1.c.col2.in_(["t1col2r1", "t1col2r2"])), select([t2.c.col3.label('col3'), t2.c.col4.label('col4')], t2.c.col2.in_(["t2col2r2", "t2col2r3"])) ) u = union(s1, s2, order_by=['col3', 'col4']) wanted = [('aaa', 'aaa'), ('bbb', 'bbb'), ('bbb', 'ccc'), ('ccc', 'aaa')] eq_(u.alias('bar').select().execute().fetchall(), wanted) @testing.crashes('oracle', 'FIXME: unknown, verify not fails_on') @testing.fails_on( 'firebird', "has trouble extracting anonymous column from union subquery") @testing.fails_on('mysql', 'FIXME: unknown') @testing.fails_on('sqlite', 'FIXME: unknown') def test_union_all(self): e = union_all( select([t1.c.col3]), union( select([t1.c.col3]), select([t1.c.col3]), ) ) wanted = [('aaa',), ('aaa',), ('bbb',), ('bbb',), ('ccc',), ('ccc',)] found1 = self._fetchall_sorted(e.execute()) eq_(found1, wanted) found2 = self._fetchall_sorted(e.alias('foo').select().execute()) eq_(found2, wanted) def test_union_all_lightweight(self): """like test_union_all, but breaks the sub-union into a subquery with an explicit column reference on the outside, more palatable to a wider variety of engines. """ u = union( select([t1.c.col3]), select([t1.c.col3]), ).alias() e = union_all( select([t1.c.col3]), select([u.c.col3]) ) wanted = [('aaa',), ('aaa',), ('bbb',), ('bbb',), ('ccc',), ('ccc',)] found1 = self._fetchall_sorted(e.execute()) eq_(found1, wanted) found2 = self._fetchall_sorted(e.alias('foo').select().execute()) eq_(found2, wanted) @testing.requires.intersect def test_intersect(self): i = intersect( select([t2.c.col3, t2.c.col4]), select([t2.c.col3, t2.c.col4], t2.c.col4 == t3.c.col3) ) wanted = [('aaa', 'bbb'), ('bbb', 'ccc'), ('ccc', 'aaa')] found1 = self._fetchall_sorted(i.execute()) eq_(found1, wanted) found2 = self._fetchall_sorted(i.alias('bar').select().execute()) eq_(found2, wanted) @testing.requires.except_ @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]), select([t2.c.col3, t2.c.col4]), select([t3.c.col3, t3.c.col4]), ), select([t2.c.col3, t2.c.col4])) wanted = [('aaa', 'aaa'), ('aaa', 'ccc'), ('bbb', 'aaa'), ('bbb', 'bbb'), ('ccc', 'bbb'), ('ccc', 'ccc')] found = self._fetchall_sorted(e.alias().select().execute()) eq_(found, wanted) @testing.requires.except_ 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().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().select().execute()) eq_(found2, wanted) @testing.fails_on('sqlite', "Can't handle this style of nesting") @testing.requires.except_ def test_except_style3(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 ) ) eq_(e.execute().fetchall(), [('ccc',)]) eq_(e.alias('foo').select().execute().fetchall(), [('ccc',)]) @testing.requires.except_ 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.requires.intersect @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.requires.intersect 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.requires.intersect def test_intersect_unions_3(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().select() ) wanted = [('aaa', 'bbb'), ('bbb', 'ccc'), ('ccc', 'aaa')] found = self._fetchall_sorted(u.execute()) eq_(found, wanted) @testing.requires.intersect def test_composite_alias(self): ua = 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().select() ).alias() wanted = [('aaa', 'bbb'), ('bbb', 'ccc'), ('ccc', 'aaa')] found = self._fetchall_sorted(ua.select().execute()) eq_(found, wanted) t1 = t2 = t3 = None class JoinTest(fixtures.TestBase): """Tests join execution. The compiled SQL emitted by the dialect might be ANSI joins or theta joins ('old oracle style', with (+) for OUTER). This test tries to exercise join syntax and uncover any inconsistencies in `JOIN rhs ON lhs.col=rhs.col` vs `rhs.col=lhs.col`. At least one database seems to be sensitive to this. """ __backend__ = True @classmethod def setup_class(cls): global metadata global t1, t2, t3 metadata = MetaData(testing.db) t1 = Table('t1', metadata, Column('t1_id', Integer, primary_key=True), Column('name', String(32))) t2 = Table('t2', metadata, Column('t2_id', Integer, primary_key=True), Column('t1_id', Integer, ForeignKey('t1.t1_id')), Column('name', String(32))) t3 = Table('t3', metadata, Column('t3_id', Integer, primary_key=True), Column('t2_id', Integer, ForeignKey('t2.t2_id')), Column('name', String(32))) metadata.drop_all() metadata.create_all() # t1.10 -> t2.20 -> t3.30 # t1.11 -> t2.21 # t1.12 t1.insert().execute({'t1_id': 10, 'name': 't1 #10'}, {'t1_id': 11, 'name': 't1 #11'}, {'t1_id': 12, 'name': 't1 #12'}) t2.insert().execute({'t2_id': 20, 't1_id': 10, 'name': 't2 #20'}, {'t2_id': 21, 't1_id': 11, 'name': 't2 #21'}) t3.insert().execute({'t3_id': 30, 't2_id': 20, 'name': 't3 #30'}) @classmethod def teardown_class(cls): metadata.drop_all() def assertRows(self, statement, expected): """Execute a statement and assert that rows returned equal expected.""" found = sorted([tuple(row) for row in statement.execute().fetchall()]) eq_(found, sorted(expected)) def test_join_x1(self): """Joins t1->t2.""" for criteria in (t1.c.t1_id == t2.c.t1_id, t2.c.t1_id == t1.c.t1_id): expr = select( [t1.c.t1_id, t2.c.t2_id], from_obj=[t1.join(t2, criteria)]) self.assertRows(expr, [(10, 20), (11, 21)]) def test_join_x2(self): """Joins t1->t2->t3.""" for criteria in (t1.c.t1_id == t2.c.t1_id, t2.c.t1_id == t1.c.t1_id): expr = select( [t1.c.t1_id, t2.c.t2_id], from_obj=[t1.join(t2, criteria)]) self.assertRows(expr, [(10, 20), (11, 21)]) def test_outerjoin_x1(self): """Outer joins t1->t2.""" for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id): expr = select( [t1.c.t1_id, t2.c.t2_id], from_obj=[t1.join(t2).join(t3, criteria)]) self.assertRows(expr, [(10, 20)]) def test_outerjoin_x2(self): """Outer joins t1->t2,t3.""" for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id): expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], from_obj=[t1.outerjoin(t2, t1.c.t1_id == t2.c.t1_id). outerjoin(t3, criteria)]) self.assertRows( expr, [(10, 20, 30), (11, 21, None), (12, None, None)]) def test_outerjoin_where_x2_t1(self): """Outer joins t1->t2,t3, where on t1.""" for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id): expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], t1.c.name == 't1 #10', from_obj=[(t1.outerjoin(t2, t1.c.t1_id == t2.c.t1_id). outerjoin(t3, criteria))]) self.assertRows(expr, [(10, 20, 30)]) expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], t1.c.t1_id < 12, from_obj=[(t1.outerjoin(t2, t1.c.t1_id == t2.c.t1_id). outerjoin(t3, criteria))]) self.assertRows(expr, [(10, 20, 30), (11, 21, None)]) def test_outerjoin_where_x2_t2(self): """Outer joins t1->t2,t3, where on t2.""" for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id): expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], t2.c.name == 't2 #20', from_obj=[(t1.outerjoin(t2, t1.c.t1_id == t2.c.t1_id). outerjoin(t3, criteria))]) self.assertRows(expr, [(10, 20, 30)]) expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], t2.c.t2_id < 29, from_obj=[(t1.outerjoin(t2, t1.c.t1_id == t2.c.t1_id). outerjoin(t3, criteria))]) self.assertRows(expr, [(10, 20, 30), (11, 21, None)]) def test_outerjoin_where_x2_t3(self): """Outer joins t1->t2,t3, where on t3.""" for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id): expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], t3.c.name == 't3 #30', from_obj=[(t1.outerjoin(t2, t1.c.t1_id == t2.c.t1_id). outerjoin(t3, criteria))]) self.assertRows(expr, [(10, 20, 30)]) expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], t3.c.t3_id < 39, from_obj=[(t1.outerjoin(t2, t1.c.t1_id == t2.c.t1_id). outerjoin(t3, criteria))]) self.assertRows(expr, [(10, 20, 30)]) def test_outerjoin_where_x2_t1t3(self): """Outer joins t1->t2,t3, where on t1 and t3.""" for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id): expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], and_(t1.c.name == 't1 #10', t3.c.name == 't3 #30'), from_obj=[(t1.outerjoin(t2, t1.c.t1_id == t2.c.t1_id). outerjoin(t3, criteria))]) self.assertRows(expr, [(10, 20, 30)]) expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], and_(t1.c.t1_id < 19, t3.c.t3_id < 39), from_obj=[(t1.outerjoin(t2, t1.c.t1_id == t2.c.t1_id). outerjoin(t3, criteria))]) self.assertRows(expr, [(10, 20, 30)]) def test_outerjoin_where_x2_t1t2(self): """Outer joins t1->t2,t3, where on t1 and t2.""" for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id): expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], and_(t1.c.name == 't1 #10', t2.c.name == 't2 #20'), from_obj=[(t1.outerjoin(t2, t1.c.t1_id == t2.c.t1_id). outerjoin(t3, criteria))]) self.assertRows(expr, [(10, 20, 30)]) expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], and_(t1.c.t1_id < 12, t2.c.t2_id < 39), from_obj=[(t1.outerjoin(t2, t1.c.t1_id == t2.c.t1_id). outerjoin(t3, criteria))]) self.assertRows(expr, [(10, 20, 30), (11, 21, None)]) def test_outerjoin_where_x2_t1t2t3(self): """Outer joins t1->t2,t3, where on t1, t2 and t3.""" for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id): expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], and_(t1.c.name == 't1 #10', t2.c.name == 't2 #20', t3.c.name == 't3 #30'), from_obj=[(t1.outerjoin(t2, t1.c.t1_id == t2.c.t1_id). outerjoin(t3, criteria))]) self.assertRows(expr, [(10, 20, 30)]) expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], and_(t1.c.t1_id < 19, t2.c.t2_id < 29, t3.c.t3_id < 39), from_obj=[ (t1.outerjoin(t2, t1.c.t1_id == t2.c.t1_id). outerjoin(t3, criteria))]) self.assertRows(expr, [(10, 20, 30)]) def test_mixed(self): """Joins t1->t2, outer t2->t3.""" for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id): expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], from_obj=[(t1.join(t2).outerjoin(t3, criteria))]) print(expr) self.assertRows(expr, [(10, 20, 30), (11, 21, None)]) def test_mixed_where(self): """Joins t1->t2, outer t2->t3, plus a where on each table in turn.""" for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id): expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], t1.c.name == 't1 #10', from_obj=[(t1.join(t2).outerjoin(t3, criteria))]) self.assertRows(expr, [(10, 20, 30)]) expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], t2.c.name == 't2 #20', from_obj=[(t1.join(t2).outerjoin(t3, criteria))]) self.assertRows(expr, [(10, 20, 30)]) expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], t3.c.name == 't3 #30', from_obj=[(t1.join(t2).outerjoin(t3, criteria))]) self.assertRows(expr, [(10, 20, 30)]) expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], and_(t1.c.name == 't1 #10', t2.c.name == 't2 #20'), from_obj=[(t1.join(t2).outerjoin(t3, criteria))]) self.assertRows(expr, [(10, 20, 30)]) expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], and_(t2.c.name == 't2 #20', t3.c.name == 't3 #30'), from_obj=[(t1.join(t2).outerjoin(t3, criteria))]) self.assertRows(expr, [(10, 20, 30)]) expr = select( [t1.c.t1_id, t2.c.t2_id, t3.c.t3_id], and_(t1.c.name == 't1 #10', t2.c.name == 't2 #20', t3.c.name == 't3 #30'), from_obj=[(t1.join(t2).outerjoin(t3, criteria))]) self.assertRows(expr, [(10, 20, 30)]) metadata = flds = None class OperatorTest(fixtures.TestBase): __backend__ = True @classmethod def setup_class(cls): global metadata, flds metadata = MetaData(testing.db) flds = Table( 'flds', metadata, Column( 'idcol', Integer, primary_key=True, test_needs_autoincrement=True), Column('intcol', Integer), Column('strcol', String(50)), ) metadata.create_all() flds.insert().execute([ dict(intcol=5, strcol='foo'), dict(intcol=13, strcol='bar') ]) @classmethod def teardown_class(cls): metadata.drop_all() # TODO: seems like more tests warranted for this setup. def test_modulo(self): eq_( select([flds.c.intcol % 3], order_by=flds.c.idcol).execute().fetchall(), [(2,), (1,)] ) @testing.requires.window_functions def test_over(self): eq_( select([ flds.c.intcol, func.row_number().over(order_by=flds.c.strcol) ]).execute().fetchall(), [(13, 1), (5, 2)] )