1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
|
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)
|
