return None
+def _normalize_computed_default(sqltext):
+ """we want to warn if a computed sql expression has changed. however
+ we don't want false positives and the warning is not that critical.
+ so filter out most forms of variability from the SQL text.
+
+ """
+
+ return re.sub(r"[ \(\)'\"`\[\]]", "", sqltext).lower()
+
+
+def _compare_computed_default(
+ autogen_context,
+ alter_column_op,
+ schema,
+ tname,
+ cname,
+ conn_col,
+ metadata_col,
+):
+ rendered_metadata_default = str(
+ metadata_col.server_default.sqltext.compile(
+ dialect=autogen_context.dialect,
+ compile_kwargs={"literal_binds": True},
+ )
+ )
+
+ # since we cannot change computed columns, we do only a crude comparison
+ # here where we try to eliminate syntactical differences in order to
+ # get a minimal comparison just to emit a warning.
+
+ rendered_metadata_default = _normalize_computed_default(
+ rendered_metadata_default
+ )
+
+ if isinstance(conn_col.server_default, sa_schema.Computed):
+ rendered_conn_default = str(
+ conn_col.server_default.sqltext.compile(
+ dialect=autogen_context.dialect,
+ compile_kwargs={"literal_binds": True},
+ )
+ )
+ if rendered_conn_default is None:
+ rendered_conn_default = ""
+ else:
+ rendered_conn_default = _normalize_computed_default(
+ rendered_conn_default
+ )
+ else:
+ rendered_conn_default = ""
+
+ if rendered_metadata_default != rendered_conn_default:
+ _warn_computed_not_supported(tname, cname)
+
+
+def _warn_computed_not_supported(tname, cname):
+ util.warn("Computed default on %s.%s cannot be modified" % (tname, cname))
+
+
@comparators.dispatch_for("column")
def _compare_server_default(
autogen_context,
# Once SQLAlchemy can reflect "GENERATED" as the "computed" element,
# we would also want to ignore and/or warn for changes vs. the
# metadata (or support backend specific DDL if applicable).
- return False
+ if not sqla_compat.has_computed_reflection:
+ return False
+ else:
+ return _compare_computed_default(
+ autogen_context,
+ alter_column_op,
+ schema,
+ tname,
+ cname,
+ conn_col,
+ metadata_col,
+ )
rendered_metadata_default = _render_server_default_for_compare(
metadata_default, metadata_col, autogen_context
)
- rendered_conn_default = (
- conn_col.server_default.arg.text if conn_col.server_default else None
- )
+ if sqla_compat.has_computed_reflection and isinstance(
+ conn_col.server_default, sa_schema.Computed
+ ):
+ _warn_computed_not_supported(tname, cname)
+ return False
+ else:
+ rendered_conn_default = (
+ conn_col.server_default.arg.text
+ if conn_col.server_default
+ else None
+ )
alter_column_op.existing_server_default = conn_col_default
from sqlalchemy import Computed # noqa
has_computed = True
+
+ has_computed_reflection = _vers >= (1, 3, 16)
except ImportError:
has_computed = False
+ has_computed_reflection = False
AUTOINCREMENT_DEFAULT = "auto"
--- /dev/null
+.. change::
+ :tags: usecase, autogenerate
+ :tickets: 669
+
+ Adjusted autogen comparison to accommodate for backends that support
+ computed column reflection, dependent on SQLAlchemy version 1.3.16 or
+ higher. This emits a warning if the SQL expression inside of a
+ :class:`.Computed` value changes between the metadata and the database, as
+ these expressions can't be changed without dropping and recreating the
+ column.
+
["postgresql >= 12", "oracle", "mssql", "mysql >= 5.7"]
)
+ @property
+ def computed_reflects_normally(self):
+ return exclusions.only_if(
+ exclusions.BooleanPredicate(sqla_compat.has_computed_reflection)
+ )
+
@property
def computed_reflects_as_server_default(self):
# note that this rule will go away when SQLAlchemy correctly
# supports reflection of the "computed" construct; the element
# will consistently be present as both column.computed and
# column.server_default for all supported backends.
- return self.computed_columns + exclusions.only_if(
- ["postgresql", "oracle"],
- "backend reflects computed construct as a server default",
+ return (
+ self.computed_columns
+ + exclusions.only_if(
+ ["postgresql", "oracle"],
+ "backend reflects computed construct as a server default",
+ )
+ + exclusions.skip_if(self.computed_reflects_normally)
)
@property
# supports reflection of the "computed" construct; the element
# will consistently be present as both column.computed and
# column.server_default for all supported backends.
- return self.computed_columns + exclusions.skip_if(
- ["postgresql", "oracle"],
- "backend reflects computed construct as a server default",
+ return (
+ self.computed_columns
+ + exclusions.skip_if(
+ ["postgresql", "oracle"],
+ "backend reflects computed construct as a server default",
+ )
+ + exclusions.skip_if(self.computed_reflects_normally)
)
@property
from alembic.testing import exclusions
from alembic.testing import is_
from alembic.testing import is_true
+from alembic.testing import mock
from alembic.testing import TestBase
from ._autogen_fixtures import AutogenFixtureTest
c = diffs[0][3]
eq_(c.name, "foo")
- is_(c.computed, None)
+ if config.requirements.computed_reflects_normally.enabled:
+ is_true(isinstance(c.computed, sa.Computed))
+ else:
+ is_(c.computed, None)
if config.requirements.computed_reflects_as_server_default.enabled:
is_true(isinstance(c.server_default, sa.DefaultClause))
eq_(str(c.server_default.arg.text), "5")
+ elif config.requirements.computed_reflects_normally.enabled:
+ is_true(isinstance(c.computed, sa.Computed))
else:
- is_(c.server_default, None)
+ is_(c.computed, None)
@testing.combinations(
- lambda: (sa.Computed("5"), sa.Computed("5")),
- lambda: (sa.Computed("bar*5"), sa.Computed("bar*5")),
- lambda: (sa.Computed("bar*5"), sa.Computed("bar * 42")),
+ lambda: (None, sa.Computed("bar*5")),
+ (lambda: (sa.Computed("bar*5"), None)),
lambda: (
sa.Computed("bar*5"),
sa.Computed("bar * 42", persisted=True),
),
- lambda: (None, sa.Computed("bar*5")),
+ lambda: (sa.Computed("bar*5"), sa.Computed("bar * 42")),
+ )
+ @config.requirements.computed_reflects_normally
+ def test_cant_change_computed_warning(self, test_case):
+ arg_before, arg_after = testing.resolve_lambda(test_case, **locals())
+ m1 = MetaData()
+ m2 = MetaData()
+
+ arg_before = [] if arg_before is None else [arg_before]
+ arg_after = [] if arg_after is None else [arg_after]
+
+ Table(
+ "user",
+ m1,
+ Column("id", Integer, primary_key=True),
+ Column("bar", Integer),
+ Column("foo", Integer, *arg_before),
+ )
+
+ Table(
+ "user",
+ m2,
+ Column("id", Integer, primary_key=True),
+ Column("bar", Integer),
+ Column("foo", Integer, *arg_after),
+ )
+
+ with mock.patch("alembic.util.warn") as mock_warn:
+ diffs = self._fixture(m1, m2)
+
+ eq_(
+ mock_warn.mock_calls,
+ [mock.call("Computed default on user.foo cannot be modified")],
+ )
+
+ eq_(len(diffs), 0)
+
+ @testing.combinations(
+ lambda: (None, None),
+ lambda: (sa.Computed("5"), sa.Computed("5")),
+ lambda: (sa.Computed("bar*5"), sa.Computed("bar*5")),
(
lambda: (sa.Computed("bar*5"), None),
config.requirements.computed_doesnt_reflect_as_server_default,
Column("foo", Integer, *arg_after),
)
- diffs = self._fixture(m1, m2)
+ with mock.patch("alembic.util.warn") as mock_warn:
+ diffs = self._fixture(m1, m2)
+ eq_(mock_warn.mock_calls, [])
eq_(len(diffs), 0)
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
