Added Table and Column level comments for supported backends.
`create_table`, `add_column` and `alter_column` now optionally
take `comment="X"` kwarg. Support for autogenerate for Table
and Column objects has also been added
Fixes: #422
Change-Id: I1fd37bb7fe3d167baf7b1e7bf7ff5bfd48e7cf54
log.info("Detected server default on column '%s.%s'", tname, cname)
+@comparators.dispatch_for("column")
+def _compare_column_comment(
+ autogen_context,
+ alter_column_op,
+ schema,
+ tname,
+ cname,
+ conn_col,
+ metadata_col,
+):
+
+ if not sqla_compat._dialect_supports_comments(autogen_context.dialect):
+ return
+
+ metadata_comment = metadata_col.comment
+ conn_col_comment = conn_col.comment
+ if conn_col_comment is None and metadata_comment is None:
+ return False
+
+ alter_column_op.existing_comment = conn_col_comment
+
+ if conn_col_comment != metadata_comment:
+ alter_column_op.modify_comment = metadata_comment
+ log.info("Detected column comment '%s.%s'", tname, cname)
+
+
@comparators.dispatch_for("table")
def _compare_foreign_keys(
autogen_context,
else None
)
_add_fk(const, compare_to)
+
+
+@comparators.dispatch_for("table")
+def _compare_table_comment(
+ autogen_context,
+ modify_table_ops,
+ schema,
+ tname,
+ conn_table,
+ metadata_table,
+):
+
+ if not sqla_compat._dialect_supports_comments(autogen_context.dialect):
+ return
+
+ # if we're doing CREATE TABLE, comments will be created inline
+ # with the create_table op.
+ if conn_table is None or metadata_table is None:
+ return
+
+ if conn_table.comment is None and metadata_table.comment is None:
+ return
+
+ if metadata_table.comment is None and conn_table.comment is not None:
+ modify_table_ops.ops.append(
+ ops.DropTableCommentOp(
+ tname, existing_comment=conn_table.comment, schema=schema
+ )
+ )
+ elif metadata_table.comment != conn_table.comment:
+ modify_table_ops.ops.append(
+ ops.CreateTableCommentOp(
+ tname,
+ metadata_table.comment,
+ existing_comment=conn_table.comment,
+ schema=schema,
+ )
+ )
return ["pass"]
+@renderers.dispatch_for(ops.CreateTableCommentOp)
+def _render_create_table_comment(autogen_context, op):
+
+ templ = (
+ "{prefix}create_table_comment(\n"
+ "{indent}'{tname}',\n"
+ "{indent}{comment},\n"
+ "{indent}existing_comment={existing},\n"
+ "{indent}schema={schema}\n"
+ ")"
+ )
+ return templ.format(
+ prefix=_alembic_autogenerate_prefix(autogen_context),
+ tname=op.table_name,
+ comment="'%s'" % op.comment if op.comment is not None else None,
+ existing="'%s'" % op.existing_comment
+ if op.existing_comment is not None
+ else None,
+ schema="'%s'" % op.schema if op.schema is not None else None,
+ indent=" ",
+ )
+
+
+@renderers.dispatch_for(ops.DropTableCommentOp)
+def _render_drop_table_comment(autogen_context, op):
+
+ templ = (
+ "{prefix}drop_table_comment(\n"
+ "{indent}'{tname}',\n"
+ "{indent}existing_comment={existing},\n"
+ "{indent}schema={schema}\n"
+ ")"
+ )
+ return templ.format(
+ prefix=_alembic_autogenerate_prefix(autogen_context),
+ tname=op.table_name,
+ existing="'%s'" % op.existing_comment
+ if op.existing_comment is not None
+ else None,
+ schema="'%s'" % op.schema if op.schema is not None else None,
+ indent=" ",
+ )
+
+
@renderers.dispatch_for(ops.CreateTableOp)
def _add_table(autogen_context, op):
table = op.to_table()
}
if op.schema:
text += ",\nschema=%r" % _ident(op.schema)
+
+ comment = sqla_compat._comment_attribute(table)
+ if comment:
+ text += ",\ncomment=%r" % _ident(comment)
for k in sorted(op.kw):
text += ",\n%s=%r" % (k.replace(" ", "_"), op.kw[k])
text += "\n)"
server_default = op.modify_server_default
type_ = op.modify_type
nullable = op.modify_nullable
+ comment = op.modify_comment
autoincrement = op.kw.get("autoincrement", None)
existing_type = op.existing_type
existing_nullable = op.existing_nullable
+ existing_comment = op.existing_comment
existing_server_default = op.existing_server_default
schema = op.schema
text += ",\n%stype_=%s" % (indent, _repr_type(type_, autogen_context))
if nullable is not None:
text += ",\n%snullable=%r" % (indent, nullable)
+ if comment is not False:
+ text += ",\n%scomment=%r" % (indent, comment)
+ if existing_comment is not None:
+ text += ",\n%sexisting_comment=%r" % (indent, existing_comment)
if nullable is None and existing_nullable is not None:
text += ",\n%sexisting_nullable=%r" % (indent, existing_nullable)
if autoincrement is not None:
if column.system:
opts.append(("system", column.system))
+ comment = sqla_compat._comment_attribute(column)
+ if comment:
+ opts.append(("comment", "'%s'" % comment))
+
# TODO: for non-ascii colname, assign a "key"
return "%(prefix)sColumn(%(name)r, %(type)s, %(kw)s)" % {
"prefix": _sqlalchemy_autogenerate_prefix(autogen_context),
existing_type=None,
existing_nullable=None,
existing_server_default=None,
+ existing_comment=None,
):
super(AlterColumn, self).__init__(name, schema=schema)
self.column_name = column_name
)
self.existing_nullable = existing_nullable
self.existing_server_default = existing_server_default
+ self.existing_comment = existing_comment
class ColumnNullable(AlterColumn):
self.column = column
+class ColumnComment(AlterColumn):
+ def __init__(self, name, column_name, comment, **kw):
+ super(ColumnComment, self).__init__(name, column_name, **kw)
+ self.comment = comment
+
+
@compiles(RenameTable)
def visit_rename_table(element, compiler, **kw):
return "%s RENAME TO %s" % (
type_=None,
schema=None,
autoincrement=None,
+ comment=False,
+ existing_comment=None,
existing_type=None,
existing_server_default=None,
existing_nullable=None,
existing_type=existing_type,
existing_server_default=existing_server_default,
existing_nullable=existing_nullable,
+ existing_comment=existing_comment,
)
)
if server_default is not False:
existing_type=existing_type,
existing_server_default=existing_server_default,
existing_nullable=existing_nullable,
+ existing_comment=existing_comment,
)
)
if type_ is not None:
existing_type=existing_type,
existing_server_default=existing_server_default,
existing_nullable=existing_nullable,
+ existing_comment=existing_comment,
)
)
+
+ if comment is not False:
+ self._exec(
+ base.ColumnComment(
+ table_name,
+ column_name,
+ comment,
+ schema=schema,
+ existing_type=existing_type,
+ existing_server_default=existing_server_default,
+ existing_nullable=existing_nullable,
+ existing_comment=existing_comment,
+ )
+ )
+
# do the new name last ;)
if name is not None:
self._exec(
for index in table.indexes:
self._exec(schema.CreateIndex(index))
+ with_comment = (
+ sqla_compat._dialect_supports_comments(self.dialect)
+ and not self.dialect.inline_comments
+ )
+ comment = sqla_compat._comment_attribute(table)
+ if comment and with_comment:
+ self.create_table_comment(table)
+
+ for column in table.columns:
+ comment = sqla_compat._comment_attribute(column)
+ if comment and with_comment:
+ self.create_column_comment(column)
+
def drop_table(self, table):
self._exec(schema.DropTable(table))
def create_index(self, index):
self._exec(schema.CreateIndex(index))
+ def create_table_comment(self, table):
+ self._exec(schema.SetTableComment(table))
+
+ def drop_table_comment(self, table):
+ self._exec(schema.DropTableComment(table))
+
+ def create_column_comment(self, column):
+ self._exec(schema.SetColumnComment(column))
+
def drop_index(self, index):
self._exec(schema.DropIndex(index))
existing_nullable=None,
autoincrement=None,
existing_autoincrement=None,
+ comment=False,
+ existing_comment=None,
**kw
):
if name is not None:
nullable is not None
or type_ is not None
or autoincrement is not None
+ or comment is not False
):
self._exec(
MySQLModifyColumn(
autoincrement=autoincrement
if autoincrement is not None
else existing_autoincrement,
+ comment=comment
+ if comment is not False
+ else existing_comment,
)
)
elif server_default is not False:
nullable=None,
default=False,
autoincrement=None,
+ comment=False,
):
super(AlterColumn, self).__init__(name, schema=schema)
self.column_name = column_name
self.newname = newname
self.default = default
self.autoincrement = autoincrement
+ self.comment = comment
if type_ is None:
raise util.CommandError(
"All MySQL CHANGE/MODIFY COLUMN operations "
server_default=element.default,
type_=element.type_,
autoincrement=element.autoincrement,
+ comment=element.comment,
),
)
server_default=element.default,
type_=element.type_,
autoincrement=element.autoincrement,
+ comment=element.comment,
),
)
return compiler.sql_compiler.process(expr)
-def _mysql_colspec(compiler, nullable, server_default, type_, autoincrement):
+def _mysql_colspec(
+ compiler, nullable, server_default, type_, autoincrement, comment
+):
spec = "%s %s" % (
compiler.dialect.type_compiler.process(type_),
"NULL" if nullable else "NOT NULL",
spec += " AUTO_INCREMENT"
if server_default is not False and server_default is not None:
spec += " DEFAULT %s" % _render_value(compiler, server_default)
+ if comment:
+ spec += " COMMENT %s" % compiler.sql_compiler.render_literal_value(
+ comment, sqltypes.String()
+ )
return spec
from sqlalchemy.ext.compiler import compiles
+from sqlalchemy.sql import sqltypes
from .base import AddColumn
from .base import alter_table
+from .base import ColumnComment
from .base import ColumnDefault
from .base import ColumnName
from .base import ColumnNullable
)
+@compiles(ColumnComment, "oracle")
+def visit_column_comment(element, compiler, **kw):
+ ddl = "COMMENT ON COLUMN {table_name}.{column_name} IS {comment}"
+
+ comment = compiler.sql_compiler.render_literal_value(
+ (element.comment if element.comment is not None else ""),
+ sqltypes.String(),
+ )
+
+ return ddl.format(
+ table_name=element.table_name,
+ column_name=element.column_name,
+ comment=comment,
+ )
+
+
def alter_column(compiler, name):
return "MODIFY %s" % format_column_name(compiler, name)
from .base import alter_column
from .base import alter_table
from .base import AlterColumn
+from .base import ColumnComment
from .base import compiles
from .base import format_table_name
from .base import format_type
)
+@compiles(ColumnComment, "postgresql")
+def visit_column_comment(element, compiler, **kw):
+ ddl = "COMMENT ON COLUMN {table_name}.{column_name} IS {comment}"
+ comment = (
+ compiler.sql_compiler.render_literal_value(
+ element.comment, sqltypes.String()
+ )
+ if element.comment is not None
+ else "NULL"
+ )
+
+ return ddl.format(
+ table_name=element.table_name,
+ column_name=element.column_name,
+ comment=comment,
+ )
+
+
@Operations.register_operation("create_exclude_constraint")
@BatchOperations.register_operation(
"create_exclude_constraint", "batch_create_exclude_constraint"
from .ops import MigrateOperation
-
__all__ = ["Operations", "BatchOperations", "MigrateOperation"]
return operations.invoke(op)
+@Operations.register_operation("create_table_comment")
+class CreateTableCommentOp(AlterTableOp):
+ """Represent a COMMENT ON `table` operation.
+ """
+
+ def __init__(
+ self, table_name, comment, schema=None, existing_comment=None
+ ):
+ self.table_name = table_name
+ self.comment = comment
+ self.existing_comment = existing_comment
+ self.schema = schema
+
+ @classmethod
+ def create_table_comment(
+ cls,
+ operations,
+ table_name,
+ comment,
+ existing_comment=None,
+ schema=None,
+ ):
+ """Invokes the `:func:`alembic.operations.toimpl.create_table_comment`
+ impl to initiate a new COMMENT ON `table` operation.
+
+ :param table_name: string name of the target table.
+ :param comment: string value of the comment being registered against
+ the specified table.
+ :param existing_comment: An optional string value of a comment
+ already registered
+ on the specified table.
+
+ """
+
+ op = cls(
+ table_name,
+ comment,
+ existing_comment=existing_comment,
+ schema=schema,
+ )
+ return operations.invoke(op)
+
+ def reverse(self):
+ """Reverses the COMMENT ON operation against a table.
+ """
+ if self.existing_comment is None:
+ return DropTableCommentOp(
+ self.table_name,
+ existing_comment=self.comment,
+ schema=self.schema,
+ )
+ else:
+ return CreateTableCommentOp(
+ self.table_name,
+ self.existing_comment,
+ existing_comment=self.comment,
+ schema=self.schema,
+ )
+
+ def to_table(self, migration_context=None):
+ schema_obj = schemaobj.SchemaObjects(migration_context)
+
+ return schema_obj.table(
+ self.table_name, schema=self.schema, comment=self.comment
+ )
+
+ def to_diff_tuple(self):
+ return ("add_table_comment", self.to_table(), self.existing_comment)
+
+
+@Operations.register_operation("drop_table_comment")
+class DropTableCommentOp(AlterTableOp):
+ """Represent a COMMENT ON `table` operation.
+ """
+
+ def __init__(self, table_name, schema=None, existing_comment=None):
+ self.table_name = table_name
+ self.existing_comment = existing_comment
+ self.schema = schema
+
+ @classmethod
+ def drop_table_comment(
+ cls, operations, table_name, existing_comment=None, schema=None
+ ):
+ """Invokes the `:func:`alembic.operations.toimpl.drop_table_comment` to
+ remove an existing comment set on a table.
+
+ :param table_name: string name of the target table.
+ :param existing_comment: An optional string value of a comment already
+ registered
+ on the specified table.
+
+ """
+
+ op = cls(table_name, existing_comment=existing_comment, schema=schema)
+ return operations.invoke(op)
+
+ def reverse(self):
+ """Reverses the COMMENT ON operation against a table.
+ """
+ return CreateTableCommentOp(
+ self.table_name, self.existing_comment, schema=self.schema
+ )
+
+ def to_table(self, migration_context=None):
+ schema_obj = schemaobj.SchemaObjects(migration_context)
+
+ return schema_obj.table(self.table_name, schema=self.schema)
+
+ def to_diff_tuple(self):
+ return ("remove_table_comment", self.to_table())
+
+
@Operations.register_operation("alter_column")
@BatchOperations.register_operation("alter_column", "batch_alter_column")
class AlterColumnOp(AlterTableOp):
existing_type=None,
existing_server_default=False,
existing_nullable=None,
+ existing_comment=None,
modify_nullable=None,
+ modify_comment=False,
modify_server_default=False,
modify_name=None,
modify_type=None,
self.existing_type = existing_type
self.existing_server_default = existing_server_default
self.existing_nullable = existing_nullable
+ self.existing_comment = existing_comment
self.modify_nullable = modify_nullable
+ self.modify_comment = modify_comment
self.modify_server_default = modify_server_default
self.modify_name = modify_name
self.modify_type = modify_type
"existing_server_default": (
self.existing_server_default
),
+ "existing_comment": self.existing_comment,
},
self.existing_type,
self.modify_type,
"existing_server_default": (
self.existing_server_default
),
+ "existing_comment": self.existing_comment,
},
self.existing_nullable,
self.modify_nullable,
{
"existing_nullable": self.existing_nullable,
"existing_type": self.existing_type,
+ "existing_comment": self.existing_comment,
},
self.existing_server_default,
self.modify_server_default,
)
)
+ if self.modify_comment is not False:
+ col_diff.append(
+ (
+ "modify_comment",
+ schema,
+ tname,
+ cname,
+ {
+ "existing_nullable": self.existing_nullable,
+ "existing_type": self.existing_type,
+ "existing_server_default": (
+ self.existing_server_default
+ ),
+ },
+ self.existing_comment,
+ self.modify_comment,
+ )
+ )
+
return col_diff
def has_changes(self):
self.modify_nullable is not None
or self.modify_server_default is not False
or self.modify_type is not None
+ or self.modify_comment is not False
)
if hc1:
return True
kw["existing_type"] = self.existing_type
kw["existing_nullable"] = self.existing_nullable
kw["existing_server_default"] = self.existing_server_default
+ kw["existing_comment"] = self.existing_comment
if self.modify_type is not None:
kw["modify_type"] = self.modify_type
if self.modify_nullable is not None:
kw["modify_nullable"] = self.modify_nullable
if self.modify_server_default is not False:
kw["modify_server_default"] = self.modify_server_default
+ if self.modify_comment is not False:
+ kw["modify_comment"] = self.modify_comment
# TODO: make this a little simpler
all_keys = set(
table_name,
column_name,
nullable=None,
+ comment=False,
server_default=False,
new_column_name=None,
type_=None,
existing_type=None,
existing_server_default=False,
existing_nullable=None,
+ existing_comment=None,
schema=None,
**kw
):
existing_type=existing_type,
existing_server_default=existing_server_default,
existing_nullable=existing_nullable,
+ existing_comment=existing_comment,
modify_name=new_column_name,
modify_type=type_,
modify_server_default=server_default,
modify_nullable=nullable,
+ modify_comment=comment,
**kw
)
operations,
column_name,
nullable=None,
+ comment=False,
server_default=False,
new_column_name=None,
type_=None,
existing_type=None,
existing_server_default=False,
existing_nullable=None,
+ existing_comment=None,
**kw
):
"""Issue an "alter column" instruction using the current
existing_type=existing_type,
existing_server_default=existing_server_default,
existing_nullable=existing_nullable,
+ existing_comment=existing_comment,
modify_name=new_column_name,
modify_type=type_,
modify_server_default=server_default,
modify_nullable=nullable,
+ modify_comment=comment,
**kw
)
from . import ops
from .base import Operations
+from ..util import sqla_compat
@Operations.implementation_for(ops.AlterColumnOp)
server_default = operation.modify_server_default
new_column_name = operation.modify_name
nullable = operation.modify_nullable
+ comment = operation.modify_comment
+ existing_comment = operation.existing_comment
def _count_constraint(constraint):
return not isinstance(constraint, sa_schema.PrimaryKeyConstraint) and (
existing_type=existing_type,
existing_server_default=existing_server_default,
existing_nullable=existing_nullable,
+ comment=comment,
+ existing_comment=existing_comment,
**operation.kw
)
)
+@Operations.implementation_for(ops.CreateTableCommentOp)
+def create_table_comment(operations, operation):
+ table = operation.to_table(operations.migration_context)
+ operations.impl.create_table_comment(table)
+
+
+@Operations.implementation_for(ops.DropTableCommentOp)
+def drop_table_comment(operations, operation):
+ table = operation.to_table(operations.migration_context)
+ operations.impl.drop_table_comment(table)
+
+
@Operations.implementation_for(ops.AddColumnOp)
def add_column(operations, operation):
table_name = operation.table_name
for index in t.indexes:
operations.impl.create_index(index)
+ with_comment = (
+ sqla_compat._dialect_supports_comments(operations.impl.dialect)
+ and not operations.impl.dialect.inline_comments
+ )
+ comment = sqla_compat._comment_attribute(column)
+ if comment and with_comment:
+ operations.impl.create_column_comment(column)
+
@Operations.implementation_for(ops.AddConstraintOp)
def create_constraint(operations, operation):
from alembic import util
+from alembic.util import sqla_compat
from . import exclusions
if util.sqla_094:
"SQLAlchemy 1.1.0 or greater required",
)
+ @property
+ def sqlalchemy_1216(self):
+ return exclusions.skip_if(
+ lambda config: not util.sqla_1216,
+ "SQLAlchemy 1.2.16 or greater required",
+ )
+
@property
def pep3147(self):
return exclusions.only_if(lambda config: util.compat.has_pep3147())
+
+ @property
+ def comments(self):
+ return exclusions.only_if(
+ lambda config: sqla_compat._dialect_supports_comments(
+ config.db.dialect
+ )
+ )
+
+ @property
+ def comments_api(self):
+ return exclusions.only_if(lambda config: util.sqla_120)
from .sqla_compat import sqla_105 # noqa
from .sqla_compat import sqla_110 # noqa
from .sqla_compat import sqla_1115 # noqa
+from .sqla_compat import sqla_120 # noqa
+from .sqla_compat import sqla_1216 # noqa
if not sqla_09:
sqla_110 = _vers >= (1, 1, 0)
sqla_1014 = _vers >= (1, 0, 14)
sqla_1115 = _vers >= (1, 1, 15)
+sqla_120 = _vers >= (1, 2, 0)
+sqla_1216 = _vers >= (1, 2, 16)
if sqla_110:
return dialect.ddl_compiler(dialect, None)._prepared_index_name(idx)
+def _dialect_supports_comments(dialect):
+ if sqla_120:
+ return dialect.supports_comments
+ else:
+ return False
+
+
+def _comment_attribute(obj):
+ """return the .comment attribute from a Table or Column"""
+
+ if sqla_120:
+ return obj.comment
+ else:
+ return None
+
+
def _is_mariadb(mysql_dialect):
return "MariaDB" in mysql_dialect.server_version_info
--- /dev/null
+.. change::
+ :tags: feature, operations
+ :tickets: 422
+
+ Added Table and Column level comments for supported backends.
+ `create_table`, `add_column` and `alter_column` now all optionally
+ take `comment="X"` kwarg. Support for autogenerate for Table
+ and Column objects has also been added.
\ No newline at end of file
--- /dev/null
+import sys
+
+from sqlalchemy import Column
+from sqlalchemy import Float
+from sqlalchemy import MetaData
+from sqlalchemy import String
+from sqlalchemy import Table
+
+from alembic.testing import eq_
+from alembic.testing import mock
+from alembic.testing import TestBase
+from ._autogen_fixtures import AutogenFixtureTest
+
+py3k = sys.version_info.major >= 3
+
+
+class AutogenerateCommentsTest(AutogenFixtureTest, TestBase):
+ __backend__ = True
+
+ __requires__ = ("comments",)
+
+ def test_existing_table_comment_no_change(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table(
+ "some_table",
+ m1,
+ Column("test", String(10), primary_key=True),
+ comment="this is some table",
+ )
+
+ Table(
+ "some_table",
+ m2,
+ Column("test", String(10), primary_key=True),
+ comment="this is some table",
+ )
+
+ diffs = self._fixture(m1, m2)
+
+ eq_(diffs, [])
+
+ def test_add_table_comment(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table("some_table", m1, Column("test", String(10), primary_key=True))
+
+ Table(
+ "some_table",
+ m2,
+ Column("test", String(10), primary_key=True),
+ comment="this is some table",
+ )
+
+ diffs = self._fixture(m1, m2)
+
+ eq_(diffs[0][0], "add_table_comment")
+ eq_(diffs[0][1].comment, "this is some table")
+ eq_(diffs[0][2], None)
+
+ def test_remove_table_comment(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table(
+ "some_table",
+ m1,
+ Column("test", String(10), primary_key=True),
+ comment="this is some table",
+ )
+
+ Table("some_table", m2, Column("test", String(10), primary_key=True))
+
+ diffs = self._fixture(m1, m2)
+
+ eq_(diffs[0][0], "remove_table_comment")
+ eq_(diffs[0][1].comment, None)
+
+ def test_alter_table_comment(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table(
+ "some_table",
+ m1,
+ Column("test", String(10), primary_key=True),
+ comment="this is some table",
+ )
+
+ Table(
+ "some_table",
+ m2,
+ Column("test", String(10), primary_key=True),
+ comment="this is also some table",
+ )
+
+ diffs = self._fixture(m1, m2)
+
+ eq_(diffs[0][0], "add_table_comment")
+ eq_(diffs[0][1].comment, "this is also some table")
+ eq_(diffs[0][2], "this is some table")
+
+ def test_existing_column_comment_no_change(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table(
+ "some_table",
+ m1,
+ Column("test", String(10), primary_key=True),
+ Column("amount", Float, comment="the amount"),
+ )
+
+ Table(
+ "some_table",
+ m2,
+ Column("test", String(10), primary_key=True),
+ Column("amount", Float, comment="the amount"),
+ )
+
+ diffs = self._fixture(m1, m2)
+
+ eq_(diffs, [])
+
+ def test_add_column_comment(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table(
+ "some_table",
+ m1,
+ Column("test", String(10), primary_key=True),
+ Column("amount", Float),
+ )
+
+ Table(
+ "some_table",
+ m2,
+ Column("test", String(10), primary_key=True),
+ Column("amount", Float, comment="the amount"),
+ )
+
+ diffs = self._fixture(m1, m2)
+ eq_(
+ diffs,
+ [
+ [
+ (
+ "modify_comment",
+ None,
+ "some_table",
+ "amount",
+ {
+ "existing_nullable": True,
+ "existing_type": mock.ANY,
+ "existing_server_default": False,
+ },
+ None,
+ "the amount",
+ )
+ ]
+ ],
+ )
+
+ def test_remove_column_comment(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table(
+ "some_table",
+ m1,
+ Column("test", String(10), primary_key=True),
+ Column("amount", Float, comment="the amount"),
+ )
+
+ Table(
+ "some_table",
+ m2,
+ Column("test", String(10), primary_key=True),
+ Column("amount", Float),
+ )
+
+ diffs = self._fixture(m1, m2)
+ eq_(
+ diffs,
+ [
+ [
+ (
+ "modify_comment",
+ None,
+ "some_table",
+ "amount",
+ {
+ "existing_nullable": True,
+ "existing_type": mock.ANY,
+ "existing_server_default": False,
+ },
+ "the amount",
+ None,
+ )
+ ]
+ ],
+ )
+
+ def test_alter_column_comment(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table(
+ "some_table",
+ m1,
+ Column("test", String(10), primary_key=True),
+ Column("amount", Float, comment="the amount"),
+ )
+
+ Table(
+ "some_table",
+ m2,
+ Column("test", String(10), primary_key=True),
+ Column("amount", Float, comment="the adjusted amount"),
+ )
+
+ diffs = self._fixture(m1, m2)
+
+ eq_(
+ diffs,
+ [
+ [
+ (
+ "modify_comment",
+ None,
+ "some_table",
+ "amount",
+ {
+ "existing_nullable": True,
+ "existing_type": mock.ANY,
+ "existing_server_default": False,
+ },
+ "the amount",
+ "the adjusted amount",
+ )
+ ]
+ ],
+ )
batch_op.drop_index('pw_idx')
batch_op.drop_column('pw')
- # ### end Alembic commands ###""" # noqa,
+ # ### end Alembic commands ###""", # noqa,
)
eq_(
sa.ForeignKeyConstraint(['uid'], ['user.id'], )
)
op.drop_table('item')
- # ### end Alembic commands ###""" # noqa,
+ # ### end Alembic commands ###""", # noqa,
)
def test_imports_maintined(self):
"nullable=False)",
)
+ @config.requirements.comments_api
+ def test_render_col_with_comment(self):
+ c = Column("some_key", Integer, comment="This is a comment")
+ Table("some_table", MetaData(), c)
+ result = autogenerate.render._render_column(c, self.autogen_context)
+ eq_ignore_whitespace(
+ result,
+ "sa.Column('some_key', sa.Integer(), "
+ "nullable=True, "
+ "comment='This is a comment')",
+ )
+
def test_render_col_autoinc_false_mysql(self):
c = Column("some_key", Integer, primary_key=True, autoincrement=False)
Table("some_table", MetaData(), c)
op_obj,
)
+ @config.requirements.comments_api
+ def test_render_alter_column_modify_comment(self):
+ op_obj = ops.AlterColumnOp(
+ "sometable", "somecolumn", modify_comment="This is a comment"
+ )
+ eq_ignore_whitespace(
+ autogenerate.render_op_text(self.autogen_context, op_obj),
+ "op.alter_column('sometable', 'somecolumn', "
+ "comment='This is a comment')",
+ )
+
+ @config.requirements.comments_api
+ def test_render_alter_column_existing_comment(self):
+ op_obj = ops.AlterColumnOp(
+ "sometable", "somecolumn", existing_comment="This is a comment"
+ )
+ eq_ignore_whitespace(
+ autogenerate.render_op_text(self.autogen_context, op_obj),
+ "op.alter_column('sometable', 'somecolumn', "
+ "existing_comment='This is a comment')",
+ )
+
+ @config.requirements.comments_api
+ def test_render_col_drop_comment(self):
+ op_obj = ops.AlterColumnOp(
+ "sometable",
+ "somecolumn",
+ existing_comment="This is a comment",
+ modify_comment=None,
+ )
+ eq_ignore_whitespace(
+ autogenerate.render_op_text(self.autogen_context, op_obj),
+ "op.alter_column('sometable', 'somecolumn', "
+ "comment=None, "
+ "existing_comment='This is a comment')",
+ )
+
+ @config.requirements.comments_api
+ def test_render_table_with_comment(self):
+ m = MetaData()
+ t = Table(
+ "test",
+ m,
+ Column("id", Integer, primary_key=True),
+ Column("q", Integer, ForeignKey("address.id")),
+ comment="test comment",
+ )
+ op_obj = ops.CreateTableOp.from_table(t)
+ eq_ignore_whitespace(
+ autogenerate.render_op_text(self.autogen_context, op_obj),
+ "op.create_table('test',"
+ "sa.Column('id', sa.Integer(), nullable=False),"
+ "sa.Column('q', sa.Integer(), nullable=True),"
+ "sa.ForeignKeyConstraint(['q'], ['address.id'], ),"
+ "sa.PrimaryKeyConstraint('id'),"
+ "comment='test comment'"
+ ")",
+ )
+
+ @config.requirements.comments_api
+ def test_render_add_column_with_comment(self):
+ op_obj = ops.AddColumnOp(
+ "foo", Column("x", Integer, comment="This is a Column")
+ )
+ eq_ignore_whitespace(
+ autogenerate.render_op_text(self.autogen_context, op_obj),
+ "op.add_column('foo', sa.Column('x', sa.Integer(), "
+ "nullable=True, comment='This is a Column'))",
+ )
+
+ @config.requirements.comments_api
+ def test_render_create_table_comment_op(self):
+ op_obj = ops.CreateTableCommentOp("table_name", "comment")
+ eq_ignore_whitespace(
+ autogenerate.render_op_text(self.autogen_context, op_obj),
+ "op.create_table_comment("
+ " 'table_name',"
+ " 'comment',"
+ " existing_comment=None,"
+ " schema=None"
+ ")",
+ )
+
+ def test_render_create_table_comment_op_with_existing_comment(self):
+ op_obj = ops.CreateTableCommentOp(
+ "table_name", "comment", existing_comment="old comment"
+ )
+ eq_ignore_whitespace(
+ autogenerate.render_op_text(self.autogen_context, op_obj),
+ "op.create_table_comment("
+ " 'table_name',"
+ " 'comment',"
+ " existing_comment='old comment',"
+ " schema=None"
+ ")",
+ )
+
+ def test_render_create_table_comment_op_with_schema(self):
+ op_obj = ops.CreateTableCommentOp(
+ "table_name", "comment", schema="SomeSchema"
+ )
+ eq_ignore_whitespace(
+ autogenerate.render_op_text(self.autogen_context, op_obj),
+ "op.create_table_comment("
+ " 'table_name',"
+ " 'comment',"
+ " existing_comment=None,"
+ " schema='SomeSchema'"
+ ")",
+ )
+
+ def test_render_drop_table_comment_op(self):
+ op_obj = ops.DropTableCommentOp("table_name")
+ eq_ignore_whitespace(
+ autogenerate.render_op_text(self.autogen_context, op_obj),
+ "op.drop_table_comment("
+ " 'table_name',"
+ " existing_comment=None,"
+ " schema=None"
+ ")",
+ )
+
class RenderNamingConventionTest(TestBase):
__requires__ = ("sqlalchemy_094",)
with self._fixture() as batch:
batch.add_column(column)
- eq_(
- batch.impl.operations.impl.mock_calls,
- [mock.call.add_column("tname", column, schema=None)],
+ assert (
+ mock.call.add_column("tname", column, schema=None)
+ in batch.impl.operations.impl.mock_calls
)
def test_create_fk(self):
class MySQLOpTest(TestBase):
+ @config.requirements.comments_api
+ def test_create_table_with_comment(self):
+ context = op_fixture("mysql")
+ op.create_table(
+ "t2",
+ Column("c1", Integer, primary_key=True),
+ comment="This is a table comment",
+ )
+ context.assert_contains("COMMENT='This is a table comment'")
+
+ @config.requirements.comments_api
+ def test_create_table_with_column_comments(self):
+ context = op_fixture("mysql")
+ op.create_table(
+ "t2",
+ Column("c1", Integer, primary_key=True, comment="c1 comment"),
+ Column("c2", Integer, comment="c2 comment"),
+ comment="This is a table comment",
+ )
+
+ context.assert_(
+ "CREATE TABLE t2 "
+ "(c1 INTEGER NOT NULL COMMENT 'c1 comment' AUTO_INCREMENT, "
+ # TODO: why is there no space at the end here? is that on the
+ # SQLA side?
+ "c2 INTEGER COMMENT 'c2 comment', PRIMARY KEY (c1))"
+ "COMMENT='This is a table comment'"
+ )
+
+ @config.requirements.comments_api
+ def test_add_column_with_comment(self):
+ context = op_fixture("mysql")
+ op.add_column("t", Column("q", Integer, comment="This is a comment"))
+ context.assert_(
+ "ALTER TABLE t ADD COLUMN q INTEGER COMMENT 'This is a comment'"
+ )
+
def test_rename_column(self):
context = op_fixture("mysql")
op.alter_column(
server_default="q",
)
+ @config.requirements.comments_api
+ def test_alter_column_add_comment(self):
+ context = op_fixture("mysql")
+ op.alter_column(
+ "t1",
+ "c1",
+ comment="This is a column comment",
+ existing_type=Boolean(),
+ schema="foo",
+ )
+
+ context.assert_(
+ "ALTER TABLE foo.t1 MODIFY c1 BOOL NULL "
+ "COMMENT 'This is a column comment'"
+ )
+
+ @config.requirements.comments_api
+ def test_alter_column_add_comment_quoting(self):
+ context = op_fixture("mysql")
+ op.alter_column(
+ "t1",
+ "c1",
+ comment="This is a 'column' comment",
+ existing_type=Boolean(),
+ schema="foo",
+ )
+
+ context.assert_(
+ "ALTER TABLE foo.t1 MODIFY c1 BOOL NULL "
+ "COMMENT 'This is a ''column'' comment'"
+ )
+
+ @config.requirements.comments_api
+ def test_alter_column_drop_comment(self):
+ context = op_fixture("mysql")
+ op.alter_column(
+ "t",
+ "c",
+ existing_type=Boolean(),
+ schema="foo",
+ comment=None,
+ existing_comment="This is a column comment",
+ )
+
+ context.assert_("ALTER TABLE foo.t MODIFY c BOOL NULL")
+
+ @config.requirements.comments_api
+ def test_alter_column_existing_comment(self):
+ context = op_fixture("mysql")
+ op.alter_column(
+ "t1",
+ "c1",
+ nullable=False,
+ existing_comment="existing column comment",
+ existing_type=Integer,
+ )
+
+ context.assert_(
+ "ALTER TABLE t1 MODIFY c1 INTEGER NOT NULL "
+ "COMMENT 'existing column comment'"
+ )
+
+ @config.requirements.comments_api
+ def test_alter_column_new_comment_replaces_existing(self):
+ context = op_fixture("mysql")
+ op.alter_column(
+ "t1",
+ "c1",
+ nullable=False,
+ comment="This is a column comment",
+ existing_comment="existing column comment",
+ existing_type=Integer,
+ )
+
+ context.assert_(
+ "ALTER TABLE t1 MODIFY c1 INTEGER NOT NULL "
+ "COMMENT 'This is a column comment'"
+ )
+
+ @config.requirements.comments_api
+ def test_create_table_comment(self):
+ # this is handled by SQLAlchemy's compilers
+ context = op_fixture("mysql")
+ op.create_table_comment("t2", comment="t2 table", schema="foo")
+ context.assert_("ALTER TABLE foo.t2 COMMENT 't2 table'")
+
+ @config.requirements.comments_api
+ @config.requirements.sqlalchemy_1216
+ def test_drop_table_comment(self):
+ # this is handled by SQLAlchemy's compilers
+ context = op_fixture("mysql")
+ op.drop_table_comment("t2", existing_comment="t2 table", schema="foo")
+ context.assert_("ALTER TABLE foo.t2 COMMENT ''")
+
def test_drop_fk(self):
context = op_fixture("mysql")
op.drop_constraint("f1", "t1", "foreignkey")
op.drop_index("ik_test", tablename="t1")
context.assert_("DROP INDEX ik_test ON t1")
+ @config.requirements.comments
+ def test_create_table_comment_op(self):
+ context = op_fixture()
+
+ op.create_table_comment("some_table", "table comment")
+
+ context.assert_("COMMENT ON TABLE some_table IS 'table comment'")
+
+ @config.requirements.comments
+ def test_drop_table_comment_op(self):
+ context = op_fixture()
+
+ op.drop_table_comment("some_table")
+
+ context.assert_("COMMENT ON TABLE some_table IS NULL")
+
class SQLModeOpTest(TestBase):
def test_auto_literals(self):
from alembic import command
from alembic import op
+from alembic.testing import config
from alembic.testing.env import _no_sql_testing_config
from alembic.testing.env import clear_staging_env
from alembic.testing.env import staging_env
)
context.assert_("ALTER TABLE t1 ADD c1 INTEGER DEFAULT '12' NOT NULL")
+ @config.requirements.comments
+ def test_add_column_with_comment(self):
+ context = op_fixture("oracle")
+ op.add_column(
+ "t1", Column("c1", Integer, nullable=False, comment="c1 comment")
+ )
+ context.assert_(
+ "ALTER TABLE t1 ADD c1 INTEGER NOT NULL",
+ "COMMENT ON COLUMN t1.c1 IS 'c1 comment'",
+ )
+
def test_alter_column_rename_oracle(self):
context = op_fixture("oracle")
op.alter_column("t", "c", name="x")
op.alter_column("t", "c", type_=Integer)
context.assert_("ALTER TABLE t MODIFY c INTEGER")
+ def test_alter_column_add_comment(self):
+ context = op_fixture("oracle")
+ op.alter_column("t", "c", type_=Integer, comment="c comment")
+ context.assert_(
+ "ALTER TABLE t MODIFY c INTEGER",
+ "COMMENT ON COLUMN t.c IS 'c comment'",
+ )
+
+ def test_alter_column_add_comment_quotes(self):
+ context = op_fixture("oracle")
+ op.alter_column("t", "c", type_=Integer, comment="c 'comment'")
+ context.assert_(
+ "ALTER TABLE t MODIFY c INTEGER",
+ "COMMENT ON COLUMN t.c IS 'c ''comment'''",
+ )
+
+ def test_alter_column_drop_comment(self):
+ context = op_fixture("oracle")
+ op.alter_column("t", "c", type_=Integer, comment=None)
+ context.assert_(
+ "ALTER TABLE t MODIFY c INTEGER", "COMMENT ON COLUMN t.c IS ''"
+ )
+
+ @config.requirements.comments_api
+ def test_create_table_comment(self):
+ # this is handled by SQLAlchemy's compilers
+ context = op_fixture("oracle")
+ op.create_table_comment(
+ 't2',
+ comment='t2 table',
+ schema='foo'
+ )
+ context.assert_(
+ "COMMENT ON TABLE foo.t2 IS 't2 table'"
+ )
+
+ @config.requirements.comments_api
+ @config.requirements.sqlalchemy_1216
+ def test_drop_table_comment(self):
+ # this is handled by SQLAlchemy's compilers
+ context = op_fixture("oracle")
+ op.drop_table_comment(
+ 't2',
+ existing_comment='t2 table',
+ schema='foo'
+ )
+ context.assert_(
+ "COMMENT ON TABLE foo.t2 IS ''"
+ )
+
def test_drop_index(self):
context = op_fixture("oracle")
op.drop_index("my_idx", "my_table")
"ALTER TABLE t RENAME COLUMN c TO c2",
)
+ @config.requirements.comments
+ def test_create_table_with_column_comments(self):
+ context = op_fixture("oracle")
+ op.create_table(
+ "t2", Column("c1", Integer, primary_key=True), comment="t2 comment"
+ )
+ context.assert_(
+ "CREATE TABLE t2 (c1 INTEGER NOT NULL, PRIMARY KEY (c1))",
+ "COMMENT ON TABLE t2 IS 't2 comment'",
+ )
+
# TODO: when we add schema support
# def test_alter_column_rename_oracle_schema(self):
# context = op_fixture('oracle')
'USING gist ("SomeColumn" WITH >) WHERE ("SomeColumn" > 5)'
)
+ @config.requirements.comments_api
+ def test_add_column_with_comment(self):
+ context = op_fixture("postgresql")
+ op.add_column("t", Column("q", Integer, comment="This is a comment"))
+ context.assert_(
+ "ALTER TABLE t ADD COLUMN q INTEGER",
+ "COMMENT ON COLUMN t.q IS 'This is a comment'",
+ )
+
+ @config.requirements.comments_api
+ def test_alter_column_with_comment(self):
+ context = op_fixture("postgresql")
+ op.alter_column(
+ "t",
+ "c",
+ nullable=False,
+ existing_type=Boolean(),
+ schema="foo",
+ comment="This is a column comment",
+ )
+
+ context.assert_(
+ "ALTER TABLE foo.t ALTER COLUMN c SET NOT NULL",
+ "COMMENT ON COLUMN t.c IS 'This is a column comment'",
+ )
+
+ @config.requirements.comments_api
+ def test_alter_column_add_comment(self):
+ context = op_fixture("postgresql")
+ op.alter_column(
+ "t",
+ "c",
+ existing_type=Boolean(),
+ schema="foo",
+ comment="This is a column comment",
+ )
+
+ context.assert_("COMMENT ON COLUMN t.c IS 'This is a column comment'")
+
+ @config.requirements.comments_api
+ def test_alter_column_add_comment_quoting(self):
+ context = op_fixture("postgresql")
+ op.alter_column(
+ "t",
+ "c",
+ existing_type=Boolean(),
+ schema="foo",
+ comment="This is a column 'comment'",
+ )
+
+ context.assert_(
+ "COMMENT ON COLUMN t.c IS 'This is a column ''comment'''"
+ )
+
+ @config.requirements.comments_api
+ def test_alter_column_drop_comment(self):
+ context = op_fixture("postgresql")
+ op.alter_column(
+ "t",
+ "c",
+ existing_type=Boolean(),
+ schema="foo",
+ comment=None,
+ existing_comment="This is a column comment",
+ )
+
+ context.assert_("COMMENT ON COLUMN t.c IS NULL")
+
+ @config.requirements.comments_api
+ def test_create_table_with_comment(self):
+ context = op_fixture("postgresql")
+ op.create_table(
+ "t2",
+ Column("c1", Integer, primary_key=True),
+ Column("c2", Integer),
+ comment="t2 comment",
+ )
+ context.assert_(
+ "CREATE TABLE t2 (c1 SERIAL NOT NULL, "
+ "c2 INTEGER, PRIMARY KEY (c1))",
+ "COMMENT ON TABLE t2 IS 't2 comment'",
+ )
+
+ @config.requirements.comments_api
+ def test_create_table_with_column_comments(self):
+ context = op_fixture("postgresql")
+ op.create_table(
+ "t2",
+ Column("c1", Integer, primary_key=True, comment="c1 comment"),
+ Column("c2", Integer, comment="c2 comment"),
+ comment="t2 comment",
+ )
+ context.assert_(
+ "CREATE TABLE t2 (c1 SERIAL NOT NULL, "
+ "c2 INTEGER, PRIMARY KEY (c1))",
+ "COMMENT ON TABLE t2 IS 't2 comment'",
+ "COMMENT ON COLUMN t2.c1 IS 'c1 comment'",
+ "COMMENT ON COLUMN t2.c2 IS 'c2 comment'",
+ )
+
+ @config.requirements.comments_api
+ def test_create_table_comment(self):
+ # this is handled by SQLAlchemy's compilers
+ context = op_fixture("postgresql")
+ op.create_table_comment(
+ 't2',
+ comment='t2 table',
+ schema='foo'
+ )
+ context.assert_(
+ "COMMENT ON TABLE foo.t2 IS 't2 table'"
+ )
+
+ @config.requirements.comments_api
+ def test_drop_table_comment(self):
+ # this is handled by SQLAlchemy's compilers
+ context = op_fixture("postgresql")
+ op.drop_table_comment(
+ 't2',
+ existing_comment='t2 table',
+ schema='foo'
+ )
+ context.assert_(
+ "COMMENT ON TABLE foo.t2 IS NULL"
+ )
+
class PGOfflineEnumTest(TestBase):
def setUp(self):
Revision("e", ("d",), branch_labels=["xy1"]),
Revision("f", ("e",)),
]
+
assert_raises_message(
RevisionError,
r"Branch name 'xy1' in revision (?:e|c) already "
Revision("c2", ("b",)),
Revision("d", ("c1", "c2")),
]
+
map_ = RevisionMap(fn)
c1 = map_.get_revision("c1")
c2 = map_.get_revision("c2")
from alembic import op
from alembic.testing import assert_raises_message
+from alembic.testing import config
from alembic.testing.fixtures import op_fixture
from alembic.testing.fixtures import TestBase
"foo",
"sometable",
)
+
+ @config.requirements.comments
+ def test_create_table_with_comment_ignored(self):
+
+ context = op_fixture("sqlite")
+ op.create_table(
+ "t2",
+ Column("c1", Integer, primary_key=True),
+ Column("c2", Integer),
+ comment="This is a table comment",
+ )
+ context.assert_(
+ "CREATE TABLE t2 (c1 INTEGER NOT NULL, "
+ "c2 INTEGER, PRIMARY KEY (c1))"
+ )
+
+ @config.requirements.comments
+ def test_add_column_with_comment_ignored(self):
+
+ context = op_fixture("sqlite")
+ op.add_column("t1", Column("c1", Integer, comment="c1 comment"))
+ context.assert_("ALTER TABLE t1 ADD COLUMN c1 INTEGER")
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
