existing_comment=None,
**kw
):
- if name is not None:
+ if name is not None or self._is_mysql_allowed_functional_default(
+ type_ if type_ is not None else existing_type, server_default
+ ):
self._exec(
MySQLChangeColumn(
table_name,
column_name,
schema=schema,
- newname=name,
+ newname=name if name is not None else column_name,
nullable=nullable
if nullable is not None
else existing_nullable
super(MySQLImpl, self).drop_constraint(const)
+ def _is_mysql_allowed_functional_default(self, type_, server_default):
+ return (
+ type_ is not None
+ and type_._type_affinity is sqltypes.DateTime
+ and server_default is not None
+ )
+
def compare_server_default(
self,
inspector_column,
)
return rendered_inspector_default != rendered_metadata_default
elif rendered_inspector_default and rendered_metadata_default:
- # adjust for "function()" vs. "FUNCTION"
+ # adjust for "function()" vs. "FUNCTION" as can occur particularly
+ # for the CURRENT_TIMESTAMP function on newer MariaDB versions
+
+ # SQLAlchemy MySQL dialect bundles ON UPDATE into the server
+ # default; adjust for this possibly being present.
+ onupdate_ins = re.match(
+ r"(.*) (on update.*?)(?:\(\))?$",
+ rendered_inspector_default.lower(),
+ )
+ onupdate_met = re.match(
+ r"(.*) (on update.*?)(?:\(\))?$",
+ rendered_metadata_default.lower(),
+ )
+
+ if onupdate_ins:
+ if not onupdate_met:
+ return True
+ elif onupdate_ins.group(2) != onupdate_met.group(2):
+ return True
+
+ rendered_inspector_default = onupdate_ins.group(1)
+ rendered_metadata_default = onupdate_met.group(1)
+
return re.sub(
r"(.*?)(?:\(\))?$", r"\1", rendered_inspector_default.lower()
) != re.sub(
self.metadata.drop_all(self.conn)
self.conn.close()
- def _run_alter_col(self, from_, to_):
+ def _run_alter_col(self, from_, to_, compare=None):
column = Column(
from_.get("name", "colname"),
from_.get("type", String(10)),
insp = inspect(self.conn)
new_col = insp.get_columns("x")[0]
- eq_(new_col["name"], to_["name"] if "name" in to_ else column.name)
- self._compare_type(new_col["type"], to_.get("type", old_col["type"]))
- eq_(new_col["nullable"], to_.get("nullable", column.nullable))
+ if compare is None:
+ compare = to_
+
+ eq_(
+ new_col["name"],
+ compare["name"] if "name" in compare else column.name,
+ )
+ self._compare_type(
+ new_col["type"], compare.get("type", old_col["type"])
+ )
+ eq_(new_col["nullable"], compare.get("nullable", column.nullable))
self._compare_server_default(
new_col["type"],
new_col.get("default", None),
- to_.get("type", old_col["type"]),
- to_["server_default"].text
- if "server_default" in to_
+ compare.get("type", old_col["type"]),
+ compare["server_default"].text
+ if "server_default" in compare
else column.server_default.arg.text
if column.server_default is not None
else None,
--- /dev/null
+.. change::
+ :tags: bug, mysql, operations
+ :tickets: 564
+
+ Fixed issue where MySQL databases need to use CHANGE COLUMN when altering a
+ server default of CURRENT_TIMESTAMP, NOW() and probably other functions
+ that are only usable with DATETIME/TIMESTAMP columns. While MariaDB
+ supports both CHANGE and ALTER COLUMN in this case, MySQL databases only
+ support CHANGE. So the new logic is that if the server default change is
+ against a DateTime-oriented column, the CHANGE format is used
+ unconditionally, as in the vast majority of cases the server default is to
+ be CURRENT_TIMESTAMP which may also be potentially bundled with an "ON
+ UPDATE CURRENT_TIMESTAMP" directive, which SQLAlchemy does not currently
+ support as a distinct field. The fix addiionally improves the server
+ default comparison logic when the "ON UPDATE" clause is present and
+ there are parenthesis to be adjusted for as is the case on some MariaDB
+ versions.
+
+
from sqlalchemy import Boolean
from sqlalchemy import Column
+from sqlalchemy import DATETIME
from sqlalchemy import func
from sqlalchemy import Integer
from sqlalchemy import MetaData
from alembic.testing import config
from alembic.testing.env import clear_staging_env
from alembic.testing.env import staging_env
+from alembic.testing.fixtures import AlterColRoundTripFixture
from alembic.testing.fixtures import op_fixture
from alembic.testing.fixtures import TestBase
op.alter_column("t", "c", server_default="1")
context.assert_("ALTER TABLE t ALTER COLUMN c SET DEFAULT '1'")
+ def test_alter_column_modify_datetime_default(self):
+ # use CHANGE format when the datatype is DATETIME or TIMESTAMP,
+ # as this is needed for a functional default which is what you'd
+ # get with a DATETIME/TIMESTAMP. Will also work in the very unlikely
+ # case the default is a fixed timestamp value.
+ context = op_fixture("mysql")
+ op.alter_column(
+ "t",
+ "c",
+ existing_type=DATETIME(),
+ server_default=text("CURRENT_TIMESTAMP"),
+ )
+ context.assert_(
+ "ALTER TABLE t CHANGE c c DATETIME NULL DEFAULT CURRENT_TIMESTAMP"
+ )
+
def test_col_not_nullable(self):
context = op_fixture("mysql")
op.alter_column("t1", "c1", nullable=False, existing_type=Integer)
)
+class MySQLBackendOpTest(AlterColRoundTripFixture, TestBase):
+ __only_on__ = "mysql"
+ __backend__ = True
+
+ def test_add_timestamp_server_default_current_timestamp(self):
+ self._run_alter_col(
+ {"type": TIMESTAMP()},
+ {"server_default": text("CURRENT_TIMESTAMP")},
+ )
+
+ def test_add_datetime_server_default_current_timestamp(self):
+ self._run_alter_col(
+ {"type": DATETIME()}, {"server_default": text("CURRENT_TIMESTAMP")}
+ )
+
+ def test_add_timestamp_server_default_now(self):
+ self._run_alter_col(
+ {"type": TIMESTAMP()},
+ {"server_default": text("NOW()")},
+ compare={"server_default": text("CURRENT_TIMESTAMP")},
+ )
+
+ def test_add_datetime_server_default_now(self):
+ self._run_alter_col(
+ {"type": DATETIME()},
+ {"server_default": text("NOW()")},
+ compare={"server_default": text("CURRENT_TIMESTAMP")},
+ )
+
+ def test_add_timestamp_server_default_current_timestamp_bundle_onupdate(
+ self
+ ):
+ # note SQLAlchemy reflection bundles the ON UPDATE part into the
+ # server default reflection see
+ # https://github.com/sqlalchemy/sqlalchemy/issues/4652
+ self._run_alter_col(
+ {"type": TIMESTAMP()},
+ {
+ "server_default": text(
+ "CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"
+ )
+ },
+ )
+
+ def test_add_datetime_server_default_current_timestamp_bundle_onupdate(
+ self
+ ):
+ # note SQLAlchemy reflection bundles the ON UPDATE part into the
+ # server default reflection see
+ # https://github.com/sqlalchemy/sqlalchemy/issues/4652
+ self._run_alter_col(
+ {"type": DATETIME()},
+ {
+ "server_default": text(
+ "CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"
+ )
+ },
+ )
+
+
class MySQLDefaultCompareTest(TestBase):
__only_on__ = "mysql"
__backend__ = True
def test_compare_timestamp_current_timestamp_diff(self):
self._compare_default_roundtrip(TIMESTAMP(), None, "CURRENT_TIMESTAMP")
+ def test_compare_timestamp_current_timestamp_bundle_onupdate(self):
+ self._compare_default_roundtrip(
+ TIMESTAMP(), "CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"
+ )
+
+ def test_compare_timestamp_current_timestamp_diff_bundle_onupdate(self):
+ self._compare_default_roundtrip(
+ TIMESTAMP(), None, "CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"
+ )
+
def test_compare_integer_from_none(self):
self._compare_default_roundtrip(Integer(), None, "0")
)
-# MARKMARK
class BackendAlterColumnTest(AlterColRoundTripFixture, TestBase):
__backend__ = True
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
