schema: Optional[Union[quoted_name, str]] = None,
if_not_exists: Optional[bool] = None,
inline_references: Optional[bool] = None,
+ inline_primary_key: Optional[bool] = None,
) -> None:
super().__init__(name, schema=schema)
self.column = column
self.if_not_exists = if_not_exists
self.inline_references = inline_references
+ self.inline_primary_key = inline_primary_key
class DropColumn(AlterTable):
element.column,
if_not_exists=element.if_not_exists,
inline_references=element.inline_references,
+ inline_primary_key=element.inline_primary_key,
**kw,
),
)
column: Column[Any],
if_not_exists: Optional[bool] = None,
inline_references: Optional[bool] = None,
+ inline_primary_key: Optional[bool] = None,
**kw,
) -> str:
text = "ADD COLUMN %s%s" % (
compiler.get_column_specification(column, **kw),
)
- if column.primary_key:
+ if inline_primary_key and column.primary_key:
text += " PRIMARY KEY"
# Handle inline REFERENCES if requested
schema: Optional[Union[str, quoted_name]] = None,
if_not_exists: Optional[bool] = None,
inline_references: Optional[bool] = None,
+ inline_primary_key: Optional[bool] = None,
) -> None:
self._exec(
base.AddColumn(
schema=schema,
if_not_exists=if_not_exists,
inline_references=inline_references,
+ inline_primary_key=inline_primary_key,
)
)
schema: Optional[str] = None,
if_not_exists: Optional[bool] = None,
inline_references: Optional[bool] = None,
+ inline_primary_key: Optional[bool] = None,
) -> None:
"""Issue an "add column" instruction using the current
migration context.
.. versionadded:: 1.18.2
+ :param inline_primary_key: If True, renders the PRIMARY KEY constraint
+ inline within the ADD COLUMN directive, rather than rendering it
+ separately. This is a purely syntactic option and should only be
+ used for single-column primary keys.
+
+ .. versionadded:: 1.18.4
+
"""
def alter_column(
schema: Optional[str] = None,
if_not_exists: Optional[bool] = None,
inline_references: Optional[bool] = None,
+ inline_primary_key: Optional[bool] = None,
) -> None:
"""Issue an "add column" instruction using the current
migration context.
.. versionadded:: 1.18.2
+ :param inline_primary_key: If True, renders the PRIMARY KEY constraint
+ inline within the ADD COLUMN directive, rather than rendering it
+ separately. This is a purely syntactic option and should only be
+ used for single-column primary keys.
+
+ .. versionadded:: 1.18.4
+
""" # noqa: E501
...
insert_after: Optional[str] = None,
if_not_exists: Optional[bool] = None,
inline_references: Optional[bool] = None,
+ inline_primary_key: Optional[bool] = None,
) -> None:
"""Issue an "add column" instruction using the current
batch migration context.
schema: Optional[str] = None,
if_not_exists: Optional[bool] = None,
inline_references: Optional[bool] = None,
+ inline_primary_key: Optional[bool] = None,
**kw: Any,
) -> None:
super().__init__(table_name, schema=schema)
self.column = column
self.if_not_exists = if_not_exists
self.inline_references = inline_references
+ self.inline_primary_key = inline_primary_key
self.kw = kw
def reverse(self) -> DropColumnOp:
schema: Optional[str] = None,
if_not_exists: Optional[bool] = None,
inline_references: Optional[bool] = None,
+ inline_primary_key: Optional[bool] = None,
) -> None:
"""Issue an "add column" instruction using the current
migration context.
.. versionadded:: 1.18.2
+ :param inline_primary_key: If True, renders the PRIMARY KEY constraint
+ inline within the ADD COLUMN directive, rather than rendering it
+ separately. This is a purely syntactic option and should only be
+ used for single-column primary keys.
+
+ .. versionadded:: 1.18.4
+
"""
op = cls(
schema=schema,
if_not_exists=if_not_exists,
inline_references=inline_references,
+ inline_primary_key=inline_primary_key,
)
return operations.invoke(op)
insert_after: Optional[str] = None,
if_not_exists: Optional[bool] = None,
inline_references: Optional[bool] = None,
+ inline_primary_key: Optional[bool] = None,
) -> None:
"""Issue an "add column" instruction using the current
batch migration context.
schema=operations.impl.schema,
if_not_exists=if_not_exists,
inline_references=inline_references,
+ inline_primary_key=inline_primary_key,
**kw,
)
return operations.invoke(op)
schema = operation.schema
kw = operation.kw
inline_references = operation.inline_references
+ inline_primary_key = operation.inline_primary_key
if column.table is not None:
column = _copy(column)
schema=schema,
if_not_exists=operation.if_not_exists,
inline_references=inline_references,
+ inline_primary_key=inline_primary_key,
**kw,
)
status with any existing primary key constraint or particular backend
limitations on adding columns to the primary key.
+ .. note::
+
+ As of version 1.18.4, this behavior has been amended to be opt-in
+ via the new ``inline_primary_key`` parameter to
+ :meth:`.Operations.add_column`, rather than occurring automatically
+ when ``primary_key=True`` is set on the :class:`.Column` object.
+
.. change::
:tags: bug, typing
:tickets: 1669
--- /dev/null
+.. change::
+ :tags: bug, operations
+ :tickets: 1232
+
+ Reverted the behavior of :meth:`.Operations.add_column` that would
+ automatically render the "PRIMARY KEY" keyword inline when a
+ :class:`.Column` with ``primary_key=True`` is added. The automatic
+ behavior, added in version 1.18.2, is now opt-in via the new
+ :paramref:`.Operations.add_column.inline_primary_key` parameter. This
+ change restores the ability to render a PostgreSQL SERIAL column, which is
+ required to be ``primary_key=True``, while not impacting the ability to
+ render a separate primary key constraint. This also provides consistency
+ with the :paramref:`.Operations.add_column.inline_references` parameter and
+ gives users explicit control over SQL generation.
+
+ To render PRIMARY KEY inline, use the
+ :paramref:`.Operations.add_column.inline_primary_key` parameter set to
+ ``True``::
+
+ op.add_column(
+ "my_table",
+ Column("id", Integer, primary_key=True),
+ inline_primary_key=True
+ )
schema=None,
if_not_exists=None,
inline_references=None,
+ inline_primary_key=None,
)
in batch.impl.operations.impl.mock_calls
)
pk_const = inspect(self.conn).get_pk_constraint("foo")
eq_(pk_const["constrained_columns"], [])
- def test_drop_pk_col_readd_pk_col(self):
+ @testing.variation(
+ "use_inline_pk",
+ [
+ True,
+ (False, exclusions.fails_on(["postgresql", "mysql", "mariadb"])),
+ ],
+ )
+ def test_drop_pk_col_readd_pk_col(self, use_inline_pk):
# drop a column, add it back with primary_key=True, should remain
with self.op.batch_alter_table("foo") as batch_op:
batch_op.drop_column("id")
- batch_op.add_column(Column("id", Integer, primary_key=True))
+ batch_op.add_column(
+ Column("id", Integer, primary_key=True),
+ inline_primary_key=bool(use_inline_pk),
+ )
pk_const = inspect(self.conn).get_pk_constraint("foo")
eq_(pk_const["constrained_columns"], ["id"])
def _datetime_server_default_fixture(self):
return func.current_timestamp()
- def test_drop_pk_col_readd_pk_col(self):
- super().test_drop_pk_col_readd_pk_col()
-
@exclusions.fails()
def test_drop_pk_col_readd_col_also_pk_const(self):
super().test_drop_pk_col_readd_col_also_pk_const()
def _datetime_server_default_fixture(self):
return func.current_timestamp()
- def test_drop_pk_col_readd_pk_col(self):
- super().test_drop_pk_col_readd_pk_col()
-
@exclusions.fails()
def test_drop_pk_col_readd_col_also_pk_const(self):
super().test_drop_pk_col_readd_col_also_pk_const()
from sqlalchemy.sql.schema import quoted_name
from alembic import op
+from alembic import testing
from alembic.operations import MigrateOperation
from alembic.operations import Operations
from alembic.operations import ops
'ALTER TABLE "some.schema".somename ADD COLUMN colname VARCHAR'
)
- def test_add_column_primary_key(self):
+ @testing.variation("use_inline", [True, False, "none"])
+ def test_add_column_primary_key(self, use_inline):
+ context = op_fixture("postgresql")
+
+ if use_inline.none:
+ op.add_column(
+ "somename",
+ Column("colname", String, primary_key=True),
+ )
+ else:
+ op.add_column(
+ "somename",
+ Column("colname", String, primary_key=True),
+ inline_primary_key=bool(use_inline),
+ )
+
+ if use_inline.use_inline:
+ context.assert_(
+ "ALTER TABLE somename ADD COLUMN colname "
+ "VARCHAR NOT NULL PRIMARY KEY"
+ )
+ else:
+ context.assert_(
+ "ALTER TABLE somename ADD COLUMN colname " "VARCHAR NOT NULL"
+ )
+
+ def test_add_column_primary_key_not_inline_by_default(self):
context = op_fixture("postgresql")
op.add_column(
"somename",
)
context.assert_(
- "ALTER TABLE somename ADD COLUMN colname "
- "VARCHAR NOT NULL PRIMARY KEY"
+ "ALTER TABLE somename ADD COLUMN colname VARCHAR NOT NULL"
)
+ def test_add_column_inline_primary_key_no_pk_flag(self):
+ context = op_fixture("postgresql")
+ op.add_column(
+ "somename",
+ Column("colname", String),
+ inline_primary_key=True,
+ )
+
+ context.assert_("ALTER TABLE somename ADD COLUMN colname VARCHAR")
+
def test_rename_table_schema_hard_quoting(self):
context = op_fixture("postgresql")
op.rename_table(
def test_col_w_pk_is_serial(self):
context = op_fixture("postgresql")
- op.add_column("some_table", Column("q", Integer, primary_key=True))
+ op.add_column(
+ "some_table",
+ Column("q", Integer, primary_key=True),
+ inline_primary_key=True,
+ )
context.assert_(
"ALTER TABLE some_table ADD COLUMN q SERIAL NOT NULL PRIMARY KEY"
)
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
