) -> None:
if nullable is not None:
- if existing_type is None:
- if type_ is not None:
- existing_type = type_
- # the NULL/NOT NULL alter will handle
- # the type alteration
- type_ = None
- else:
- raise util.CommandError(
- "MS-SQL ALTER COLUMN operations "
- "with NULL or NOT NULL require the "
- "existing_type or a new type_ be passed."
- )
- elif type_ is not None:
+ if type_ is not None:
# the NULL/NOT NULL alter will handle
# the type alteration
existing_type = type_
type_ = None
+ elif existing_type is None:
+ raise util.CommandError(
+ "MS-SQL ALTER COLUMN operations "
+ "with NULL or NOT NULL require the "
+ "existing_type or a new type_ be passed."
+ )
+ elif existing_nullable is not None and type_ is not None:
+ nullable = existing_nullable
+
+ # the NULL/NOT NULL alter will handle
+ # the type alteration
+ existing_type = type_
+ type_ = None
+
+ elif type_ is not None:
+ util.warn(
+ "MS-SQL ALTER COLUMN operations that specify type_= "
+ "should also specify a nullable= or "
+ "existing_nullable= argument to avoid implicit conversion "
+ "of NOT NULL columns to NULL."
+ )
used_default = False
if sqla_compat._server_default_is_identity(
--- /dev/null
+.. change::
+ :tags: bug, mssql
+ :tickets: 977
+
+ Fixed bug where an alter_column() operation would change a "NOT NULL"
+ column to "NULL" by emitting an ALTER COLUMN statement that did not specify
+ "NOT NULL". (In the absence of "NOT NULL" T-SQL was implicitly assuming
+ "NULL"). An alter_column() operation that specifies `type_=` should also
+ specify either `nullable=` or `existing_nullable=` to inform Alembic as to
+ whether the emitted DDL should include "NULL" or "NOT NULL"; a warning is
+ now emitted if this is missing under this scenario.
"""Test op functions against MSSQL."""
+from typing import Any
+from typing import Dict
+
from sqlalchemy import Column
from sqlalchemy import exc
from sqlalchemy import Integer
from alembic.testing import combinations
from alembic.testing import config
from alembic.testing import eq_
+from alembic.testing import expect_warnings
from alembic.testing.env import _no_sql_testing_config
from alembic.testing.env import clear_staging_env
from alembic.testing.env import staging_env
def test_alter_column_rename_mssql(self):
context = op_fixture("mssql")
- op.alter_column("t", "c", new_column_name="x")
- context.assert_("EXEC sp_rename 't.c', x, 'COLUMN'")
+ op.alter_column(
+ "t",
+ "c",
+ new_column_name="x",
+ )
+ context.assert_(
+ "EXEC sp_rename 't.c', x, 'COLUMN'",
+ )
def test_alter_column_rename_quoted_mssql(self):
context = op_fixture("mssql")
- op.alter_column("t", "c", new_column_name="SomeFancyName")
- context.assert_("EXEC sp_rename 't.c', [SomeFancyName], 'COLUMN'")
+ op.alter_column(
+ "t",
+ "c",
+ new_column_name="SomeFancyName",
+ )
+ context.assert_(
+ "EXEC sp_rename 't.c', [SomeFancyName], 'COLUMN'",
+ )
@combinations((True,), (False,), argnames="pass_existing_type")
+ @combinations((True,), (False,), argnames="existing_nullability")
@combinations((True,), (False,), argnames="change_nullability")
def test_alter_column_type_and_nullability(
- self, pass_existing_type, change_nullability
+ self, pass_existing_type, existing_nullability, change_nullability
):
context = op_fixture("mssql")
- args = dict(type_=Integer)
- if pass_existing_type:
- args["existing_type"] = String(15)
+ args: Dict[str, Any] = dict(type_=Integer)
if change_nullability:
- args["nullable"] = False
+ expected_nullability = not existing_nullability
+ args["nullable"] = expected_nullability
+ else:
+ args[
+ "existing_nullable"
+ ] = expected_nullability = existing_nullability
op.alter_column("t", "c", **args)
- if change_nullability:
- context.assert_("ALTER TABLE t ALTER COLUMN c INTEGER NOT NULL")
- else:
- context.assert_("ALTER TABLE t ALTER COLUMN c INTEGER")
+ context.assert_(
+ f"ALTER TABLE t ALTER COLUMN c INTEGER "
+ f"{'NOT NULL' if not expected_nullability else 'NULL'}"
+ )
+
+ def test_alter_column_type_dont_change_nullability(self):
+ context = op_fixture("mssql")
+
+ op.alter_column("t", "c", type_=String(99), existing_nullable=False)
+ context.assert_contains("ALTER COLUMN c VARCHAR(99) NOT NULL")
+
+ def test_alter_column_type_should_have_existing_nullable(self):
+ context = op_fixture("mssql") # noqa
+ with expect_warnings(
+ "MS-SQL ALTER COLUMN operations that specify type_= should"
+ ):
+ op.alter_column(
+ "t",
+ "c",
+ type_=String(99),
+ )
+ context.assert_contains("ALTER COLUMN c VARCHAR(99)")
def test_alter_column_dont_touch_constraints(self):
context = op_fixture("mssql")
def test_alter_column_drop_default(self):
context = op_fixture("mssql")
- op.alter_column("t", "c", server_default=None)
+ op.alter_column(
+ "t",
+ "c",
+ server_default=None,
+ )
context.assert_contains(
"declare @const_name varchar(256)select @const_name = [name] "
"from sys.default_constraintswhere parent_object_id = "
def test_alter_column_drop_default_w_schema(self):
context = op_fixture("mssql")
- op.alter_column("t", "c", server_default=None, schema="xyz")
+ op.alter_column(
+ "t",
+ "c",
+ server_default=None,
+ schema="xyz",
+ )
context.assert_contains(
"declare @const_name varchar(256)select @const_name = [name] "
"from sys.default_constraintswhere parent_object_id = "
def test_alter_column_dont_drop_default(self):
context = op_fixture("mssql")
- op.alter_column("t", "c", server_default=False)
+ op.alter_column(
+ "t",
+ "c",
+ server_default=False,
+ )
context.assert_()
def test_drop_column_w_schema(self):
def test_alter_add_server_default(self):
context = op_fixture("mssql")
- op.alter_column("t", "c", server_default="5")
- context.assert_("ALTER TABLE t ADD DEFAULT '5' FOR c")
+ op.alter_column(
+ "t",
+ "c",
+ server_default="5",
+ )
+ context.assert_(
+ "ALTER TABLE t ADD DEFAULT '5' FOR c",
+ )
def test_alter_replace_server_default(self):
context = op_fixture("mssql")
op.alter_column(
- "t", "c", server_default="5", existing_server_default="6"
+ "t",
+ "c",
+ server_default="5",
+ existing_server_default="6",
)
context.assert_contains(
"exec('alter table t drop constraint ' + @const_name)"
def test_alter_remove_server_default(self):
context = op_fixture("mssql")
- op.alter_column("t", "c", server_default=None)
+ op.alter_column(
+ "t",
+ "c",
+ server_default=None,
+ )
context.assert_contains(
"exec('alter table t drop constraint ' + @const_name)"
)
def test_alter_column_rename_mssql_schema(self):
context = op_fixture("mssql")
- op.alter_column("t", "c", new_column_name="x", schema="y")
- context.assert_("EXEC sp_rename 'y.t.c', x, 'COLUMN'")
+ op.alter_column(
+ "t",
+ "c",
+ new_column_name="x",
+ schema="y",
+ )
+ context.assert_(
+ "EXEC sp_rename 'y.t.c', x, 'COLUMN'",
+ )
def test_create_index_mssql_include(self):
context = op_fixture("mssql")
)
@config.requirements.computed_columns_api
def test_alter_column_computed_not_supported(self, sd, esd):
- op_fixture("mssql")
+ op_fixture("mysql")
assert_raises_message(
exc.CompileError,
'Adding or removing a "computed" construct, e.g. '
from alembic.testing import combinations
from alembic.testing import config
from alembic.testing import eq_
+from alembic.testing import expect_warnings
from alembic.testing import is_not_
from alembic.testing import mock
from alembic.testing.fixtures import op_fixture
existing_server_default=esd(),
)
- def test_alter_column_schema_type_unnamed(self):
+ @combinations((True,), (False,), (None,), argnames="existing_nullable")
+ def test_alter_column_schema_type_unnamed(self, existing_nullable):
context = op_fixture("mssql", native_boolean=False)
- op.alter_column("t", "c", type_=Boolean(create_constraint=True))
- context.assert_(
- "ALTER TABLE t ALTER COLUMN c BIT",
- "ALTER TABLE t ADD CHECK (c IN (0, 1))",
- )
+ if existing_nullable is None:
+ with expect_warnings(
+ "MS-SQL ALTER COLUMN operations that specify type_= should"
+ ):
+ op.alter_column(
+ "t",
+ "c",
+ type_=Boolean(create_constraint=True),
+ )
+ context.assert_(
+ "ALTER TABLE t ALTER COLUMN c BIT",
+ "ALTER TABLE t ADD CHECK (c IN (0, 1))",
+ )
+ else:
+ op.alter_column(
+ "t",
+ "c",
+ type_=Boolean(create_constraint=True),
+ existing_nullable=existing_nullable,
+ )
+ context.assert_(
+ f"ALTER TABLE t ALTER COLUMN c BIT "
+ f"{'NULL' if existing_nullable else 'NOT NULL'}",
+ "ALTER TABLE t ADD CHECK (c IN (0, 1))",
+ )
def test_alter_column_schema_schema_type_unnamed(self):
context = op_fixture("mssql", native_boolean=False)
op.alter_column(
- "t", "c", type_=Boolean(create_constraint=True), schema="foo"
+ "t",
+ "c",
+ type_=Boolean(create_constraint=True),
+ existing_nullable=False,
+ schema="foo",
)
context.assert_(
- "ALTER TABLE foo.t ALTER COLUMN c BIT",
+ "ALTER TABLE foo.t ALTER COLUMN c BIT NOT NULL",
"ALTER TABLE foo.t ADD CHECK (c IN (0, 1))",
)
def test_alter_column_schema_type_named(self):
context = op_fixture("mssql", native_boolean=False)
op.alter_column(
- "t", "c", type_=Boolean(name="xyz", create_constraint=True)
+ "t",
+ "c",
+ type_=Boolean(name="xyz", create_constraint=True),
+ existing_nullable=False,
)
context.assert_(
- "ALTER TABLE t ALTER COLUMN c BIT",
+ "ALTER TABLE t ALTER COLUMN c BIT NOT NULL",
"ALTER TABLE t ADD CONSTRAINT xyz CHECK (c IN (0, 1))",
)
"t",
"c",
type_=Boolean(name="xyz", create_constraint=True),
+ existing_nullable=False,
schema="foo",
)
context.assert_(
- "ALTER TABLE foo.t ALTER COLUMN c BIT",
+ "ALTER TABLE foo.t ALTER COLUMN c BIT NOT NULL",
"ALTER TABLE foo.t ADD CONSTRAINT xyz CHECK (c IN (0, 1))",
)
"c",
type_=String(10),
existing_type=Boolean(name="xyz", create_constraint=True),
+ existing_nullable=False,
)
context.assert_(
"ALTER TABLE t DROP CONSTRAINT xyz",
- "ALTER TABLE t ALTER COLUMN c VARCHAR(10)",
+ "ALTER TABLE t ALTER COLUMN c VARCHAR(10) NOT NULL",
)
def test_alter_column_schema_schema_type_existing_type(self):
"c",
type_=String(10),
existing_type=Boolean(name="xyz", create_constraint=True),
+ existing_nullable=False,
schema="foo",
)
context.assert_(
"ALTER TABLE foo.t DROP CONSTRAINT xyz",
- "ALTER TABLE foo.t ALTER COLUMN c VARCHAR(10)",
+ "ALTER TABLE foo.t ALTER COLUMN c VARCHAR(10) NOT NULL",
)
def test_alter_column_schema_type_existing_type_no_const(self):
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
