log = logging.getLogger(__name__)
+try:
+ from sqlalchemy.sql.naming import conv
+ def _render_gen_name(autogen_context, name):
+ if isinstance(name, conv):
+ return _f_name(_alembic_autogenerate_prefix(autogen_context), name)
+ else:
+ return name
+except ImportError:
+ def _render_gen_name(autogen_context, name):
+ return name
+
+class _f_name(object):
+ def __init__(self, prefix, name):
+ self.prefix = prefix
+ self.name = name
+
+ def __repr__(self):
+ return "%sf(%r)" % (self.prefix, self.name)
+
def _render_potential_expr(value, autogen_context):
if isinstance(value, sql.ClauseElement):
if compat.sqla_08:
text = "%(prefix)screate_index('%(name)s', '%(table)s', %(columns)s, "\
"unique=%(unique)r%(schema)s%(kwargs)s)" % {
'prefix': _alembic_autogenerate_prefix(autogen_context),
- 'name': index.name,
+ 'name': _render_gen_name(autogen_context, index.name),
'table': index.table.name,
'columns': _get_index_column_names(index),
'unique': index.unique or False,
text = "%(prefix)sdrop_index('%(name)s', "\
"table_name='%(table_name)s'%(schema)s)" % {
'prefix': _alembic_autogenerate_prefix(autogen_context),
- 'name': index.name,
+ 'name': _render_gen_name(autogen_context, index.name),
'table_name': index.table.name,
'schema': ((", schema='%s'" % index.table.schema)
if index.table.schema else '')
if alter and constraint.table.schema:
opts.append(("schema", str(constraint.table.schema)))
if not alter and constraint.name:
- opts.append(("name", constraint.name))
+ opts.append(("name", _render_gen_name(autogen_context, constraint.name)))
if alter:
- args = [repr(constraint.name), repr(constraint.table.name)]
+ args = [repr(_render_gen_name(autogen_context, constraint.name)),
+ repr(constraint.table.name)]
args.append(repr([col.name for col in constraint.columns]))
args.extend(["%s=%r" % (k, v) for k, v in opts])
return "%(prefix)screate_unique_constraint(%(args)s)" % {
"""
text = "%(prefix)sdrop_constraint(%(name)r, '%(table_name)s'%(schema)s)" % {
'prefix': _alembic_autogenerate_prefix(autogen_context),
- 'name': constraint.name,
+ 'name': _render_gen_name(autogen_context, constraint.name),
'table_name': constraint.table.name,
'schema': (", schema='%s'" % constraint.table.schema)
if constraint.table.schema else '',
opts = []
if constraint.name:
- opts.append(("name", repr(constraint.name)))
+ opts.append(("name", repr(_render_gen_name(autogen_context, constraint.name))))
return "%(prefix)sPrimaryKeyConstraint(%(args)s)" % {
"prefix": _sqlalchemy_autogenerate_prefix(autogen_context),
"args": ", ".join(
opts = []
if constraint.name:
- opts.append(("name", repr(constraint.name)))
+ opts.append(("name", repr(_render_gen_name(autogen_context, constraint.name))))
if constraint.onupdate:
opts.append(("onupdate", repr(constraint.onupdate)))
if constraint.ondelete:
return None
opts = []
if constraint.name:
- opts.append(("name", repr(constraint.name)))
+ opts.append(("name", repr(_render_gen_name(autogen_context, constraint.name))))
return "%(prefix)sCheckConstraint(%(sqltext)r%(opts)s)" % {
"prefix": _sqlalchemy_autogenerate_prefix(autogen_context),
"opts": ", " + (", ".join("%s=%s" % (k, v)
__all__ = ('Operations',)
+try:
+ from sqlalchemy.sql.naming import conv
+except:
+ conv = None
+
class Operations(object):
"""Define high level migration operations.
def _primary_key_constraint(self, name, table_name, cols, schema=None):
- m = sa_schema.MetaData()
+ m = self._metadata()
columns = [sa_schema.Column(n, NULLTYPE) for n in cols]
t1 = sa_schema.Table(table_name, m,
*columns,
onupdate=None, ondelete=None,
deferrable=None, source_schema=None,
referent_schema=None):
- m = sa_schema.MetaData()
+ m = self._metadata()
if source == referent:
t1_cols = local_cols + remote_cols
else:
return f
def _unique_constraint(self, name, source, local_cols, schema=None, **kw):
- t = sa_schema.Table(source, sa_schema.MetaData(),
+ t = sa_schema.Table(source, self._metadata(),
*[sa_schema.Column(n, NULLTYPE) for n in local_cols],
schema=schema)
kw['name'] = name
return uq
def _check_constraint(self, name, source, condition, schema=None, **kw):
- t = sa_schema.Table(source, sa_schema.MetaData(),
+ t = sa_schema.Table(source, self._metadata(),
sa_schema.Column('x', Integer), schema=schema)
ck = sa_schema.CheckConstraint(condition, name=name, **kw)
t.append_constraint(ck)
return ck
+ def _metadata(self):
+ kw = {}
+ if 'target_metadata' in self.migration_context.opts:
+ mt = self.migration_context.opts['target_metadata']
+ if hasattr(mt, 'naming_convention'):
+ kw['naming_convention'] = mt.naming_convention
+ return sa_schema.MetaData(**kw)
+
def _table(self, name, *columns, **kw):
- m = sa_schema.MetaData()
+ m = self._metadata()
t = sa_schema.Table(name, m, *columns, **kw)
for f in t.foreign_keys:
self._ensure_table_for_fk(m, f)
return sa_schema.Column(name, type_, **kw)
def _index(self, name, tablename, columns, schema=None, **kw):
- t = sa_schema.Table(tablename or 'no_table', sa_schema.MetaData(),
+ t = sa_schema.Table(tablename or 'no_table', self._metadata(),
*[sa_schema.Column(n, NULLTYPE) for n in columns],
schema=schema
)
if _count_constraint(constraint):
self.impl.add_constraint(constraint)
+ def f(self, name):
+ """Indicate a string name that has already had a naming convention
+ applied to it.
+
+ This feature combines with the SQLAlchemy ``naming_convention`` feature
+ to disambiguate constraint names that have already had naming
+ conventions applied to them, versus those that have not. This is
+ necessary in the case that the ``"%(constraint_name)s"`` token
+ is used within a naming convention, so that it can be identified
+ that this particular name should remain fixed.
+
+ If the :meth:`.Operations.f` is used on a constraint, the naming
+ convention will not take effect::
+
+ op.add_column('t', 'x', Boolean(name=op.f('ck_bool_t_x')))
+
+ Above, the CHECK constraint generated will have the name ``ck_bool_t_x``
+ regardless of whether or not a naming convention is in use.
+
+ Alternatively, if a naming convention is in use, and 'f' is not used,
+ names will be converted along conventions. If the ``target_metadata``
+ contains the naming convention
+ ``{"ck": "ck_bool_%(table_name)s_%(constraint_name)s"}``, then the
+ output of the following:
+
+ op.add_column('t', 'x', Boolean(name='x'))
+
+ will be::
+
+ CONSTRAINT ck_bool_t_x CHECK (x in (1, 0)))
+
+ The function is rendered in the output of autogenerate when
+ a particular constraint name is already converted, for SQLAlchemy
+ version **0.9.4 and greater only**. Even though ``naming_convention``
+ was introduced in 0.9.2, the string disambiguation service is new
+ as of 0.9.4.
+
+ .. versionadded:: 0.6.4
+
+ """
+ if conv:
+ return conv(name)
+ else:
+ raise NotImplementedError(
+ "op.f() feature requires SQLAlchemy 0.9.4 or greater.")
+
def add_column(self, table_name, column, schema=None):
"""Issue an "add column" instruction using the current
migration context.
sqla_08 = _vers >= (0, 8, 0, 'b2')
sqla_09 = _vers >= (0, 9, 0)
sqla_092 = _vers >= (0, 9, 2)
+sqla_094 = _vers >= (0, 9, 4)
if not sqla_07:
raise CommandError(
"SQLAlchemy 0.7.3 or greater is required. ")
.. changelog::
:version: 0.6.4
+ .. change::
+ :tags: bug
+ :tickets: 183
+
+ Extensive changes have been made to more fully support SQLAlchemy's new
+ naming conventions feature. Note that while SQLAlchemy has added this
+ feature as of 0.9.2, some additional fixes in 0.9.4 are needed to
+ resolve some of the issues:
+
+ 1. The :class:`.Operations` object now takes into account the naming
+ conventions that are present on the :class:`.MetaData` object that's
+ associated using :paramref:`~.EnvironmentContext.configure.target_metadata`.
+ When :class:`.Operations` renders a constraint directive like
+ ``ADD CONSTRAINT``, it now will make use of this naming convention
+ when it produces its own temporary :class:`.MetaData` object.
+
+ 2. Note however that the autogenerate feature in most cases generates
+ constraints like foreign keys and unique constraints with the
+ final names intact; the only exception are the constraints implicit
+ with a schema-type like Boolean or Enum. In most of these cases,
+ the naming convention feature will not take effect for these constraints
+ and will instead use the given name as is, with one exception....
+
+ 3. Naming conventions which use the ``"%(constraint_name)s"`` token, that
+ is, produce a new name that uses the original name as a component,
+ will still be pulled into the naming convention converter and be
+ converted. The problem arises when autogenerate renders a constraint
+ with it's already-generated name present in the migration file's source
+ code, the name will be doubled up at render time due to the combination
+ of #1 and #2. So to work around this, autogenerate now renders these
+ already-tokenized names using the new :meth:`.Operations.f` component.
+ This component is only generated if **SQLAlchemy 0.9.4** or greater
+ is in use.
+
+ Therefore it is highly recommended that an upgrade to Alembic 0.6.4
+ be accompanied by an upgrade of SQLAlchemy 0.9.4, if the new naming
+ conventions feature is used.
+
+ .. seealso::
+
+ :ref:`autogen_naming_conventions`
+
.. change::
:tags: bug
:tickets: 160
naming convention dictionary will be used to provide names for all constraints
and indexes.
+.. _autogen_naming_conventions:
+
+Integration of Naming Conventions into Operations, Autogenerate
+---------------------------------------------------------------
+
+As of Alembic 0.6.4, the naming convention feature is integrated into the
+:class:`.Operations` object, so that the convention takes effect for any
+constraint that is otherwise unnamed. The naming convention is passed to
+:class:`.Operations` using the :paramref:`.MigrationsContext.configure.target_metadata`
+parameter in ``env.py``, which is normally configured when autogenerate is
+used::
+
+ # in your application's model:
+
+ meta = MetaData(naming_convention={
+ "ix": 'ix_%(column_0_label)s',
+ "uq": "uq_%(table_name)s_%(column_0_name)s",
+ "ck": "ck_%(table_name)s_%(constraint_name)s",
+ "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
+ "pk": "pk_%(table_name)s"
+ })
+
+ # .. in your Alembic env.py:
+
+ # add your model's MetaData object here
+ # for 'autogenerate' support
+ from myapp import mymodel
+ target_metadata = mymodel.Base.metadata
+
+ # ...
+
+ def run_migrations_online():
+
+ # ...
+
+ context.configure(
+ connection=connection,
+ target_metadata=target_metadata
+ )
+
+Above, when we render a directive like the following::
+
+ op.add_column('sometable', Column('q', Boolean(name='q_bool')))
+
+The Boolean type will render a CHECK constraint with the name
+``"ck_sometable_q_bool"``, assuming the backend in use does not support
+native boolean types.
+
+We can also use op directives with constraints and not give them a name
+at all, if the naming convention doesn't require one. The value of
+``None`` will be converted into a name that follows the appopriate naming
+conventions::
+
+ def upgrade():
+ op.create_unique_constraint(None, 'some_table', 'x')
+
+When autogenerate renders constraints in a migration script, it renders them
+typically with their completed name. If using at least Alembic 0.6.4 as well
+as SQLAlchemy 0.9.4, these will be rendered with a special directive
+:meth:`.Operations.f` which denotes that the string has already been
+tokenized::
+
+ def upgrade():
+ op.create_unique_constraint(op.f('uq_const_x'), 'some_table', 'x')
+
+
For more detail on the naming convention feature, see :ref:`sqla:constraint_naming_conventions`.
from nose import SkipTest
from sqlalchemy.engine import default
-from sqlalchemy import create_engine, text
+from sqlalchemy import create_engine, text, MetaData
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.util import decorator
raise SkipTest("SQLAlchemy 0.9.2 or greater required")
return fn(*arg, **kw)
+@decorator
+def requires_094(fn, *arg, **kw):
+ if not util.sqla_094:
+ raise SkipTest("SQLAlchemy 0.9.4 or greater required")
+ return fn(*arg, **kw)
+
_dialects = {}
def _get_dialect(name):
if name is None or name == 'default':
assert re.search(msg, str(e)), "%r !~ %s" % (msg, e)
print(text_type(e))
-def op_fixture(dialect='default', as_sql=False):
+def op_fixture(dialect='default', as_sql=False, naming_convention=None):
impl = _impls[dialect]
class Impl(impl):
def __init__(self, dialect, as_sql):
sql
)
+ opts = {}
+ if naming_convention:
+ if not util.sqla_092:
+ raise SkipTest(
+ "naming_convention feature requires "
+ "sqla 0.9.2 or greater")
+ opts['target_metadata'] = MetaData(naming_convention=naming_convention)
class ctx(MigrationContext):
def __init__(self, dialect='default', as_sql=False):
self.dialect = _get_dialect(dialect)
self.impl = Impl(self.dialect, as_sql)
-
+ self.opts = opts
self.as_sql = as_sql
def assert_(self, *sql):
from sqlalchemy.sql import and_, column, literal_column
from alembic import autogenerate, util, compat
-from . import eq_, eq_ignore_whitespace, requires_092, requires_09
+from . import eq_, eq_ignore_whitespace, requires_092, requires_09, requires_094
py3k = sys.version_info >= (3, )
"sa.CheckConstraint('im a constraint', name='cc1')"
)
+
def test_render_check_constraint_sqlexpr(self):
c = column('c')
five = literal_column('5')
class RenderNamingConventionTest(TestCase):
@classmethod
- @requires_092
+ @requires_094
def setup_class(cls):
cls.autogen_context = {
'opts': {
naming_convention=convention
)
+ def test_schema_type_boolean(self):
+ t = Table('t', self.metadata, Column('c', Boolean(name='xyz')))
+ eq_ignore_whitespace(
+ autogenerate.render._add_column(
+ None, "t", t.c.c,
+ self.autogen_context),
+ "op.add_column('t', "
+ "sa.Column('c', sa.Boolean(name='xyz'), nullable=True))"
+ )
+
def test_explicit_unique_constraint(self):
t = Table('t', self.metadata, Column('c', Integer))
eq_ignore_whitespace(
UniqueConstraint(t.c.c, deferrable='XYZ'),
self.autogen_context
),
- "sa.UniqueConstraint('c', deferrable='XYZ', name='uq_ct_t_c')"
+ "sa.UniqueConstraint('c', deferrable='XYZ', name=op.f('uq_ct_t_c'))"
)
def test_explicit_named_unique_constraint(self):
autogenerate.render._render_unique_constraint(uq,
self.autogen_context
),
- "sa.UniqueConstraint('c', name='uq_ct_t_c')"
+ "sa.UniqueConstraint('c', name=op.f('uq_ct_t_c'))"
)
def test_inline_pk_constraint(self):
eq_ignore_whitespace(
autogenerate.render._add_table(t, self.autogen_context),
"op.create_table('t',sa.Column('c', sa.Integer(), nullable=False),"
- "sa.PrimaryKeyConstraint('c', name='pk_ct_t'))"
+ "sa.PrimaryKeyConstraint('c', name=op.f('pk_ct_t')))"
)
def test_inline_ck_constraint(self):
eq_ignore_whitespace(
autogenerate.render._add_table(t, self.autogen_context),
"op.create_table('t',sa.Column('c', sa.Integer(), nullable=True),"
- "sa.CheckConstraint('c > 5', name='ck_ct_t'))"
+ "sa.CheckConstraint('c > 5', name=op.f('ck_ct_t')))"
)
def test_inline_fk(self):
eq_ignore_whitespace(
autogenerate.render._add_table(t, self.autogen_context),
"op.create_table('t',sa.Column('c', sa.Integer(), nullable=True),"
- "sa.ForeignKeyConstraint(['c'], ['q.id'], name='fk_ct_t_c_q'))"
+ "sa.ForeignKeyConstraint(['c'], ['q.id'], name=op.f('fk_ct_t_c_q')))"
+ )
+
+ def test_render_check_constraint_renamed(self):
+ """test that constraints from autogenerate render with
+ the naming convention name explicitly. These names should
+ be frozen into the migration scripts so that they remain
+ the same if the application's naming convention changes.
+
+ However, op.create_table() and others need to be careful that
+ these don't double up when the "%(constraint_name)s" token is
+ used.
+
+ """
+ m1 = MetaData(naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"})
+ ck = CheckConstraint("im a constraint", name="cc1")
+ Table('t', m1, Column('x'), ck)
+
+ eq_ignore_whitespace(
+ autogenerate.render._render_check_constraint(
+ ck,
+ self.autogen_context
+ ),
+ "sa.CheckConstraint('im a constraint', name=op.f('ck_t_cc1'))"
)
"""Test against the builders in the op.* module."""
from sqlalchemy import Integer, Column, ForeignKey, \
- Table, String, Boolean
+ Table, String, Boolean, MetaData, CheckConstraint
from sqlalchemy.sql import column, func, text
from sqlalchemy import event
from alembic import op
-from . import op_fixture, assert_raises_message
+from . import op_fixture, assert_raises_message, requires_094
@event.listens_for(Table, "after_parent_attach")
def _add_cols(table, metadata):
'ALTER TABLE t1 ADD CHECK (c1 IN (0, 1))'
)
+
def test_add_column_schema_schema_type():
"""Test that a schema type generates its constraints...."""
context = op_fixture()
"ALTER TABLE bar.t1 ADD CONSTRAINT pk_test PRIMARY KEY (foo)"
)
+
def test_add_check_constraint():
context = op_fixture()
op.create_check_constraint(
--- /dev/null
+from sqlalchemy import Integer, Column, ForeignKey, \
+ Table, String, Boolean, MetaData, CheckConstraint
+from sqlalchemy.sql import column, func, text
+from sqlalchemy import event
+
+from alembic import op
+from . import op_fixture, assert_raises_message, requires_094
+
+@requires_094
+def test_add_check_constraint():
+ context = op_fixture(naming_convention={
+ "ck": "ck_%(table_name)s_%(constraint_name)s"
+ })
+ op.create_check_constraint(
+ "foo",
+ "user_table",
+ func.len(column('name')) > 5
+ )
+ context.assert_(
+ "ALTER TABLE user_table ADD CONSTRAINT ck_user_table_foo "
+ "CHECK (len(name) > 5)"
+ )
+
+@requires_094
+def test_add_check_constraint_name_is_none():
+ context = op_fixture(naming_convention={
+ "ck": "ck_%(table_name)s_foo"
+ })
+ op.create_check_constraint(
+ None,
+ "user_table",
+ func.len(column('name')) > 5
+ )
+ context.assert_(
+ "ALTER TABLE user_table ADD CONSTRAINT ck_user_table_foo "
+ "CHECK (len(name) > 5)"
+ )
+
+@requires_094
+def test_add_unique_constraint_name_is_none():
+ context = op_fixture(naming_convention={
+ "uq": "uq_%(table_name)s_foo"
+ })
+ op.create_unique_constraint(
+ None,
+ "user_table",
+ 'x'
+ )
+ context.assert_(
+ "ALTER TABLE user_table ADD CONSTRAINT uq_user_table_foo UNIQUE (x)"
+ )
+
+
+@requires_094
+def test_add_index_name_is_none():
+ context = op_fixture(naming_convention={
+ "ix": "ix_%(table_name)s_foo"
+ })
+ op.create_index(
+ None,
+ "user_table",
+ 'x'
+ )
+ context.assert_(
+ "CREATE INDEX ix_user_table_foo ON user_table (x)"
+ )
+
+
+
+@requires_094
+def test_add_check_constraint_already_named_from_schema():
+ m1 = MetaData(naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"})
+ ck = CheckConstraint("im a constraint", name="cc1")
+ Table('t', m1, Column('x'), ck)
+
+ context = op_fixture(
+ naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"})
+
+ op.create_table(
+ "some_table",
+ Column('x', Integer, ck),
+ )
+ context.assert_(
+ "CREATE TABLE some_table "
+ "(x INTEGER CONSTRAINT ck_t_cc1 CHECK (im a constraint))"
+ )
+
+@requires_094
+def test_add_check_constraint_inline_on_table():
+ context = op_fixture(
+ naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"})
+ op.create_table(
+ "some_table",
+ Column('x', Integer),
+ CheckConstraint("im a constraint", name="cc1")
+ )
+ context.assert_(
+ "CREATE TABLE some_table "
+ "(x INTEGER, CONSTRAINT ck_some_table_cc1 CHECK (im a constraint))"
+ )
+
+@requires_094
+def test_add_check_constraint_inline_on_table_w_f():
+ context = op_fixture(
+ naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"})
+ op.create_table(
+ "some_table",
+ Column('x', Integer),
+ CheckConstraint("im a constraint", name=op.f("ck_some_table_cc1"))
+ )
+ context.assert_(
+ "CREATE TABLE some_table "
+ "(x INTEGER, CONSTRAINT ck_some_table_cc1 CHECK (im a constraint))"
+ )
+
+@requires_094
+def test_add_check_constraint_inline_on_column():
+ context = op_fixture(
+ naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"})
+ op.create_table(
+ "some_table",
+ Column('x', Integer, CheckConstraint("im a constraint", name="cc1"))
+ )
+ context.assert_(
+ "CREATE TABLE some_table "
+ "(x INTEGER CONSTRAINT ck_some_table_cc1 CHECK (im a constraint))"
+ )
+
+@requires_094
+def test_add_check_constraint_inline_on_column_w_f():
+ context = op_fixture(
+ naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"})
+ op.create_table(
+ "some_table",
+ Column('x', Integer, CheckConstraint("im a constraint", name=op.f("ck_q_cc1")))
+ )
+ context.assert_(
+ "CREATE TABLE some_table "
+ "(x INTEGER CONSTRAINT ck_q_cc1 CHECK (im a constraint))"
+ )
+
+
+@requires_094
+def test_add_column_schema_type():
+ context = op_fixture(naming_convention={
+ "ck": "ck_%(table_name)s_%(constraint_name)s"
+ })
+ op.add_column('t1', Column('c1', Boolean(name='foo'), nullable=False))
+ context.assert_(
+ 'ALTER TABLE t1 ADD COLUMN c1 BOOLEAN NOT NULL',
+ 'ALTER TABLE t1 ADD CONSTRAINT ck_t1_foo CHECK (c1 IN (0, 1))'
+ )
+
+
+@requires_094
+def test_add_column_schema_type_w_f():
+ context = op_fixture(naming_convention={
+ "ck": "ck_%(table_name)s_%(constraint_name)s"
+ })
+ op.add_column('t1', Column('c1', Boolean(name=op.f('foo')), nullable=False))
+ context.assert_(
+ 'ALTER TABLE t1 ADD COLUMN c1 BOOLEAN NOT NULL',
+ 'ALTER TABLE t1 ADD CONSTRAINT foo CHECK (c1 IN (0, 1))'
+ )
+
+
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
