mirror of
https://github.com/sqlalchemy/sqlalchemy.git
synced 2026-05-24 09:31:48 -04:00
Mike Bayer
5188afe870
Added new :func:`.create_engine` parameter :paramref:`.create_engine.max_identifier_length`. This overrides the dialect-coded "max identifier length" in order to accommodate for databases that have recently changed this length and the SQLAlchemy dialect has not yet been adjusted to detect for that version. This parameter interacts with the existing :paramref:`.create_engine.label_length` parameter in that it establishes the maximum (and default) value for anonymously generated labels. The Oracle dialect now emits a warning if Oracle version 12.2 or greater is used, and the :paramref:`.create_engine.max_identifier_length` parameter is not set. The version in this specific case defaults to that of the "compatibility" version set in the Oracle server configuration, not the actual server version. In version 1.4, the default max_identifier_length for 12.2 or greater will move to 128 characters. In order to maintain forwards compatibility, applications should set :paramref:`.create_engine.max_identifier_length` to 30 in order to maintain the same length behavior, or to 128 in order to test the upcoming behavior. This length determines among other things how generated constraint names are truncated for statements like ``CREATE CONSTRAINT`` and ``DROP CONSTRAINT``, which means a the new length may produce a name-mismatch against a name that was generated with the old length, impacting database migrations. Fixes: #4857 Change-Id: Ib62efb00c6180c375869029b57353d90385d7950 (cherry picked from commit 88761b8b0b0cfa67cdd6a4913e3a0ea5cba93cbb)
753 lines
25 KiB
Python
753 lines
25 KiB
Python
from sqlalchemy import bindparam
|
|
from sqlalchemy import exc as exceptions
|
|
from sqlalchemy import Integer
|
|
from sqlalchemy import MetaData
|
|
from sqlalchemy import or_
|
|
from sqlalchemy import select
|
|
from sqlalchemy import testing
|
|
from sqlalchemy.engine import default
|
|
from sqlalchemy.sql import column
|
|
from sqlalchemy.sql import table
|
|
from sqlalchemy.sql.elements import _truncated_label
|
|
from sqlalchemy.testing import assert_raises
|
|
from sqlalchemy.testing import assert_raises_message
|
|
from sqlalchemy.testing import AssertsCompiledSQL
|
|
from sqlalchemy.testing import engines
|
|
from sqlalchemy.testing import eq_
|
|
from sqlalchemy.testing import fixtures
|
|
from sqlalchemy.testing import mock
|
|
from sqlalchemy.testing.schema import Column
|
|
from sqlalchemy.testing.schema import Table
|
|
|
|
IDENT_LENGTH = 29
|
|
|
|
|
|
class MaxIdentTest(fixtures.TestBase, AssertsCompiledSQL):
|
|
__dialect__ = "DefaultDialect"
|
|
|
|
table1 = table(
|
|
"some_large_named_table",
|
|
column("this_is_the_primarykey_column"),
|
|
column("this_is_the_data_column"),
|
|
)
|
|
|
|
table2 = table(
|
|
"table_with_exactly_29_characs",
|
|
column("this_is_the_primarykey_column"),
|
|
column("this_is_the_data_column"),
|
|
)
|
|
|
|
def _length_fixture(self, length=IDENT_LENGTH, positional=False):
|
|
dialect = default.DefaultDialect()
|
|
dialect.max_identifier_length = (
|
|
dialect._user_defined_max_identifier_length
|
|
) = length
|
|
if positional:
|
|
dialect.paramstyle = "format"
|
|
dialect.positional = True
|
|
return dialect
|
|
|
|
def _engine_fixture(self, length=IDENT_LENGTH):
|
|
eng = engines.testing_engine()
|
|
eng.dialect.max_identifier_length = (
|
|
eng.dialect._user_defined_max_identifier_length
|
|
) = length
|
|
return eng
|
|
|
|
def test_label_length_raise_too_large(self):
|
|
max_ident_length = testing.db.dialect.max_identifier_length
|
|
eng = engines.testing_engine(
|
|
options={"label_length": max_ident_length + 10}
|
|
)
|
|
assert_raises_message(
|
|
exceptions.ArgumentError,
|
|
"Label length of %d is greater than this dialect's maximum "
|
|
"identifier length of %d"
|
|
% (max_ident_length + 10, max_ident_length),
|
|
eng.connect,
|
|
)
|
|
|
|
def test_label_length_custom_maxlen(self):
|
|
max_ident_length = testing.db.dialect.max_identifier_length
|
|
eng = engines.testing_engine(
|
|
options={
|
|
"label_length": max_ident_length + 10,
|
|
"max_identifier_length": max_ident_length + 20,
|
|
}
|
|
)
|
|
with eng.connect() as conn:
|
|
eq_(conn.dialect.max_identifier_length, max_ident_length + 20)
|
|
|
|
def test_label_length_custom_maxlen_dialect_only(self):
|
|
dialect = default.DefaultDialect(max_identifier_length=47)
|
|
eq_(dialect.max_identifier_length, 47)
|
|
|
|
def test_label_length_custom_maxlen_user_set_manually(self):
|
|
eng = engines.testing_engine()
|
|
eng.dialect.max_identifier_length = 47
|
|
|
|
# assume the dialect has no on-connect change
|
|
with mock.patch.object(
|
|
eng.dialect,
|
|
"_check_max_identifier_length",
|
|
side_effect=lambda conn: None,
|
|
):
|
|
with eng.connect():
|
|
pass
|
|
|
|
# it was maintained
|
|
eq_(eng.dialect.max_identifier_length, 47)
|
|
|
|
def test_label_length_too_large_custom_maxlen(self):
|
|
max_ident_length = testing.db.dialect.max_identifier_length
|
|
eng = engines.testing_engine(
|
|
options={
|
|
"label_length": max_ident_length - 10,
|
|
"max_identifier_length": max_ident_length - 20,
|
|
}
|
|
)
|
|
assert_raises_message(
|
|
exceptions.ArgumentError,
|
|
"Label length of %d is greater than this dialect's maximum "
|
|
"identifier length of %d"
|
|
% (max_ident_length - 10, max_ident_length - 20),
|
|
eng.connect,
|
|
)
|
|
|
|
def test_custom_max_identifier_length(self):
|
|
max_ident_length = testing.db.dialect.max_identifier_length
|
|
eng = engines.testing_engine(
|
|
options={"max_identifier_length": max_ident_length + 20}
|
|
)
|
|
with eng.connect() as conn:
|
|
eq_(conn.dialect.max_identifier_length, max_ident_length + 20)
|
|
|
|
def test_max_identifier_length_onconnect(self):
|
|
eng = engines.testing_engine()
|
|
|
|
def _check_max_identifer_length(conn):
|
|
return 47
|
|
|
|
with mock.patch.object(
|
|
eng.dialect,
|
|
"_check_max_identifier_length",
|
|
side_effect=_check_max_identifer_length,
|
|
) as mock_:
|
|
with eng.connect():
|
|
eq_(eng.dialect.max_identifier_length, 47)
|
|
eq_(mock_.mock_calls, [mock.call(mock.ANY)])
|
|
|
|
def test_max_identifier_length_onconnect_returns_none(self):
|
|
eng = engines.testing_engine()
|
|
|
|
max_ident_length = eng.dialect.max_identifier_length
|
|
|
|
def _check_max_identifer_length(conn):
|
|
return None
|
|
|
|
with mock.patch.object(
|
|
eng.dialect,
|
|
"_check_max_identifier_length",
|
|
side_effect=_check_max_identifer_length,
|
|
) as mock_:
|
|
with eng.connect():
|
|
eq_(eng.dialect.max_identifier_length, max_ident_length)
|
|
eq_(mock_.mock_calls, [mock.call(mock.ANY)])
|
|
|
|
def test_custom_max_identifier_length_onconnect(self):
|
|
eng = engines.testing_engine(options={"max_identifier_length": 49})
|
|
|
|
def _check_max_identifer_length(conn):
|
|
return 47
|
|
|
|
with mock.patch.object(
|
|
eng.dialect,
|
|
"_check_max_identifier_length",
|
|
side_effect=_check_max_identifer_length,
|
|
) as mock_:
|
|
with eng.connect():
|
|
eq_(eng.dialect.max_identifier_length, 49)
|
|
eq_(mock_.mock_calls, []) # was not called
|
|
|
|
def test_table_alias_1(self):
|
|
self.assert_compile(
|
|
self.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",
|
|
dialect=self._length_fixture(),
|
|
)
|
|
|
|
def test_table_alias_2(self):
|
|
table1 = self.table1
|
|
table2 = self.table2
|
|
ta = table2.alias()
|
|
on = table1.c.this_is_the_data_column == ta.c.this_is_the_data_column
|
|
self.assert_compile(
|
|
select([table1, ta])
|
|
.select_from(table1.join(ta, on))
|
|
.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=self._length_fixture(),
|
|
)
|
|
|
|
def test_too_long_name_disallowed(self):
|
|
m = MetaData()
|
|
t = Table(
|
|
"this_name_is_too_long_for_what_were_doing_in_this_test",
|
|
m,
|
|
Column("foo", Integer),
|
|
)
|
|
eng = self._engine_fixture()
|
|
methods = (t.create, t.drop, m.create_all, m.drop_all)
|
|
for meth in methods:
|
|
assert_raises(exceptions.IdentifierError, meth, eng)
|
|
|
|
def _assert_labeled_table1_select(self, s):
|
|
table1 = self.table1
|
|
compiled = s.compile(dialect=self._length_fixture())
|
|
|
|
assert set(
|
|
compiled._create_result_map()["some_large_named_table__2"][1]
|
|
).issuperset(
|
|
[
|
|
"some_large_named_table_this_is_the_data_column",
|
|
"some_large_named_table__2",
|
|
table1.c.this_is_the_data_column,
|
|
]
|
|
)
|
|
|
|
assert set(
|
|
compiled._create_result_map()["some_large_named_table__1"][1]
|
|
).issuperset(
|
|
[
|
|
"some_large_named_table_this_is_the_primarykey_column",
|
|
"some_large_named_table__1",
|
|
table1.c.this_is_the_primarykey_column,
|
|
]
|
|
)
|
|
|
|
def test_result_map_use_labels(self):
|
|
table1 = self.table1
|
|
s = (
|
|
table1.select()
|
|
.apply_labels()
|
|
.order_by(table1.c.this_is_the_primarykey_column)
|
|
)
|
|
|
|
self._assert_labeled_table1_select(s)
|
|
|
|
def test_result_map_limit(self):
|
|
table1 = self.table1
|
|
# 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
|
|
s = table1.select(
|
|
use_labels=True, order_by=[table1.c.this_is_the_primarykey_column]
|
|
).limit(2)
|
|
self._assert_labeled_table1_select(s)
|
|
|
|
def test_result_map_subquery(self):
|
|
table1 = self.table1
|
|
s = table1.select(table1.c.this_is_the_primarykey_column == 4).alias(
|
|
"foo"
|
|
)
|
|
s2 = select([s])
|
|
compiled = s2.compile(dialect=self._length_fixture())
|
|
assert set(
|
|
compiled._create_result_map()["this_is_the_data_column"][1]
|
|
).issuperset(["this_is_the_data_column", s.c.this_is_the_data_column])
|
|
assert set(
|
|
compiled._create_result_map()["this_is_the_primarykey__1"][1]
|
|
).issuperset(
|
|
[
|
|
"this_is_the_primarykey_column",
|
|
"this_is_the_primarykey__1",
|
|
s.c.this_is_the_primarykey_column,
|
|
]
|
|
)
|
|
|
|
def test_result_map_anon_alias(self):
|
|
table1 = self.table1
|
|
dialect = self._length_fixture()
|
|
|
|
q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias()
|
|
s = select([q]).apply_labels()
|
|
|
|
self.assert_compile(
|
|
s,
|
|
"SELECT "
|
|
"anon_1.this_is_the_primarykey__2 AS anon_1_this_is_the_prim_1, "
|
|
"anon_1.this_is_the_data_column AS anon_1_this_is_the_data_3 "
|
|
"FROM ("
|
|
"SELECT "
|
|
"some_large_named_table."
|
|
"this_is_the_primarykey_column AS this_is_the_primarykey__2, "
|
|
"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=dialect,
|
|
)
|
|
|
|
compiled = s.compile(dialect=dialect)
|
|
assert set(
|
|
compiled._create_result_map()["anon_1_this_is_the_data_3"][1]
|
|
).issuperset(
|
|
[
|
|
"anon_1_this_is_the_data_3",
|
|
q.corresponding_column(table1.c.this_is_the_data_column),
|
|
]
|
|
)
|
|
|
|
assert set(
|
|
compiled._create_result_map()["anon_1_this_is_the_prim_1"][1]
|
|
).issuperset(
|
|
[
|
|
"anon_1_this_is_the_prim_1",
|
|
q.corresponding_column(table1.c.this_is_the_primarykey_column),
|
|
]
|
|
)
|
|
|
|
def test_column_bind_labels_1(self):
|
|
table1 = self.table1
|
|
|
|
s = table1.select(table1.c.this_is_the_primarykey_column == 4)
|
|
self.assert_compile(
|
|
s,
|
|
"SELECT some_large_named_table.this_is_the_primarykey_column, "
|
|
"some_large_named_table.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",
|
|
checkparams={"this_is_the_primarykey__1": 4},
|
|
dialect=self._length_fixture(),
|
|
)
|
|
|
|
self.assert_compile(
|
|
s,
|
|
"SELECT some_large_named_table.this_is_the_primarykey_column, "
|
|
"some_large_named_table.this_is_the_data_column "
|
|
"FROM some_large_named_table WHERE "
|
|
"some_large_named_table.this_is_the_primarykey_column = "
|
|
"%s",
|
|
checkpositional=(4,),
|
|
checkparams={"this_is_the_primarykey__1": 4},
|
|
dialect=self._length_fixture(positional=True),
|
|
)
|
|
|
|
def test_column_bind_labels_2(self):
|
|
table1 = self.table1
|
|
|
|
s = table1.select(
|
|
or_(
|
|
table1.c.this_is_the_primarykey_column == 4,
|
|
table1.c.this_is_the_primarykey_column == 2,
|
|
)
|
|
)
|
|
self.assert_compile(
|
|
s,
|
|
"SELECT some_large_named_table.this_is_the_primarykey_column, "
|
|
"some_large_named_table.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 OR "
|
|
"some_large_named_table.this_is_the_primarykey_column = "
|
|
":this_is_the_primarykey__2",
|
|
checkparams={
|
|
"this_is_the_primarykey__1": 4,
|
|
"this_is_the_primarykey__2": 2,
|
|
},
|
|
dialect=self._length_fixture(),
|
|
)
|
|
self.assert_compile(
|
|
s,
|
|
"SELECT some_large_named_table.this_is_the_primarykey_column, "
|
|
"some_large_named_table.this_is_the_data_column "
|
|
"FROM some_large_named_table WHERE "
|
|
"some_large_named_table.this_is_the_primarykey_column = "
|
|
"%s OR "
|
|
"some_large_named_table.this_is_the_primarykey_column = "
|
|
"%s",
|
|
checkparams={
|
|
"this_is_the_primarykey__1": 4,
|
|
"this_is_the_primarykey__2": 2,
|
|
},
|
|
checkpositional=(4, 2),
|
|
dialect=self._length_fixture(positional=True),
|
|
)
|
|
|
|
def test_bind_param_non_truncated(self):
|
|
table1 = self.table1
|
|
stmt = table1.insert().values(
|
|
this_is_the_data_column=bindparam(
|
|
"this_is_the_long_bindparam_name"
|
|
)
|
|
)
|
|
compiled = stmt.compile(dialect=self._length_fixture(length=10))
|
|
eq_(
|
|
compiled.construct_params(
|
|
params={"this_is_the_long_bindparam_name": 5}
|
|
),
|
|
{"this_is_the_long_bindparam_name": 5},
|
|
)
|
|
|
|
def test_bind_param_truncated_named(self):
|
|
table1 = self.table1
|
|
bp = bindparam(_truncated_label("this_is_the_long_bindparam_name"))
|
|
stmt = table1.insert().values(this_is_the_data_column=bp)
|
|
compiled = stmt.compile(dialect=self._length_fixture(length=10))
|
|
eq_(
|
|
compiled.construct_params(
|
|
params={"this_is_the_long_bindparam_name": 5}
|
|
),
|
|
{"this_1": 5},
|
|
)
|
|
|
|
def test_bind_param_truncated_positional(self):
|
|
table1 = self.table1
|
|
bp = bindparam(_truncated_label("this_is_the_long_bindparam_name"))
|
|
stmt = table1.insert().values(this_is_the_data_column=bp)
|
|
compiled = stmt.compile(
|
|
dialect=self._length_fixture(length=10, positional=True)
|
|
)
|
|
|
|
eq_(
|
|
compiled.construct_params(
|
|
params={"this_is_the_long_bindparam_name": 5}
|
|
),
|
|
{"this_1": 5},
|
|
)
|
|
|
|
|
|
class LabelLengthTest(fixtures.TestBase, AssertsCompiledSQL):
|
|
__dialect__ = "DefaultDialect"
|
|
|
|
table1 = table(
|
|
"some_large_named_table",
|
|
column("this_is_the_primarykey_column"),
|
|
column("this_is_the_data_column"),
|
|
)
|
|
|
|
table2 = table(
|
|
"table_with_exactly_29_characs",
|
|
column("this_is_the_primarykey_column"),
|
|
column("this_is_the_data_column"),
|
|
)
|
|
|
|
def test_adjustable_1(self):
|
|
table1 = self.table1
|
|
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_1, foo.this_2 "
|
|
"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,
|
|
)
|
|
|
|
def test_adjustable_2(self):
|
|
table1 = self.table1
|
|
|
|
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_1, foo.this_2 "
|
|
"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,
|
|
)
|
|
|
|
def test_adjustable_3(self):
|
|
table1 = self.table1
|
|
|
|
compile_dialect = default.DefaultDialect(label_length=4)
|
|
q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias(
|
|
"foo"
|
|
)
|
|
x = select([q])
|
|
|
|
self.assert_compile(
|
|
x,
|
|
"SELECT "
|
|
"foo._1, foo._2 "
|
|
"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,
|
|
)
|
|
|
|
def test_adjustable_4(self):
|
|
table1 = self.table1
|
|
|
|
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_2 AS anon_1, "
|
|
"anon_1.this_4 AS anon_3 "
|
|
"FROM ("
|
|
"SELECT "
|
|
"some_large_named_table.this_is_the_primarykey_column "
|
|
"AS this_2, "
|
|
"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,
|
|
)
|
|
|
|
def test_adjustable_5(self):
|
|
table1 = self.table1
|
|
q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias()
|
|
x = select([q], use_labels=True)
|
|
|
|
compile_dialect = default.DefaultDialect(label_length=4)
|
|
self.assert_compile(
|
|
x,
|
|
"SELECT "
|
|
"_1._2 AS _1, "
|
|
"_1._4 AS _3 "
|
|
"FROM ("
|
|
"SELECT "
|
|
"some_large_named_table.this_is_the_primarykey_column "
|
|
"AS _2, "
|
|
"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,
|
|
)
|
|
|
|
def test_adjustable_result_schema_column_1(self):
|
|
table1 = self.table1
|
|
|
|
q = (
|
|
table1.select(table1.c.this_is_the_primarykey_column == 4)
|
|
.apply_labels()
|
|
.alias("foo")
|
|
)
|
|
|
|
dialect = default.DefaultDialect(label_length=10)
|
|
compiled = q.compile(dialect=dialect)
|
|
|
|
assert set(compiled._create_result_map()["some_2"][1]).issuperset(
|
|
[
|
|
table1.c.this_is_the_data_column,
|
|
"some_large_named_table_this_is_the_data_column",
|
|
"some_2",
|
|
]
|
|
)
|
|
|
|
assert set(compiled._create_result_map()["some_1"][1]).issuperset(
|
|
[
|
|
table1.c.this_is_the_primarykey_column,
|
|
"some_large_named_table_this_is_the_primarykey_column",
|
|
"some_1",
|
|
]
|
|
)
|
|
|
|
def test_adjustable_result_schema_column_2(self):
|
|
table1 = self.table1
|
|
|
|
q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias(
|
|
"foo"
|
|
)
|
|
x = select([q])
|
|
|
|
dialect = default.DefaultDialect(label_length=10)
|
|
compiled = x.compile(dialect=dialect)
|
|
|
|
assert set(compiled._create_result_map()["this_2"][1]).issuperset(
|
|
[
|
|
q.corresponding_column(table1.c.this_is_the_data_column),
|
|
"this_is_the_data_column",
|
|
"this_2",
|
|
]
|
|
)
|
|
|
|
assert set(compiled._create_result_map()["this_1"][1]).issuperset(
|
|
[
|
|
q.corresponding_column(table1.c.this_is_the_primarykey_column),
|
|
"this_is_the_primarykey_column",
|
|
"this_1",
|
|
]
|
|
)
|
|
|
|
def test_table_plus_column_exceeds_length(self):
|
|
"""test that the truncation only occurs when tablename + colname are
|
|
concatenated, if they are individually under the label length.
|
|
|
|
"""
|
|
|
|
compile_dialect = default.DefaultDialect(label_length=30)
|
|
a_table = table("thirty_characters_table_xxxxxx", column("id"))
|
|
|
|
other_table = table(
|
|
"other_thirty_characters_table_",
|
|
column("id"),
|
|
column("thirty_characters_table_id"),
|
|
)
|
|
|
|
anon = a_table.alias()
|
|
|
|
j1 = other_table.outerjoin(
|
|
anon, anon.c.id == other_table.c.thirty_characters_table_id
|
|
)
|
|
|
|
self.assert_compile(
|
|
select([other_table, anon]).select_from(j1).apply_labels(),
|
|
"SELECT "
|
|
"other_thirty_characters_table_.id "
|
|
"AS other_thirty_characters__1, "
|
|
"other_thirty_characters_table_.thirty_characters_table_id "
|
|
"AS other_thirty_characters__2, "
|
|
"thirty_characters_table__1.id "
|
|
"AS thirty_characters_table__3 "
|
|
"FROM "
|
|
"other_thirty_characters_table_ "
|
|
"LEFT OUTER JOIN "
|
|
"thirty_characters_table_xxxxxx AS thirty_characters_table__1 "
|
|
"ON thirty_characters_table__1.id = "
|
|
"other_thirty_characters_table_.thirty_characters_table_id",
|
|
dialect=compile_dialect,
|
|
)
|
|
|
|
def test_colnames_longer_than_labels_lowercase(self):
|
|
t1 = table("a", column("abcde"))
|
|
self._test_colnames_longer_than_labels(t1)
|
|
|
|
def test_colnames_longer_than_labels_uppercase(self):
|
|
m = MetaData()
|
|
t1 = Table("a", m, Column("abcde", Integer))
|
|
self._test_colnames_longer_than_labels(t1)
|
|
|
|
def _test_colnames_longer_than_labels(self, t1):
|
|
dialect = default.DefaultDialect(label_length=4)
|
|
a1 = t1.alias(name="asdf")
|
|
|
|
# 'abcde' is longer than 4, but rendered as itself
|
|
# needs to have all characters
|
|
s = select([a1])
|
|
self.assert_compile(
|
|
select([a1]), "SELECT asdf.abcde FROM a AS asdf", dialect=dialect
|
|
)
|
|
compiled = s.compile(dialect=dialect)
|
|
assert set(compiled._create_result_map()["abcde"][1]).issuperset(
|
|
["abcde", a1.c.abcde, "abcde"]
|
|
)
|
|
|
|
# column still there, but short label
|
|
s = select([a1]).apply_labels()
|
|
self.assert_compile(
|
|
s, "SELECT asdf.abcde AS _1 FROM a AS asdf", dialect=dialect
|
|
)
|
|
compiled = s.compile(dialect=dialect)
|
|
assert set(compiled._create_result_map()["_1"][1]).issuperset(
|
|
["asdf_abcde", a1.c.abcde, "_1"]
|
|
)
|
|
|
|
def test_label_overlap_unlabeled(self):
|
|
"""test that an anon col can't overlap with a fixed name, #3396"""
|
|
|
|
table1 = table(
|
|
"tablename", column("columnname_one"), column("columnn_1")
|
|
)
|
|
|
|
stmt = select([table1]).apply_labels()
|
|
|
|
dialect = default.DefaultDialect(label_length=23)
|
|
self.assert_compile(
|
|
stmt,
|
|
"SELECT tablename.columnname_one AS tablename_columnn_1, "
|
|
"tablename.columnn_1 AS tablename_columnn_2 FROM tablename",
|
|
dialect=dialect,
|
|
)
|
|
compiled = stmt.compile(dialect=dialect)
|
|
eq_(
|
|
set(compiled._create_result_map()),
|
|
set(["tablename_columnn_1", "tablename_columnn_2"]),
|
|
)
|