def compare_metadata(context, metadata):
"""Compare a database schema to that given in a :class:`~sqlalchemy.schema.MetaData`
instance.
-
+
The database connection is presented in the context
- of a :class:`.MigrationContext` object, which
+ of a :class:`.MigrationContext` object, which
provides database connectivity as well as optional
comparison functions to use for datatypes and
server defaults - see the "autogenerate" arguments
at :meth:`.EnvironmentContext.configure`
for details on these.
-
+
The return format is a list of "diff" directives,
each representing individual differences::
from alembic.autogenerate import compare_metadata
from sqlalchemy.schema import SchemaItem
from sqlalchemy.types import TypeEngine
- from sqlalchemy import (create_engine, MetaData, Column,
+ from sqlalchemy import (create_engine, MetaData, Column,
Integer, String, Table)
import pprint
)''')
metadata = MetaData()
- Table('foo', metadata,
+ Table('foo', metadata,
Column('id', Integer, primary_key=True),
Column('data', Integer),
Column('x', Integer, nullable=False)
)
- Table('bat', metadata,
+ Table('bat', metadata,
Column('info', String)
)
diff = compare_metadata(mc, metadata)
pprint.pprint(diff, indent=2, width=20)
-
+
Output::
-
+
[ ( 'add_table',
Table('bat', MetaData(bind=None), Column('info', String(), table=<bat>), schema=None)),
( 'remove_table',
'existing_type': INTEGER()},
True,
False)]]
-
-
+
+
:param context: a :class:`.MigrationContext`
instance.
- :param metadata: a :class:`~sqlalchemy.schema.MetaData`
+ :param metadata: a :class:`~sqlalchemy.schema.MetaData`
instance.
-
+
"""
autogen_context, connection = _autogen_context(context, None)
diffs = []
_compare_tables(conn_table_names, metadata_table_names,
inspector, metadata, diffs, autogen_context)
-def _compare_tables(conn_table_names, metadata_table_names,
+def _compare_tables(conn_table_names, metadata_table_names,
inspector, metadata, diffs, autogen_context):
for tname in metadata_table_names.difference(conn_table_names):
diffs.append(("add_table", metadata.tables[tname]))
existing_tables = conn_table_names.intersection(metadata_table_names)
conn_column_info = dict(
- (tname,
+ (tname,
dict(
(rec["name"], rec)
for rec in inspector.get_columns(tname)
)
for tname in sorted(existing_tables):
- _compare_columns(tname,
- conn_column_info[tname],
+ _compare_columns(tname,
+ conn_column_info[tname],
metadata.tables[tname],
diffs, autogen_context)
- # TODO:
+ # TODO:
# index add/drop
# table constraints
# sequences
if col_diff:
diffs.append(col_diff)
-def _compare_nullable(tname, cname, conn_col,
- metadata_col_nullable, diffs,
+def _compare_nullable(tname, cname, conn_col,
+ metadata_col_nullable, diffs,
autogen_context):
conn_col_nullable = conn_col['nullable']
if conn_col_nullable is not metadata_col_nullable:
diffs.append(
- ("modify_nullable", tname, cname,
+ ("modify_nullable", tname, cname,
{
"existing_type":conn_col['type'],
"existing_server_default":conn_col['default'],
},
- conn_col_nullable,
+ conn_col_nullable,
metadata_col_nullable),
)
- log.info("Detected %s on column '%s.%s'",
+ log.info("Detected %s on column '%s.%s'",
"NULL" if metadata_col_nullable else "NOT NULL",
tname,
cname
)
-def _compare_type(tname, cname, conn_col,
- metadata_col, diffs,
+def _compare_type(tname, cname, conn_col,
+ metadata_col, diffs,
autogen_context):
conn_type = conn_col['type']
if isdiff:
diffs.append(
- ("modify_type", tname, cname,
+ ("modify_type", tname, cname,
{
"existing_nullable":conn_col['nullable'],
"existing_server_default":conn_col['default'],
},
- conn_type,
+ conn_type,
metadata_type),
)
- log.info("Detected type change from %r to %r on '%s.%s'",
+ log.info("Detected type change from %r to %r on '%s.%s'",
conn_type, metadata_type, tname, cname
)
-def _compare_server_default(tname, cname, conn_col, metadata_col,
+def _compare_server_default(tname, cname, conn_col, metadata_col,
diffs, autogen_context):
metadata_default = metadata_col.server_default
if isdiff:
conn_col_default = conn_col['default']
diffs.append(
- ("modify_default", tname, cname,
+ ("modify_default", tname, cname,
{
"existing_nullable":conn_col['nullable'],
"existing_type":conn_col['type'],
conn_col_default,
metadata_default),
)
- log.info("Detected server default on column '%s.%s'",
+ log.info("Detected server default on column '%s.%s'",
tname,
cname
)
'prefix':_alembic_autogenerate_prefix(autogen_context),
'args':',\n'.join(
[_render_column(col, autogen_context) for col in table.c] +
- sorted([rcons for rcons in
- [_render_constraint(cons, autogen_context) for cons in
+ sorted([rcons for rcons in
+ [_render_constraint(cons, autogen_context) for cons in
table.constraints]
if rcons is not None
])
"cname":column.name
}
-def _modify_col(tname, cname,
+def _modify_col(tname, cname,
autogen_context,
server_default=False,
type_=None,
sqla_prefix = _sqlalchemy_autogenerate_prefix(autogen_context)
indent = " " * 11
text = "%(prefix)salter_column(%(tname)r, %(cname)r" % {
- 'prefix':_alembic_autogenerate_prefix(autogen_context),
- 'tname':tname,
+ 'prefix':_alembic_autogenerate_prefix(autogen_context),
+ 'tname':tname,
'cname':cname}
- text += ", \n%sexisting_type=%s" % (indent,
+ text += ",\n%sexisting_type=%s" % (indent,
_repr_type(sqla_prefix, existing_type, autogen_context))
if server_default is not False:
- text += ", \n%sserver_default=%s" % (indent,
+ text += ",\n%sserver_default=%s" % (indent,
_render_server_default(server_default, autogen_context),)
if type_ is not None:
- text += ", \n%stype_=%s" % (indent, _repr_type(sqla_prefix, type_, autogen_context))
+ text += ",\n%stype_=%s" % (indent, _repr_type(sqla_prefix, type_, autogen_context))
if nullable is not None:
- text += ", \n%snullable=%r" % (
+ text += ",\n%snullable=%r" % (
indent, nullable,)
if existing_nullable is not None:
- text += ", \n%sexisting_nullable=%r" % (
+ text += ",\n%sexisting_nullable=%r" % (
indent, existing_nullable)
if existing_server_default:
- text += ", \n%sexisting_server_default=%s" % (
- indent,
+ text += ",\n%sexisting_server_default=%s" % (
+ indent,
_render_server_default(
- existing_server_default,
+ existing_server_default,
autogen_context),
)
text += ")"
def _render_column(column, autogen_context):
opts = []
if column.server_default:
- opts.append(("server_default",
+ opts.append(("server_default",
_render_server_default(column.server_default, autogen_context)))
if column.nullable is not None:
opts.append(("nullable", column.nullable))
else:
default = str(default.arg.compile(dialect=autogen_context['dialect']))
if isinstance(default, basestring):
- # TODO: this is just a hack to get
+ # TODO: this is just a hack to get
# tests to pass until we figure out
# WTF sqlite is doing
default = re.sub(r"^'|'$", "", default)
Table('order', m,
Column('order_id', Integer, primary_key=True),
- Column("amount", Numeric(8, 2), nullable=False,
+ Column("amount", Numeric(8, 2), nullable=False,
server_default="0"),
)
Table('order', m,
Column('order_id', Integer, primary_key=True),
- Column("amount", Numeric(10, 2), nullable=True,
+ Column("amount", Numeric(10, 2), nullable=True,
server_default="0"),
Column('user_id', Integer, ForeignKey('user.id')),
)
- Table('item', m,
+ Table('item', m,
Column('id', Integer, primary_key=True),
Column('description', String(100)),
Column('order_id', Integer, ForeignKey('order.order_id')),
metadata = self.m2
connection = self.context.bind
diffs = []
- autogenerate._produce_net_changes(connection, metadata, diffs,
+ autogenerate._produce_net_changes(connection, metadata, diffs,
self.autogen_context)
eq_(
op.drop_table('extra')
op.add_column('address', sa.Column('street', sa.String(length=50), nullable=True))
op.add_column('order', sa.Column('user_id', sa.Integer(), nullable=True))
- op.alter_column('order', 'amount',
- existing_type=sa.NUMERIC(precision=8, scale=2),
- type_=sa.Numeric(precision=10, scale=2),
- nullable=True,
+ op.alter_column('order', 'amount',
+ existing_type=sa.NUMERIC(precision=8, scale=2),
+ type_=sa.Numeric(precision=10, scale=2),
+ nullable=True,
existing_server_default='0')
op.drop_column('user', 'pw')
- op.alter_column('user', 'a1',
- existing_type=sa.TEXT(),
- server_default='x',
+ op.alter_column('user', 'a1',
+ existing_type=sa.TEXT(),
+ server_default='x',
existing_nullable=True)
- op.alter_column('user', 'name',
- existing_type=sa.VARCHAR(length=50),
+ op.alter_column('user', 'name',
+ existing_type=sa.VARCHAR(length=50),
nullable=False)
### end Alembic commands ###""")
eq_(re.sub(r"u'", "'", template_args['downgrades']),
"""### commands auto generated by Alembic - please adjust! ###
- op.alter_column('user', 'name',
- existing_type=sa.VARCHAR(length=50),
+ op.alter_column('user', 'name',
+ existing_type=sa.VARCHAR(length=50),
nullable=True)
- op.alter_column('user', 'a1',
- existing_type=sa.TEXT(),
- server_default=None,
+ op.alter_column('user', 'a1',
+ existing_type=sa.TEXT(),
+ server_default=None,
existing_nullable=True)
op.add_column('user', sa.Column('pw', sa.VARCHAR(length=50), nullable=True))
- op.alter_column('order', 'amount',
- existing_type=sa.Numeric(precision=10, scale=2),
- type_=sa.NUMERIC(precision=8, scale=2),
- nullable=False,
+ op.alter_column('order', 'amount',
+ existing_type=sa.Numeric(precision=10, scale=2),
+ type_=sa.NUMERIC(precision=8, scale=2),
+ nullable=False,
existing_server_default='0')
op.drop_column('order', 'user_id')
op.drop_column('address', 'street')
def test_skip_null_type_comparison_reflected(self):
diff = []
autogenerate._compare_type("sometable", "somecol",
- {"name":"somecol", "type":NULLTYPE,
+ {"name":"somecol", "type":NULLTYPE,
"nullable":True, "default":None},
Column("somecol", Integer()),
diff, self.autogen_context
def test_skip_null_type_comparison_local(self):
diff = []
autogenerate._compare_type("sometable", "somecol",
- {"name":"somecol", "type":Integer(),
+ {"name":"somecol", "type":Integer(),
"nullable":True, "default":None},
Column("somecol", NULLTYPE),
diff, self.autogen_context
diff = []
autogenerate._compare_type("sometable", "somecol",
- {"name":"somecol", "type":Integer(),
+ {"name":"somecol", "type":Integer(),
"nullable":True, "default":None},
Column("somecol", MyType()),
diff, self.autogen_context
from sqlalchemy.util import OrderedSet
inspector = Inspector.from_engine(self.bind)
autogenerate._compare_tables(
- OrderedSet(['extra', 'user']), OrderedSet(), inspector,
+ OrderedSet(['extra', 'user']), OrderedSet(), inspector,
MetaData(), diffs, self.autogen_context
)
eq_(
connection = self.empty_context.bind
diffs = []
- autogenerate._produce_net_changes(connection, metadata, diffs,
+ autogenerate._produce_net_changes(connection, metadata, diffs,
self.autogen_empty_context)
eq_(diffs[0][0], 'add_table')
def test_render_drop_table(self):
eq_(
- autogenerate._drop_table(Table("sometable", MetaData()),
+ autogenerate._drop_table(Table("sometable", MetaData()),
self.autogen_context),
"op.drop_table('sometable')"
)
def test_render_add_column(self):
eq_(
autogenerate._add_column(
- "foo", Column("x", Integer, server_default="5"),
+ "foo", Column("x", Integer, server_default="5"),
self.autogen_context),
"op.add_column('foo', sa.Column('x', sa.Integer(), "
"server_default='5', nullable=True))"
def test_render_drop_column(self):
eq_(
autogenerate._drop_column(
- "foo", Column("x", Integer, server_default="5"),
+ "foo", Column("x", Integer, server_default="5"),
self.autogen_context),
"op.drop_column('foo', 'x')"
def test_render_modify_type(self):
eq_ignore_whitespace(
autogenerate._modify_col(
- "sometable", "somecolumn",
+ "sometable", "somecolumn",
self.autogen_context,
type_=CHAR(10), existing_type=CHAR(20)),
"op.alter_column('sometable', 'somecolumn', "
def test_render_modify_nullable(self):
eq_ignore_whitespace(
autogenerate._modify_col(
- "sometable", "somecolumn",
+ "sometable", "somecolumn",
self.autogen_context,
existing_type=Integer(),
nullable=True),
def test_render_modify_nullable_w_default(self):
eq_ignore_whitespace(
autogenerate._modify_col(
- "sometable", "somecolumn",
+ "sometable", "somecolumn",
self.autogen_context,
existing_type=Integer(),
existing_server_default="5",