mirror of
https://github.com/sqlalchemy/sqlalchemy.git
synced 2026-05-10 02:39:59 -04:00
Mike Bayer
206 lines
10 KiB
Python
206 lines
10 KiB
Python
from sqlalchemy.test.testing import assert_raises, assert_raises_message
|
|
from sqlalchemy import *
|
|
from sqlalchemy import exc as exceptions
|
|
from sqlalchemy.test import *
|
|
from sqlalchemy.engine import default
|
|
|
|
IDENT_LENGTH = 29
|
|
|
|
class LabelTypeTest(TestBase):
|
|
def test_type(self):
|
|
m = MetaData()
|
|
t = Table('sometable', m,
|
|
Column('col1', Integer),
|
|
Column('col2', Float))
|
|
assert isinstance(t.c.col1.label('hi').type, Integer)
|
|
assert isinstance(select([t.c.col2]).as_scalar().label('lala').type, Float)
|
|
|
|
class LongLabelsTest(TestBase, AssertsCompiledSQL):
|
|
@classmethod
|
|
def setup_class(cls):
|
|
global metadata, table1, table2, maxlen
|
|
metadata = MetaData(testing.db)
|
|
table1 = Table("some_large_named_table", metadata,
|
|
Column("this_is_the_primarykey_column", Integer, Sequence("this_is_some_large_seq"), primary_key=True),
|
|
Column("this_is_the_data_column", String(30))
|
|
)
|
|
|
|
table2 = Table("table_with_exactly_29_characs", metadata,
|
|
Column("this_is_the_primarykey_column", Integer, Sequence("some_seq"), primary_key=True),
|
|
Column("this_is_the_data_column", String(30))
|
|
)
|
|
|
|
metadata.create_all()
|
|
|
|
maxlen = testing.db.dialect.max_identifier_length
|
|
testing.db.dialect.max_identifier_length = IDENT_LENGTH
|
|
|
|
@engines.close_first
|
|
def teardown(self):
|
|
table1.delete().execute()
|
|
|
|
@classmethod
|
|
def teardown_class(cls):
|
|
metadata.drop_all()
|
|
testing.db.dialect.max_identifier_length = maxlen
|
|
|
|
def test_too_long_name_disallowed(self):
|
|
m = MetaData(testing.db)
|
|
t1 = Table("this_name_is_too_long_for_what_were_doing_in_this_test", m, Column('foo', Integer))
|
|
assert_raises(exceptions.IdentifierError, m.create_all)
|
|
assert_raises(exceptions.IdentifierError, m.drop_all)
|
|
assert_raises(exceptions.IdentifierError, t1.create)
|
|
assert_raises(exceptions.IdentifierError, t1.drop)
|
|
|
|
def test_result(self):
|
|
table1.insert().execute(**{"this_is_the_primarykey_column":1, "this_is_the_data_column":"data1"})
|
|
table1.insert().execute(**{"this_is_the_primarykey_column":2, "this_is_the_data_column":"data2"})
|
|
table1.insert().execute(**{"this_is_the_primarykey_column":3, "this_is_the_data_column":"data3"})
|
|
table1.insert().execute(**{"this_is_the_primarykey_column":4, "this_is_the_data_column":"data4"})
|
|
|
|
s = table1.select(use_labels=True, order_by=[table1.c.this_is_the_primarykey_column])
|
|
r = s.execute()
|
|
result = []
|
|
for row in r:
|
|
result.append((row[table1.c.this_is_the_primarykey_column], row[table1.c.this_is_the_data_column]))
|
|
assert result == [
|
|
(1, "data1"),
|
|
(2, "data2"),
|
|
(3, "data3"),
|
|
(4, "data4"),
|
|
], repr(result)
|
|
|
|
# some dialects such as oracle (and possibly ms-sql in a future version)
|
|
# generate a subquery for limits/offsets.
|
|
# ensure that the generated result map corresponds to the selected table, not
|
|
# the select query
|
|
r = s.limit(2).execute()
|
|
result = []
|
|
for row in r:
|
|
result.append((row[table1.c.this_is_the_primarykey_column], row[table1.c.this_is_the_data_column]))
|
|
assert result == [
|
|
(1, "data1"),
|
|
(2, "data2"),
|
|
], repr(result)
|
|
|
|
@testing.requires.offset
|
|
def go():
|
|
r = s.limit(2).offset(1).execute()
|
|
result = []
|
|
for row in r:
|
|
result.append((row[table1.c.this_is_the_primarykey_column], row[table1.c.this_is_the_data_column]))
|
|
assert result == [
|
|
(2, "data2"),
|
|
(3, "data3"),
|
|
], repr(result)
|
|
go()
|
|
|
|
def test_table_alias_names(self):
|
|
if testing.against('oracle'):
|
|
self.assert_compile(
|
|
table2.alias().select(),
|
|
"SELECT table_with_exactly_29_c_1.this_is_the_primarykey_column, table_with_exactly_29_c_1.this_is_the_data_column FROM table_with_exactly_29_characs table_with_exactly_29_c_1"
|
|
)
|
|
else:
|
|
self.assert_compile(
|
|
table2.alias().select(),
|
|
"SELECT table_with_exactly_29_c_1.this_is_the_primarykey_column, table_with_exactly_29_c_1.this_is_the_data_column FROM table_with_exactly_29_characs AS table_with_exactly_29_c_1"
|
|
)
|
|
|
|
ta = table2.alias()
|
|
dialect = default.DefaultDialect()
|
|
dialect.max_identifier_length = IDENT_LENGTH
|
|
self.assert_compile(
|
|
select([table1, ta]).select_from(table1.join(ta, table1.c.this_is_the_data_column==ta.c.this_is_the_data_column)).\
|
|
where(ta.c.this_is_the_data_column=='data3'),
|
|
|
|
"SELECT some_large_named_table.this_is_the_primarykey_column, some_large_named_table.this_is_the_data_column, "
|
|
"table_with_exactly_29_c_1.this_is_the_primarykey_column, table_with_exactly_29_c_1.this_is_the_data_column FROM "
|
|
"some_large_named_table JOIN table_with_exactly_29_characs AS table_with_exactly_29_c_1 ON "
|
|
"some_large_named_table.this_is_the_data_column = table_with_exactly_29_c_1.this_is_the_data_column "
|
|
"WHERE table_with_exactly_29_c_1.this_is_the_data_column = :this_is_the_data_column_1",
|
|
dialect=dialect
|
|
)
|
|
|
|
table2.insert().execute(
|
|
{"this_is_the_primarykey_column":1, "this_is_the_data_column":"data1"},
|
|
{"this_is_the_primarykey_column":2, "this_is_the_data_column":"data2"},
|
|
{"this_is_the_primarykey_column":3, "this_is_the_data_column":"data3"},
|
|
{"this_is_the_primarykey_column":4, "this_is_the_data_column":"data4"},
|
|
)
|
|
|
|
r = table2.alias().select().execute()
|
|
assert r.fetchall() == [(x, "data%d" % x) for x in range(1, 5)]
|
|
|
|
def test_colbinds(self):
|
|
table1.insert().execute(**{"this_is_the_primarykey_column":1, "this_is_the_data_column":"data1"})
|
|
table1.insert().execute(**{"this_is_the_primarykey_column":2, "this_is_the_data_column":"data2"})
|
|
table1.insert().execute(**{"this_is_the_primarykey_column":3, "this_is_the_data_column":"data3"})
|
|
table1.insert().execute(**{"this_is_the_primarykey_column":4, "this_is_the_data_column":"data4"})
|
|
|
|
r = table1.select(table1.c.this_is_the_primarykey_column == 4).execute()
|
|
assert r.fetchall() == [(4, "data4")]
|
|
|
|
r = table1.select(or_(
|
|
table1.c.this_is_the_primarykey_column == 4,
|
|
table1.c.this_is_the_primarykey_column == 2
|
|
)).execute()
|
|
assert r.fetchall() == [(2, "data2"), (4, "data4")]
|
|
|
|
def test_insert_no_pk(self):
|
|
table1.insert().execute(**{"this_is_the_data_column":"data1"})
|
|
table1.insert().execute(**{"this_is_the_data_column":"data2"})
|
|
table1.insert().execute(**{"this_is_the_data_column":"data3"})
|
|
table1.insert().execute(**{"this_is_the_data_column":"data4"})
|
|
|
|
@testing.requires.subqueries
|
|
def test_subquery(self):
|
|
q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias('foo')
|
|
x = select([q])
|
|
print x.execute().fetchall()
|
|
|
|
@testing.requires.subqueries
|
|
def test_anon_alias(self):
|
|
compile_dialect = default.DefaultDialect()
|
|
compile_dialect.max_identifier_length = IDENT_LENGTH
|
|
|
|
q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias()
|
|
x = select([q], use_labels=True)
|
|
|
|
self.assert_compile(x, "SELECT anon_1.this_is_the_primarykey_column AS anon_1_this_is_the_prim_1, anon_1.this_is_the_data_column AS anon_1_this_is_the_data_2 "
|
|
"FROM (SELECT some_large_named_table.this_is_the_primarykey_column AS this_is_the_primarykey_column, some_large_named_table.this_is_the_data_column AS this_is_the_data_column "
|
|
"FROM some_large_named_table "
|
|
"WHERE some_large_named_table.this_is_the_primarykey_column = :this_is_the_primarykey__1) AS anon_1", dialect=compile_dialect)
|
|
|
|
print x.execute().fetchall()
|
|
|
|
def test_adjustable(self):
|
|
|
|
q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias('foo')
|
|
x = select([q])
|
|
|
|
compile_dialect = default.DefaultDialect(label_length=10)
|
|
self.assert_compile(x, "SELECT foo.this_is_the_primarykey_column, foo.this_is_the_data_column FROM "
|
|
"(SELECT some_large_named_table.this_is_the_primarykey_column AS this_1, some_large_named_table.this_is_the_data_column "
|
|
"AS this_2 FROM some_large_named_table WHERE some_large_named_table.this_is_the_primarykey_column = :this_1) AS foo", dialect=compile_dialect)
|
|
|
|
compile_dialect = default.DefaultDialect(label_length=4)
|
|
self.assert_compile(x, "SELECT foo.this_is_the_primarykey_column, foo.this_is_the_data_column FROM "
|
|
"(SELECT some_large_named_table.this_is_the_primarykey_column AS _1, some_large_named_table.this_is_the_data_column AS _2 "
|
|
"FROM some_large_named_table WHERE some_large_named_table.this_is_the_primarykey_column = :_1) AS foo", dialect=compile_dialect)
|
|
|
|
q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias()
|
|
x = select([q], use_labels=True)
|
|
|
|
compile_dialect = default.DefaultDialect(label_length=10)
|
|
self.assert_compile(x, "SELECT anon_1.this_is_the_primarykey_column AS anon_1, anon_1.this_is_the_data_column AS anon_2 FROM "
|
|
"(SELECT some_large_named_table.this_is_the_primarykey_column AS this_3, some_large_named_table.this_is_the_data_column AS this_4 "
|
|
"FROM some_large_named_table WHERE some_large_named_table.this_is_the_primarykey_column = :this_1) AS anon_1", dialect=compile_dialect)
|
|
|
|
compile_dialect = default.DefaultDialect(label_length=4)
|
|
self.assert_compile(x, "SELECT _1.this_is_the_primarykey_column AS _1, _1.this_is_the_data_column AS _2 FROM "
|
|
"(SELECT some_large_named_table.this_is_the_primarykey_column AS _3, some_large_named_table.this_is_the_data_column AS _4 "
|
|
"FROM some_large_named_table WHERE some_large_named_table.this_is_the_primarykey_column = :_1) AS _1", dialect=compile_dialect)
|
|
|
|
|