def _render_potential_expr(value, autogen_context, wrap_in_text=True):
if isinstance(value, sql.ClauseElement):
- if util.sqla_08:
- compile_kw = dict(compile_kwargs={
- 'literal_binds': True, "include_table": False})
- else:
- compile_kw = {}
+ compile_kw = dict(compile_kwargs={
+ 'literal_binds': True, "include_table": False})
if wrap_in_text:
template = "%(prefix)stext(%(sql)r)"
def _get_index_rendered_expressions(idx, autogen_context):
- if util.sqla_08:
- return [repr(_ident(getattr(exp, "name", None)))
- if isinstance(exp, sa_schema.Column)
- else _render_potential_expr(exp, autogen_context)
- for exp in idx.expressions]
- else:
- return [
- repr(_ident(getattr(col, "name", None))) for col in idx.columns]
+ return [repr(_ident(getattr(exp, "name", None)))
+ if isinstance(exp, sa_schema.Column)
+ else _render_potential_expr(exp, autogen_context)
+ for exp in idx.expressions]
def _uq_constraint(constraint, autogen_context, alter):
self.connection = connection
self.as_sql = as_sql
self.literal_binds = context_opts.get('literal_binds', False)
- if self.literal_binds and not util.sqla_08:
- util.warn("'literal_binds' flag not supported in SQLAlchemy 0.7")
- self.literal_binds = False
self.output_buffer = output_buffer
self.memo = {}
new_table_name, schema=schema))
def create_table(self, table):
- if util.sqla_07:
- table.dispatch.before_create(table, self.connection,
- checkfirst=False,
- _ddl_runner=self)
+ table.dispatch.before_create(table, self.connection,
+ checkfirst=False,
+ _ddl_runner=self)
self._exec(schema.CreateTable(table))
- if util.sqla_07:
- table.dispatch.after_create(table, self.connection,
- checkfirst=False,
- _ddl_runner=self)
+ table.dispatch.after_create(table, self.connection,
+ checkfirst=False,
+ _ddl_runner=self)
for index in table.indexes:
self._exec(schema.CreateIndex(index))
pass
def _compat_autogen_column_reflect(self, inspector):
- if util.sqla_08:
- return self.autogen_column_reflect
- else:
- def adapt(table, column_info):
- return self.autogen_column_reflect(
- inspector, table, column_info)
- return adapt
+ return self.autogen_column_reflect
def correct_for_autogen_foreignkeys(self, conn_fks, metadata_fks):
pass
import logging
-if util.sqla_08:
- from sqlalchemy.sql.expression import UnaryExpression
-else:
- from sqlalchemy.sql.expression import _UnaryExpression as UnaryExpression
+from sqlalchemy.sql.expression import UnaryExpression
if util.sqla_100:
from sqlalchemy.dialects.postgresql import ExcludeConstraint
for idx in list(metadata_indexes):
if idx.name in conn_indexes_by_name:
continue
- if util.sqla_08:
- exprs = idx.expressions
- else:
- exprs = idx.columns
+ exprs = idx.expressions
for expr in exprs:
while isinstance(expr, UnaryExpression):
expr = expr.element
from sqlalchemy import schema as sql_schema
from sqlalchemy.util import OrderedDict
from .. import util
-if util.sqla_08:
- from sqlalchemy.events import SchemaEventTarget
+from sqlalchemy.events import SchemaEventTarget
from ..util.sqla_compat import _columns_for_constraint, \
_is_type_bound, _fk_is_self_referential, _remove_column_from_collection
def __init__(self, operations, table_name, schema, recreate,
copy_from, table_args, table_kwargs,
reflect_args, reflect_kwargs, naming_convention):
- if not util.sqla_08:
- raise NotImplementedError(
- "batch mode requires SQLAlchemy 0.8 or greater.")
self.operations = operations
self.table_name = table_name
self.schema = schema
tname = "%s.%s" % (referent_schema, referent) if referent_schema \
else referent
- if util.sqla_08:
- # "match" kw unsupported in 0.7
- dialect_kw['match'] = match
+ dialect_kw['match'] = match
f = sa_schema.ForeignKeyConstraint(local_cols,
["%s.%s" % (tname, n)
from sqlalchemy import schema
from alembic import util
- if util.sqla_08:
- from sqlalchemy import inspect
- else:
- from sqlalchemy.engine.reflection import Inspector
- inspect = Inspector.from_engine
+ from sqlalchemy import inspect
if options.dropfirst:
for cfg in config.Config.all_configs():
@property
def unique_constraint_reflection(self):
def doesnt_have_check_uq_constraints(config):
- if not util.sqla_084:
- return True
from sqlalchemy import inspect
# temporary
pass
return False
- return exclusions.skip_if(
- lambda config: not util.sqla_084,
- "SQLAlchemy 0.8.4 or greater required"
- ) + exclusions.skip_if(doesnt_have_check_uq_constraints)
+ return exclusions.skip_if(doesnt_have_check_uq_constraints)
@property
def foreign_key_match(self):
- return exclusions.fails_if(
- lambda config: not util.sqla_08,
- "MATCH for foreign keys added in SQLAlchemy 0.8.0"
- )
+ return exclusions.open()
@property
def check_constraints_w_enforcement(self):
def reflects_fk_options(self):
return exclusions.closed()
- @property
- def fail_before_sqla_079(self):
- return exclusions.fails_if(
- lambda config: not util.sqla_079,
- "SQLAlchemy 0.7.9 or greater required"
- )
-
- @property
- def fail_before_sqla_080(self):
- return exclusions.fails_if(
- lambda config: not util.sqla_08,
- "SQLAlchemy 0.8.0 or greater required"
- )
-
- @property
- def fail_before_sqla_083(self):
- return exclusions.fails_if(
- lambda config: not util.sqla_083,
- "SQLAlchemy 0.8.3 or greater required"
- )
-
- @property
- def fail_before_sqla_084(self):
- return exclusions.fails_if(
- lambda config: not util.sqla_084,
- "SQLAlchemy 0.8.4 or greater required"
- )
-
- @property
- def fail_before_sqla_09(self):
- return exclusions.fails_if(
- lambda config: not util.sqla_09,
- "SQLAlchemy 0.9.0 or greater required"
- )
-
@property
def fail_before_sqla_100(self):
return exclusions.fails_if(
"SQLAlchemy 1.1.0 or greater required"
)
- @property
- def sqlalchemy_08(self):
-
- return exclusions.skip_if(
- lambda config: not util.sqla_08,
- "SQLAlchemy 0.8.0b2 or greater required"
- )
-
- @property
- def sqlalchemy_09(self):
- return exclusions.skip_if(
- lambda config: not util.sqla_09,
- "SQLAlchemy 0.9.0 or greater required"
- )
-
@property
def sqlalchemy_092(self):
return exclusions.skip_if(
"SQLAlchemy 0.9.4 or greater required"
)
+ @property
+ def sqlalchemy_099(self):
+ return exclusions.skip_if(
+ lambda config: not util.sqla_099,
+ "SQLAlchemy 0.9.9 or greater required"
+ )
+
@property
def sqlalchemy_100(self):
return exclusions.skip_if(
template_to_file, coerce_resource_to_filename,
pyc_file_from_path, load_python_file, edit)
from .sqla_compat import ( # noqa
- sqla_07, sqla_079, sqla_08, sqla_083, sqla_084, sqla_09, sqla_092,
- sqla_094, sqla_099, sqla_100, sqla_105, sqla_110, sqla_1010, sqla_1014,
- sqla_1115)
+ sqla_09, sqla_092, sqla_094, sqla_099, sqla_100, sqla_105, sqla_110, sqla_1010,
+ sqla_1014, sqla_1115)
from .exc import CommandError
-if not sqla_07:
+if not sqla_09:
raise CommandError(
- "SQLAlchemy 0.7.3 or greater is required. ")
+ "SQLAlchemy 0.9.0 or greater is required. ")
from sqlalchemy.sql.visitors import traverse
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import _BindParamClause
+from sqlalchemy.sql.expression import _TextClause as TextClause
from . import compat
return value
_vers = tuple(
[_safe_int(x) for x in re.findall(r'(\d+|[abc]\d)', __version__)])
-sqla_07 = _vers > (0, 7, 2)
-sqla_079 = _vers >= (0, 7, 9)
-sqla_08 = _vers >= (0, 8, 0)
-sqla_083 = _vers >= (0, 8, 3)
-sqla_084 = _vers >= (0, 8, 4)
sqla_09 = _vers >= (0, 9, 0)
sqla_092 = _vers >= (0, 9, 2)
sqla_094 = _vers >= (0, 9, 4)
sqla_1014 = _vers >= (1, 0, 14)
sqla_1115 = _vers >= (1, 1, 15)
-if sqla_08:
- from sqlalchemy.sql.expression import TextClause
-else:
- from sqlalchemy.sql.expression import _TextClause as TextClause
if sqla_110:
AUTOINCREMENT_DEFAULT = 'auto'
else:
AUTOINCREMENT_DEFAULT = True
+
def _table_for_constraint(constraint):
if isinstance(constraint, ForeignKeyConstraint):
return constraint.parent
def _get_index_expressions(idx):
- if sqla_08:
- return list(idx.expressions)
- else:
- return list(idx.columns)
+ return list(idx.expressions)
def _get_index_column_names(idx):
def _get_index_final_name(dialect, idx):
- if sqla_08:
- # trying to keep the truncation rules totally localized on the
- # SQLA side while also stepping around the quoting issue. Ideally
- # the _prepared_index_name() method on the SQLA side would have
- # a quoting option or the truncation routine would be broken out.
- #
- # test for SQLA quoted_name construct, introduced in
- # 0.9 or thereabouts.
- # this doesn't work in 0.8 and the "quote" option on Index doesn't
- # seem to work in 0.8 either.
- if hasattr(idx.name, "quote"):
- # might be quoted_name, might be truncated_name, keep it the
- # same
- quoted_name_cls = type(idx.name)
- new_name = quoted_name_cls(str(idx.name), quote=False)
- idx = schema.Index(name=new_name)
- return dialect.ddl_compiler(dialect, None)._prepared_index_name(idx)
- else:
- return idx.name
+ # trying to keep the truncation rules totally localized on the
+ # SQLA side while also stepping around the quoting issue. Ideally
+ # the _prepared_index_name() method on the SQLA side would have
+ # a quoting option or the truncation routine would be broken out.
+ #
+ # test for SQLA quoted_name construct, introduced in
+ # 0.9 or thereabouts.
+ # this doesn't work in 0.8 and the "quote" option on Index doesn't
+ # seem to work in 0.8 either.
+ if hasattr(idx.name, "quote"):
+ # might be quoted_name, might be truncated_name, keep it the
+ # same
+ quoted_name_cls = type(idx.name)
+ new_name = quoted_name_cls(str(idx.name), quote=False)
+ idx = schema.Index(name=new_name)
+ return dialect.ddl_compiler(dialect, None)._prepared_index_name(idx)
def _is_mariadb(mysql_dialect):
Alembic's install process will ensure that SQLAlchemy_
is installed, in addition to other dependencies. Alembic will work with
-SQLAlchemy as of version **0.7.3**, however more features are available with
-newer versions such as the 0.9 or 1.0 series.
+SQLAlchemy as of version **0.9.0**, however more features are available with
+newer versions such as the 1.1 or 1.2 series.
+
+.. versionchanged:: 1.0.0 Support for SQLAlchemy 0.8 and 0.7.9 was dropped.
Alembic supports Python versions 2.7, 3.4 and above.
--- /dev/null
+.. change::
+ :tags: feature, general
+
+ With the 1.0 release, Alembic's minimum SQLAlchemy support version
+ moves to 0.9.0, previously 0.7.9.
readme = os.path.join(os.path.dirname(__file__), 'README.rst')
requires = [
- 'SQLAlchemy>=0.7.6',
+ 'SQLAlchemy>=0.9.0',
'Mako',
'python-editor>=0.3',
'python-dateutil'
class AutogenerateFKOptionsTest(AutogenFixtureTest, TestBase):
__backend__ = True
- __requires__ = ('sqlalchemy_09', 'flexible_fk_cascades')
+ __requires__ = ('flexible_fk_cascades', )
def _fk_opts_fixture(self, old_opts, new_opts):
m1 = MetaData()
raise NotImplementedError()
eng.dialect.get_unique_constraints = unimpl
- @config.requirements.fail_before_sqla_083
def test_add_ix_on_table_create(self):
return super(NoUqReflection, self).test_add_ix_on_table_create()
- @config.requirements.fail_before_sqla_080
def test_add_idx_non_col(self):
return super(NoUqReflection, self).test_add_idx_non_col()
# truncation rules either. dropping these ancient versions
# is long overdue.
- @config.requirements.sqlalchemy_09
def test_unchanged_case_sensitive_implicit_idx(self):
m1 = MetaData()
m2 = MetaData()
eq_(diffs, [])
- @config.requirements.sqlalchemy_09
def test_unchanged_case_sensitive_explicit_idx(self):
m1 = MetaData()
m2 = MetaData()
class TruncatedIdxTest(AutogenFixtureTest, TestBase):
- __requires__ = ('sqlalchemy_09', )
def setUp(self):
self.bind = engines.testing_engine()
"['active', 'code'], unique=False)"
)
- @config.requirements.fail_before_sqla_080
def test_render_add_index_func(self):
m = MetaData()
t = Table(
"[sa.text(!U'lower(code)')], unique=False)"
)
- @config.requirements.fail_before_sqla_080
def test_render_add_index_cast(self):
m = MetaData()
t = Table(
"[sa.text(!U'CAST(code AS VARCHAR)')], unique=False)"
)
- @config.requirements.fail_before_sqla_080
def test_render_add_index_desc(self):
m = MetaData()
t = Table(
t2 = Table('t2', m, Column('c_rem', Integer))
fk = ForeignKeyConstraint([t1.c.c], [t2.c.c_rem], onupdate="CASCADE")
- if not util.sqla_08:
- t1.append_constraint(fk)
# SQLA 0.9 generates a u'' here for remote cols while 0.8 does not,
# so just whack out "'u" here from the generated
)
fk = ForeignKeyConstraint([t1.c.c], [t2.c.c_rem], ondelete="CASCADE")
- if not util.sqla_08:
- t1.append_constraint(fk)
op_obj = ops.AddConstraintOp.from_constraint(fk)
eq_ignore_whitespace(
)
fk = ForeignKeyConstraint([t1.c.c], [t2.c.c_rem], deferrable=True)
- if not util.sqla_08:
- t1.append_constraint(fk)
op_obj = ops.AddConstraintOp.from_constraint(fk)
eq_ignore_whitespace(
re.sub(
)
fk = ForeignKeyConstraint([t1.c.c], [t2.c.c_rem], initially="XYZ")
- if not util.sqla_08:
- t1.append_constraint(fk)
op_obj = ops.AddConstraintOp.from_constraint(fk)
eq_ignore_whitespace(
re.sub(
fk = ForeignKeyConstraint(
[t1.c.c], [t2.c.c_rem],
initially="XYZ", ondelete="CASCADE", deferrable=True)
- if not util.sqla_08:
- t1.append_constraint(fk)
op_obj = ops.AddConstraintOp.from_constraint(fk)
eq_ignore_whitespace(
re.sub(
"schema=%r)" % compat.ue('\u0411\u0435\u0437')
)
- @config.requirements.sqlalchemy_09
def test_render_table_w_unsupported_constraint(self):
from sqlalchemy.sql.schema import ColumnCollectionConstraint
t2 = Table('t2', m, Column('c_rem', Integer))
fk = ForeignKeyConstraint([t1.c.c], [t2.c.c_rem], onupdate="CASCADE")
- if not util.sqla_08:
- t1.append_constraint(fk)
# SQLA 0.9 generates a u'' here for remote cols while 0.8 does not,
# so just whack out "'u" here from the generated
)
fk = ForeignKeyConstraint([t1.c.c], [t2.c.c_rem], ondelete="CASCADE")
- if not util.sqla_08:
- t1.append_constraint(fk)
eq_ignore_whitespace(
re.sub(
)
fk = ForeignKeyConstraint([t1.c.c], [t2.c.c_rem], deferrable=True)
- if not util.sqla_08:
- t1.append_constraint(fk)
eq_ignore_whitespace(
re.sub(
r"u'", "'",
)
fk = ForeignKeyConstraint([t1.c.c], [t2.c.c_rem], initially="XYZ")
- if not util.sqla_08:
- t1.append_constraint(fk)
eq_ignore_whitespace(
re.sub(
r"u'", "'",
fk = ForeignKeyConstraint(
[t1.c.c], [t2.c.c_rem],
initially="XYZ", ondelete="CASCADE", deferrable=True)
- if not util.sqla_08:
- t1.append_constraint(fk)
eq_ignore_whitespace(
re.sub(
r"u'", "'",
t2 = Table('t2', m, Column('c_rem', Integer))
fk = ForeignKeyConstraint([t1.c.c], [t2.c.c_rem], onupdate="CASCADE")
- if not util.sqla_08:
- t1.append_constraint(fk)
eq_ignore_whitespace(
re.sub(
"sa.CheckConstraint(!U'c > 5 AND c < 10')"
)
- @config.requirements.fail_before_sqla_080
def test_render_check_constraint_literal_binds(self):
c = column('c')
eq_ignore_whitespace(
"existing_server_default='5')"
)
- @config.requirements.fail_before_sqla_079
def test_render_enum(self):
eq_ignore_whitespace(
autogenerate.render._repr_type(
"sa.Enum('one', 'two', 'three')"
)
- @exclusions.fails_if(
- lambda config: (util.sqla_09 and not util.sqla_099) or not util.sqla_079,
- "Fails on SQLAlchemy <0.7.9, 0.9.0-0.9.8"
- )
+ @config.requirements.sqlalchemy_099
def test_render_non_native_enum(self):
eq_ignore_whitespace(
autogenerate.render._repr_type(
"user.MyType()"
)
- @config.requirements.sqlalchemy_09
def test_repr_dialect_type(self):
from sqlalchemy.dialects.mysql import VARCHAR
'nullable=False)'
)
- @config.requirements.fail_before_sqla_09
def test_render_server_default_non_native_boolean(self):
c = Column(
'updated_at', Boolean(),
class BatchApplyTest(TestBase):
- __requires__ = ('sqlalchemy_08', )
def setUp(self):
self.op = Operations(mock.Mock(opts={}))
ddl_contains='FOREIGN KEY(user_id_3, user_id_version) '
'REFERENCES "user" (id, id_version)')
- # _get_colspec() in 0.8 calls upon fk.column when schema is
- # present. not sure if we want to try to fix this
- @config.requirements.fail_before_sqla_09
def test_regen_multi_fk_schema(self):
impl = self._multi_fk_fixture(schema='foo_schema')
self._assert_impl(
class BatchAPITest(TestBase):
- __requires__ = ('sqlalchemy_08', )
@contextmanager
def _fixture(self, schema=None):
class CopyFromTest(TestBase):
- __requires__ = ('sqlalchemy_08', )
def _fixture(self):
self.metadata = MetaData()
class BatchRoundTripTest(TestBase):
- __requires__ = ('sqlalchemy_09', )
__only_on__ = "sqlite"
def setUp(self):
class ExternalDialectRenderTest(TestBase):
- __requires__ = ('sqlalchemy_09', )
def setUp(self):
ctx_opts = {
nullable=False)
context.assert_('ALTER TABLE tests ALTER COLUMN col BIT NOT NULL')
- @config.requirements.fail_before_sqla_084
def test_drop_index(self):
context = op_fixture('mssql')
op.drop_index('my_idx', 'my_table')
"exec('alter table t1 drop constraint ' + @const_name)")
context.assert_contains("ALTER TABLE t1 DROP COLUMN c1")
- @config.requirements.sqlalchemy_08
def test_drop_column_w_default_in_batch(self):
context = op_fixture('mssql')
with op.batch_alter_table('t1', schema=None) as batch_op:
"exec('alter table t1 drop constraint ' + @const_name)")
context.assert_contains("ALTER TABLE t1 DROP COLUMN c1")
- @config.requirements.sqlalchemy_08
def test_drop_column_w_check_in_batch(self):
context = op_fixture('mssql')
with op.batch_alter_table('t1', schema=None) as batch_op:
"exec('alter table t1 drop constraint ' + @const_name)")
context.assert_contains("ALTER TABLE t1 DROP COLUMN c1")
- @config.requirements.sqlalchemy_08
def test_drop_column_w_fk_in_batch(self):
context = op_fixture('mssql')
with op.batch_alter_table('t1', schema=None) as batch_op:
op.create_index, 'name', 'tname', [func.foo(column('x'))]
)
- @config.requirements.sqlalchemy_09
def test_add_column_schema_hard_quoting(self):
from sqlalchemy.sql.schema import quoted_name
context = op_fixture("postgresql")
'ALTER TABLE "some.schema".somename ADD COLUMN colname VARCHAR'
)
- @config.requirements.sqlalchemy_09
def test_rename_table_schema_hard_quoting(self):
from sqlalchemy.sql.schema import quoted_name
context = op_fixture("postgresql")
'ALTER TABLE "some.schema".t1 RENAME TO t2'
)
- @config.requirements.sqlalchemy_09
def test_add_constraint_schema_hard_quoting(self):
from sqlalchemy.sql.schema import quoted_name
context = op_fixture("postgresql")
context.assert_(
'CREATE INDEX geocoded ON locations ("IShouldBeQuoted")')
- @config.requirements.fail_before_sqla_080
def test_create_index_expressions(self):
context = op_fixture()
op.create_index(
"CREATE UNIQUE INDEX ik_test ON t1 (foo, bar)"
)
- @config.requirements.fail_before_sqla_09
def test_create_index_quote_flag(self):
context = op_fixture()
op.create_index('ik_test', 't1', ['foo', 'bar'], quote=True)
op.drop_constraint("f1", "t1", type_="foreignkey")
context.assert_("ALTER TABLE t1 DROP FOREIGN KEY f1")
- @config.requirements.fail_before_sqla_084
def test_naming_changes_drop_idx(self):
context = op_fixture('mssql')
op.drop_index('ik_test', tablename='t1')
class SQLModeOpTest(TestBase):
- @config.requirements.sqlalchemy_09
def test_auto_literals(self):
context = op_fixture(as_sql=True, literal_binds=True)
from sqlalchemy.sql import table, column
op.rename_table('t1', 't2', schema="foo")
context.assert_("ALTER TABLE foo.t1 RENAME TO t2")
- @config.requirements.fail_before_sqla_080
def test_create_index_postgresql_expressions(self):
context = op_fixture("postgresql")
op.create_index(
""" % self.rid)
- @config.requirements.sqlalchemy_09
def test_offline_inline_enum_create(self):
self._inline_enum_script()
with capture_context_buffer() as buf:
# no drop since we didn't emit events
assert "DROP TYPE pgenum" not in buf.getvalue()
- @config.requirements.sqlalchemy_09
def test_offline_distinct_enum_create(self):
self._distinct_enum_script()
with capture_context_buffer() as buf:
op_obj = ops.CreateIndexOp.from_index(idx)
- if util.sqla_08:
- eq_ignore_whitespace(
- autogenerate.render_op_text(autogen_context, op_obj),
- """op.create_index('foo_idx', 't', \
+ eq_ignore_whitespace(
+ autogenerate.render_op_text(autogen_context, op_obj),
+ """op.create_index('foo_idx', 't', \
['x', 'y'], unique=False, """
- """postgresql_where=sa.text(!U"y = 'something'"))"""
- )
- else:
- eq_ignore_whitespace(
- autogenerate.render_op_text(autogen_context, op_obj),
- """op.create_index('foo_idx', 't', ['x', 'y'], \
-unique=False, """
- """postgresql_where=sa.text(!U't.y = %(y_1)s'))"""
- )
+ """postgresql_where=sa.text(!U"y = 'something'"))"""
+ )
def test_render_server_default_native_boolean(self):
c = Column(
'nullable=False)'
)
- @config.requirements.sqlalchemy_09
def test_postgresql_array_type(self):
eq_ignore_whitespace(
assert 'from sqlalchemy.dialects import postgresql' in \
self.autogen_context.imports
- @config.requirements.sqlalchemy_09
def test_array_type_user_defined_inner(self):
def repr_type(typestring, object_, autogen_context):
if typestring == 'type' and isinstance(object_, String):
"name='TExclX'))"
)
- @config.requirements.sqlalchemy_09
def test_json_type(self):
if config.requirements.sqlalchemy_110.enabled:
eq_ignore_whitespace(
"postgresql.JSON()"
)
- @config.requirements.sqlalchemy_09
def test_jsonb_type(self):
if config.requirements.sqlalchemy_110.enabled:
eq_ignore_whitespace(
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
