summaryrefslogtreecommitdiff
path: root/lib
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2021-01-30 16:57:50 -0500
committerMike Bayer <mike_mp@zzzcomputing.com>2021-01-30 23:12:31 -0500
commit2db1eccb382c3be2c03e5be80c6f68e41a8545a2 (patch)
tree531462b017eb25d5ba2eff1672f0a0ef572c9515 /lib
parent532026f97f402d6673cd9746f1a7daee99327a68 (diff)
downloadsqlalchemy-2db1eccb382c3be2c03e5be80c6f68e41a8545a2.tar.gz
set identifier length for MySQL constraints to 64
The rule to limit index names to 64 also applies to all DDL names, such as those coming from naming conventions. Add another limiting variable for constraint names and create test cases against all constraint types. Additionally, codified in the test suite MySQL's lack of support for naming of a FOREIGN KEY constraint after the name was given, which apparently assigns the name to an associated KEY but not the constraint itself, until MySQL 8 and MariaDB 10.5 which appear to have resolved the behavior. However it's not clear how Alembic hasn't had issues reported with this so far. Fixed long-lived bug in MySQL dialect where the maximum identifier length of 255 was too long for names of all types of constraints, not just indexes, all of which have a size limit of 64. As metadata naming conventions can create too-long names in this area, apply the limit to the identifier generator within the DDL compiler. Fixes: #5898 Change-Id: I79549474845dc29922275cf13321c07598dcea08
Diffstat (limited to 'lib')
-rw-r--r--lib/sqlalchemy/dialects/mysql/base.py1
-rw-r--r--lib/sqlalchemy/engine/default.py9
-rw-r--r--lib/sqlalchemy/sql/compiler.py8
-rw-r--r--lib/sqlalchemy/testing/exclusions.py7
-rw-r--r--lib/sqlalchemy/testing/requirements.py12
-rw-r--r--lib/sqlalchemy/testing/suite/test_ddl.py200
6 files changed, 229 insertions, 8 deletions
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index c35031613..063f750fa 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -2513,6 +2513,7 @@ class MySQLDialect(default.DefaultDialect):
# identifiers are 64, however aliases can be 255...
max_identifier_length = 255
max_index_name_length = 64
+ max_constraint_name_length = 64
supports_native_enum = True
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index e5a6384d2..7fddf2814 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -127,12 +127,11 @@ class DefaultDialect(interfaces.Dialect):
isolation_level = None
- # length at which to truncate
- # the name of an index.
- # Usually None to indicate
- # 'use max_identifier_length'.
- # thanks to MySQL, sigh
+ # sub-categories of max_identifier_length.
+ # currently these accommodate for MySQL which allows alias names
+ # of 255 but DDL names only of 64.
max_index_name_length = None
+ max_constraint_name_length = None
supports_sane_rowcount = True
supports_sane_multi_rowcount = True
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 353de2c48..3925b251d 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -4793,13 +4793,19 @@ class IdentifierPreparer(object):
name = constraint.name
if isinstance(name, elements._truncated_label):
+ # calculate these at format time so that ad-hoc changes
+ # to dialect.max_identifier_length etc. can be reflected
+ # as IdentifierPreparer is long lived
if constraint.__visit_name__ == "index":
max_ = (
self.dialect.max_index_name_length
or self.dialect.max_identifier_length
)
else:
- max_ = self.dialect.max_identifier_length
+ max_ = (
+ self.dialect.max_constraint_name_length
+ or self.dialect.max_identifier_length
+ )
if len(name) > max_:
name = name[0 : max_ - 8] + "_" + util.md5_hex(name)[-4:]
else:
diff --git a/lib/sqlalchemy/testing/exclusions.py b/lib/sqlalchemy/testing/exclusions.py
index b20e17442..6502487f0 100644
--- a/lib/sqlalchemy/testing/exclusions.py
+++ b/lib/sqlalchemy/testing/exclusions.py
@@ -40,6 +40,13 @@ class compound(object):
def __add__(self, other):
return self.add(other)
+ def as_skips(self):
+ rule = compound()
+ rule.skips.update(self.skips)
+ rule.skips.update(self.fails)
+ rule.tags.update(self.tags)
+ return rule
+
def add(self, *others):
copy = compound()
copy.fails.update(self.fails)
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index e5dda7d3e..f5286c85d 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -460,6 +460,18 @@ class SuiteRequirements(Requirements):
return exclusions.closed()
@property
+ def foreign_key_constraint_name_reflection(self):
+ """Target supports refleciton of FOREIGN KEY constraints and
+ will return the name of the constraint that was used in the
+ "CONSTRANT <name> FOREIGN KEY" DDL.
+
+ MySQL prior to version 8 and MariaDB prior to version 10.5
+ don't support this.
+
+ """
+ return exclusions.closed()
+
+ @property
def implicit_default_schema(self):
"""target system has a strong concept of 'default' schema that can
be referred to implicitly.
diff --git a/lib/sqlalchemy/testing/suite/test_ddl.py b/lib/sqlalchemy/testing/suite/test_ddl.py
index c3cf854e4..b3fee551e 100644
--- a/lib/sqlalchemy/testing/suite/test_ddl.py
+++ b/lib/sqlalchemy/testing/suite/test_ddl.py
@@ -1,3 +1,6 @@
+import random
+
+from . import testing
from .. import config
from .. import fixtures
from .. import util
@@ -5,13 +8,16 @@ from ..assertions import eq_
from ..assertions import is_false
from ..assertions import is_true
from ..config import requirements
+from ..schema import Table
+from ... import CheckConstraint
from ... import Column
+from ... import ForeignKeyConstraint
from ... import Index
from ... import inspect
from ... import Integer
from ... import schema
from ... import String
-from ... import Table
+from ... import UniqueConstraint
class TableDDLTest(fixtures.TestBase):
@@ -182,4 +188,194 @@ class FutureTableDDLTest(fixtures.FutureEngineMixin, TableDDLTest):
pass
-__all__ = ("TableDDLTest", "FutureTableDDLTest")
+class LongNameBlowoutTest(fixtures.TestBase):
+ """test the creation of a variety of DDL structures and ensure
+ label length limits pass on backends
+
+ """
+
+ __backend__ = True
+
+ def fk(self, metadata, connection):
+ convention = {
+ "fk": "foreign_key_%(table_name)s_"
+ "%(column_0_N_name)s_"
+ "%(referred_table_name)s_"
+ + (
+ "_".join(
+ "".join(random.choice("abcdef") for j in range(20))
+ for i in range(10)
+ )
+ ),
+ }
+ metadata.naming_convention = convention
+
+ Table(
+ "a_things_with_stuff",
+ metadata,
+ Column("id_long_column_name", Integer, primary_key=True),
+ test_needs_fk=True,
+ )
+
+ cons = ForeignKeyConstraint(
+ ["aid"], ["a_things_with_stuff.id_long_column_name"]
+ )
+ Table(
+ "b_related_things_of_value",
+ metadata,
+ Column(
+ "aid",
+ ),
+ cons,
+ test_needs_fk=True,
+ )
+ actual_name = cons.name
+
+ metadata.create_all(connection)
+
+ if testing.requires.foreign_key_constraint_name_reflection.enabled:
+ insp = inspect(connection)
+ fks = insp.get_foreign_keys("b_related_things_of_value")
+ reflected_name = fks[0]["name"]
+
+ return actual_name, reflected_name
+ else:
+ return actual_name, None
+
+ def pk(self, metadata, connection):
+ convention = {
+ "pk": "primary_key_%(table_name)s_"
+ "%(column_0_N_name)s"
+ + (
+ "_".join(
+ "".join(random.choice("abcdef") for j in range(30))
+ for i in range(10)
+ )
+ ),
+ }
+ metadata.naming_convention = convention
+
+ a = Table(
+ "a_things_with_stuff",
+ metadata,
+ Column("id_long_column_name", Integer, primary_key=True),
+ Column("id_another_long_name", Integer, primary_key=True),
+ )
+ cons = a.primary_key
+ actual_name = cons.name
+
+ metadata.create_all(connection)
+ insp = inspect(connection)
+ pk = insp.get_pk_constraint("a_things_with_stuff")
+ reflected_name = pk["name"]
+ return actual_name, reflected_name
+
+ def ix(self, metadata, connection):
+ convention = {
+ "ix": "index_%(table_name)s_"
+ "%(column_0_N_name)s"
+ + (
+ "_".join(
+ "".join(random.choice("abcdef") for j in range(30))
+ for i in range(10)
+ )
+ ),
+ }
+ metadata.naming_convention = convention
+
+ a = Table(
+ "a_things_with_stuff",
+ metadata,
+ Column("id_long_column_name", Integer, primary_key=True),
+ Column("id_another_long_name", Integer),
+ )
+ cons = Index(None, a.c.id_long_column_name, a.c.id_another_long_name)
+ actual_name = cons.name
+
+ metadata.create_all(connection)
+ insp = inspect(connection)
+ ix = insp.get_indexes("a_things_with_stuff")
+ reflected_name = ix[0]["name"]
+ return actual_name, reflected_name
+
+ def uq(self, metadata, connection):
+ convention = {
+ "uq": "unique_constraint_%(table_name)s_"
+ "%(column_0_N_name)s"
+ + (
+ "_".join(
+ "".join(random.choice("abcdef") for j in range(30))
+ for i in range(10)
+ )
+ ),
+ }
+ metadata.naming_convention = convention
+
+ cons = UniqueConstraint("id_long_column_name", "id_another_long_name")
+ Table(
+ "a_things_with_stuff",
+ metadata,
+ Column("id_long_column_name", Integer, primary_key=True),
+ Column("id_another_long_name", Integer),
+ cons,
+ )
+ actual_name = cons.name
+
+ metadata.create_all(connection)
+ insp = inspect(connection)
+ uq = insp.get_unique_constraints("a_things_with_stuff")
+ reflected_name = uq[0]["name"]
+ return actual_name, reflected_name
+
+ def ck(self, metadata, connection):
+ convention = {
+ "ck": "check_constraint_%(table_name)s"
+ + (
+ "_".join(
+ "".join(random.choice("abcdef") for j in range(30))
+ for i in range(10)
+ )
+ ),
+ }
+ metadata.naming_convention = convention
+
+ cons = CheckConstraint("some_long_column_name > 5")
+ Table(
+ "a_things_with_stuff",
+ metadata,
+ Column("id_long_column_name", Integer, primary_key=True),
+ Column("some_long_column_name", Integer),
+ cons,
+ )
+ actual_name = cons.name
+
+ metadata.create_all(connection)
+ insp = inspect(connection)
+ ck = insp.get_check_constraints("a_things_with_stuff")
+ reflected_name = ck[0]["name"]
+ return actual_name, reflected_name
+
+ @testing.combinations(
+ ("fk",),
+ ("pk",),
+ ("ix",),
+ ("ck", testing.requires.check_constraint_reflection.as_skips()),
+ ("uq", testing.requires.unique_constraint_reflection.as_skips()),
+ argnames="type_",
+ )
+ def test_long_convention_name(self, type_, metadata, connection):
+ actual_name, reflected_name = getattr(self, type_)(
+ metadata, connection
+ )
+
+ assert len(actual_name) > 255
+
+ if reflected_name is not None:
+ overlap = actual_name[0 : len(reflected_name)]
+ if len(overlap) < len(actual_name):
+ eq_(overlap[0:-5], reflected_name[0 : len(overlap) - 5])
+ else:
+ eq_(overlap, reflected_name)
+
+
+__all__ = ("TableDDLTest", "FutureTableDDLTest", "LongNameBlowoutTest")