from .runtime import environment
from .runtime import migration
-__version__ = '1.0.12'
+__version__ = "1.1.0"
package_dir = path.abspath(path.dirname(__file__))
if metadata_col.table._autoincrement_column is metadata_col:
alter_column_op.kw["autoincrement"] = True
- elif util.sqla_110 and metadata_col.autoincrement is True:
+ elif metadata_col.autoincrement is True:
alter_column_op.kw["autoincrement"] = True
elif metadata_col.autoincrement is False:
alter_column_op.kw["autoincrement"] = False
"""
if name is None:
return name
- elif util.sqla_09 and isinstance(name, sql.elements.quoted_name):
+ elif isinstance(name, sql.elements.quoted_name):
if compat.py2k:
# the attempt to encode to ascii here isn't super ideal,
# however we are trying to cut down on an explosion of
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.schema import Column
from sqlalchemy.schema import DDLElement
+from sqlalchemy.sql.elements import quoted_name
-from .. import util
from ..util.sqla_compat import _columns_for_constraint # noqa
from ..util.sqla_compat import _find_columns # noqa
from ..util.sqla_compat import _fk_spec # noqa
from ..util.sqla_compat import _is_type_bound # noqa
from ..util.sqla_compat import _table_for_constraint # noqa
-# backwards compat
-
-if util.sqla_09:
- from sqlalchemy.sql.elements import quoted_name
-
class AlterTable(DDLElement):
def quote_dotted(name, quote):
"""quote the elements of a dotted name"""
- if util.sqla_09 and isinstance(name, quoted_name):
+ if isinstance(name, quoted_name):
return quote(name)
result = ".".join([quote(x) for x in name.split(".")])
return result
from ..util.compat import string_types
from ..util.sqla_compat import _is_mariadb
from ..util.sqla_compat import _is_type_bound
-from ..util.sqla_compat import sqla_100
class MySQLImpl(DefaultImpl):
if idx.name in removed:
metadata_indexes.remove(idx)
- if not sqla_100:
- self._legacy_correct_for_dupe_uq_uix(
- conn_unique_constraints,
- conn_indexes,
- metadata_unique_constraints,
- metadata_indexes,
- )
-
def _legacy_correct_for_dupe_uq_uix(
self,
conn_unique_constraints,
from sqlalchemy import text
from sqlalchemy import types as sqltypes
from sqlalchemy.dialects.postgresql import BIGINT
+from sqlalchemy.dialects.postgresql import ExcludeConstraint
from sqlalchemy.dialects.postgresql import INTEGER
from sqlalchemy.sql.expression import ColumnClause
from sqlalchemy.sql.expression import UnaryExpression
from ..util import compat
from ..util import sqla_compat
-if util.sqla_100:
- from sqlalchemy.dialects.postgresql import ExcludeConstraint
-
log = logging.getLogger(__name__)
# check for unquoted string and quote for PG String types
if (
- not isinstance(inspector_column.type, Numeric) and
- metadata_column.server_default is not None
+ not isinstance(inspector_column.type, Numeric)
+ and metadata_column.server_default is not None
and isinstance(
metadata_column.server_default.arg, compat.string_types
)
metadata_indexes,
):
- conn_uniques_by_name = dict(
- (c.name, c) for c in conn_unique_constraints
- )
conn_indexes_by_name = dict((c.name, c) for c in conn_indexes)
- if not util.sqla_100:
- doubled_constraints = set(
- conn_indexes_by_name[name]
- for name in set(conn_uniques_by_name).intersection(
- conn_indexes_by_name
- )
- )
- else:
- doubled_constraints = set(
- index
- for index in conn_indexes
- if index.info.get("duplicates_constraint")
- )
+ doubled_constraints = set(
+ index
+ for index in conn_indexes
+ if index.info.get("duplicates_constraint")
+ )
for ix in doubled_constraints:
conn_indexes.remove(ix)
)
def to_constraint(self, migration_context=None):
- if not util.sqla_100:
- raise NotImplementedError(
- "ExcludeConstraint not supported until SQLAlchemy 1.0"
- )
if self._orig_constraint is not None:
return self._orig_constraint
schema_obj = schemaobj.SchemaObjects(migration_context)
return _exclude_constraint(op.to_constraint(), autogen_context, alter=True)
-if util.sqla_100:
-
- @render._constraint_renderers.dispatch_for(ExcludeConstraint)
- def _render_inline_exclude_constraint(constraint, autogen_context):
- rendered = render._user_defined_render(
- "exclude", constraint, autogen_context
- )
- if rendered is not False:
- return rendered
+@render._constraint_renderers.dispatch_for(ExcludeConstraint)
+def _render_inline_exclude_constraint(constraint, autogen_context):
+ rendered = render._user_defined_render(
+ "exclude", constraint, autogen_context
+ )
+ if rendered is not False:
+ return rendered
- return _exclude_constraint(constraint, autogen_context, False)
+ return _exclude_constraint(constraint, autogen_context, False)
def _postgresql_autogenerate_prefix(autogen_context):
return rendered_inspector_default != rendered_metadata_default
- def correct_for_autogen_constraints(
- self,
- conn_unique_constraints,
- conn_indexes,
- metadata_unique_constraints,
- metadata_indexes,
- ):
-
- if util.sqla_100:
- return
-
- # adjustments to accommodate for SQLite unnamed unique constraints
- # not being reported from the backend; this was updated in
- # SQLA 1.0.
-
- def uq_sig(uq):
- return tuple(sorted(uq.columns.keys()))
-
- conn_unique_sigs = set(uq_sig(uq) for uq in conn_unique_constraints)
-
- for idx in list(metadata_unique_constraints):
- # SQLite backend can't report on unnamed UNIQUE constraints,
- # so remove these, unless we see an exact signature match
- if idx.name is None and uq_sig(idx) not in conn_unique_sigs:
- metadata_unique_constraints.remove(idx)
-
def _guess_if_default_is_unparenthesized_sql_expr(self, expr):
"""Determine if a server default is a SQL expression or a constant.
from sqlalchemy import exc as sa_exc
from sqlalchemy.engine import default
+from sqlalchemy.testing.assertions import assert_raises # noqa
+from sqlalchemy.testing.assertions import assert_raises_message # noqa
+from sqlalchemy.testing.assertions import eq_ # noqa
+from sqlalchemy.testing.assertions import is_ # noqa
+from sqlalchemy.testing.assertions import is_not_ # noqa
+from sqlalchemy.testing.assertions import ne_ # noqa
from sqlalchemy.util import decorator
from . import config
from ..util.compat import text_type
-if not util.sqla_094:
-
- def eq_(a, b, msg=None):
- """Assert a == b, with repr messaging on failure."""
- assert a == b, msg or "%r != %r" % (a, b)
-
- def ne_(a, b, msg=None):
- """Assert a != b, with repr messaging on failure."""
- assert a != b, msg or "%r == %r" % (a, b)
-
- def is_(a, b, msg=None):
- """Assert a is b, with repr messaging on failure."""
- assert a is b, msg or "%r is not %r" % (a, b)
-
- def is_not_(a, b, msg=None):
- """Assert a is not b, with repr messaging on failure."""
- assert a is not b, msg or "%r is %r" % (a, b)
-
- def assert_raises(except_cls, callable_, *args, **kw):
- try:
- callable_(*args, **kw)
- success = False
- except except_cls:
- success = True
-
- # assert outside the block so it works for AssertionError too !
- assert success, "Callable did not raise an exception"
-
- def assert_raises_message(except_cls, msg, callable_, *args, **kwargs):
- try:
- callable_(*args, **kwargs)
- assert False, "Callable did not raise an exception"
- except except_cls as e:
- assert re.search(msg, text_type(e), re.UNICODE), "%r !~ %s" % (
- msg,
- e,
- )
- print(text_type(e).encode("utf-8"))
-
-
-else:
- from sqlalchemy.testing.assertions import assert_raises # noqa
- from sqlalchemy.testing.assertions import assert_raises_message # noqa
- from sqlalchemy.testing.assertions import eq_ # noqa
- from sqlalchemy.testing.assertions import is_ # noqa
- from sqlalchemy.testing.assertions import is_not_ # noqa
- from sqlalchemy.testing.assertions import ne_ # noqa
-
-
def eq_ignore_whitespace(a, b, msg=None):
a = re.sub(r"^\s+?|\n", "", a)
a = re.sub(r" {2,}", " ", a)
from sqlalchemy import String
from sqlalchemy import Table
from sqlalchemy import text
+from sqlalchemy.testing.fixtures import TestBase # noqa
import alembic
from . import config
from . import mock
from .assertions import _get_dialect
from .assertions import eq_
-from .plugin.plugin_base import SkipTest
-from .. import util
from ..environment import EnvironmentContext
from ..migration import MigrationContext
from ..operations import Operations
testing_config.read(["test.cfg"])
-if not util.sqla_094:
-
- class TestBase(object):
- # A sequence of database names to always run, regardless of the
- # constraints below.
- __whitelist__ = ()
-
- # A sequence of requirement names matching testing.requires decorators
- __requires__ = ()
-
- # A sequence of dialect names to exclude from the test class.
- __unsupported_on__ = ()
-
- # If present, test class is only runnable for the *single* specified
- # dialect. If you need multiple, use __unsupported_on__ and invert.
- __only_on__ = None
-
- # A sequence of no-arg callables. If any are True, the entire
- # testcase is skipped.
- __skip_if__ = None
-
- def assert_(self, val, msg=None):
- assert val, msg
-
- # apparently a handful of tests are doing this....OK
- def setup(self):
- if hasattr(self, "setUp"):
- self.setUp()
-
- def teardown(self):
- if hasattr(self, "tearDown"):
- self.tearDown()
-
-
-else:
- from sqlalchemy.testing.fixtures import TestBase # noqa
-
-
def capture_db():
buf = []
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 buffer_(object):
"""
-Bootstrapper for nose/pytest plugins.
+Bootstrapper for test framework plugins.
The entire rationale for this system is to get the modules in plugin/
imported without importing all of the supporting library, so that we can
if to_bootstrap == "pytest":
sys.modules["alembic_plugin_base"] = load_file_as_module("plugin_base")
sys.modules["alembic_pytestplugin"] = load_file_as_module("pytestplugin")
-elif to_bootstrap == "nose":
- sys.modules["alembic_plugin_base"] = load_file_as_module("plugin_base")
- sys.modules["alembic_noseplugin"] = load_file_as_module("noseplugin")
else:
raise Exception("unknown bootstrap: %s" % to_bootstrap) # noqa
+++ /dev/null
-# plugin/noseplugin.py
-# Copyright (C) 2005-2017 the SQLAlchemy authors and contributors
-# <see AUTHORS file>
-#
-# This module is part of SQLAlchemy and is released under
-# the MIT License: http://www.opensource.org/licenses/mit-license.php
-
-"""
-Enhance nose with extra options and behaviors for running SQLAlchemy tests.
-
-
-NOTE: copied/adapted from SQLAlchemy master for backwards compatibility;
-this should be removable when Alembic targets SQLAlchemy 1.0.0.
-
-"""
-
-try:
- # installed by bootstrap.py
- import alembic_plugin_base as plugin_base
-except ImportError:
- # assume we're a package, use traditional import
- from . import plugin_base
-
-import os
-import sys
-
-from nose.plugins import Plugin
-
-fixtures = None
-
-py3k = sys.version_info.major >= 3
-
-
-class NoseSQLAlchemy(Plugin):
- enabled = True
-
- name = "sqla_testing"
- score = 100
-
- def options(self, parser, env=os.environ):
- Plugin.options(self, parser, env)
- opt = parser.add_option
-
- def make_option(name, **kw):
- callback_ = kw.pop("callback", None)
- if callback_:
-
- def wrap_(option, opt_str, value, parser):
- callback_(opt_str, value, parser)
-
- kw["callback"] = wrap_
- opt(name, **kw)
-
- plugin_base.setup_options(make_option)
- plugin_base.read_config()
-
- def configure(self, options, conf):
- super(NoseSQLAlchemy, self).configure(options, conf)
- plugin_base.pre_begin(options)
-
- plugin_base.set_coverage_flag(options.enable_plugin_coverage)
-
- def begin(self):
- global fixtures
- from alembic.testing import fixtures # noqa
-
- plugin_base.post_begin()
-
- def describeTest(self, test):
- return ""
-
- def wantFunction(self, fn):
- return False
-
- def wantMethod(self, fn):
- if py3k:
- if not hasattr(fn.__self__, "cls"):
- return False
- cls = fn.__self__.cls
- else:
- cls = fn.im_class
- return plugin_base.want_method(cls, fn)
-
- def wantClass(self, cls):
- return plugin_base.want_class(cls)
-
- def beforeTest(self, test):
- plugin_base.before_test(
- test,
- test.test.cls.__module__,
- test.test.cls,
- test.test.method.__name__,
- )
-
- def afterTest(self, test):
- plugin_base.after_test(test)
-
- def startContext(self, ctx):
- if not isinstance(ctx, type) or not issubclass(ctx, fixtures.TestBase):
- return
- plugin_base.start_test_class(ctx)
-
- def stopContext(self, ctx):
- if not isinstance(ctx, type) or not issubclass(ctx, fixtures.TestBase):
- return
- plugin_base.stop_test_class(ctx)
"""Testing extensions.
this module is designed to work as a testing-framework-agnostic library,
-so that we can continue to support nose and also begin adding new
-functionality via py.test.
-
-NOTE: copied/adapted from SQLAlchemy master for backwards compatibility;
-this should be removable when Alembic targets SQLAlchemy 1.0.0
+however it currenty targets only py.test.
"""
import re
import sys
-try:
- # unitttest has a SkipTest also but pytest doesn't
- # honor it unless nose is imported too...
- from nose import SkipTest
-except ImportError:
- from pytest import skip
+from pytest import skip
- SkipTest = skip.Exception
+SkipTest = skip.Exception
py3k = sys.version_info.major >= 3
for fn in post_configure:
fn(options, file_config)
- # late imports, has to happen after config as well
- # as nose plugins like coverage
global util, fixtures, engines, exclusions, assertions
global warnings, profiling, config, testing
from alembic.testing import config, warnings, exclusions # noqa
from alembic.testing import config
from alembic.testing.exclusions import against
from sqlalchemy import schema
- from alembic import util
-
from sqlalchemy import inspect
if options.dropfirst:
)
)
- if against(cfg, "postgresql") and util.sqla_100:
+ if against(cfg, "postgresql"):
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
import sys
+from sqlalchemy.testing.requirements import Requirements
+
from alembic import util
from alembic.util import sqla_compat
from . import exclusions
-if util.sqla_094:
- from sqlalchemy.testing.requirements import Requirements
-else:
-
- class Requirements(object):
- pass
-
class SuiteRequirements(Requirements):
@property
"SQLAlchemy 1.2.16 or greater required",
)
- @property
- def sqlalchemy_10(self):
- return exclusions.skip_if(
- lambda config: not util.sqla_100,
- "SQLAlchemy 1.0.0 or greater required",
- )
-
- @property
- def fail_before_sqla_100(self):
- return exclusions.fails_if(
- lambda config: not util.sqla_100,
- "SQLAlchemy 1.0.0 or greater required",
- )
-
- @property
- def fail_before_sqla_1010(self):
- return exclusions.fails_if(
- lambda config: not util.sqla_1010,
- "SQLAlchemy 1.0.10 or greater required",
- )
-
- @property
- def fail_before_sqla_099(self):
- return exclusions.fails_if(
- lambda config: not util.sqla_099,
- "SQLAlchemy 0.9.9 or greater required",
- )
-
- @property
- def fail_before_sqla_110(self):
- return exclusions.fails_if(
- lambda config: not util.sqla_110,
- "SQLAlchemy 1.1.0 or greater required",
- )
-
- @property
- def sqlalchemy_092(self):
- return exclusions.skip_if(
- lambda config: not util.sqla_092,
- "SQLAlchemy 0.9.2 or greater required",
- )
-
- @property
- def sqlalchemy_094(self):
- return exclusions.skip_if(
- lambda config: not util.sqla_094,
- "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(
- lambda config: not util.sqla_100,
- "SQLAlchemy 1.0.0 or greater required",
- )
-
- @property
- def sqlalchemy_1014(self):
- return exclusions.skip_if(
- lambda config: not util.sqla_1014,
- "SQLAlchemy 1.0.14 or greater required",
- )
-
@property
def sqlalchemy_1115(self):
return exclusions.skip_if(
+++ /dev/null
-#!/usr/bin/env python
-# testing/runner.py
-# Copyright (C) 2005-2017 the SQLAlchemy authors and contributors
-# <see AUTHORS file>
-#
-# This module is part of SQLAlchemy and is released under
-# the MIT License: http://www.opensource.org/licenses/mit-license.php
-"""
-Nose test runner module.
-
-This script is a front-end to "nosetests" which
-installs SQLAlchemy's testing plugin into the local environment.
-
-The script is intended to be used by third-party dialects and extensions
-that run within SQLAlchemy's testing framework. The runner can
-be invoked via::
-
- python -m alembic.testing.runner
-
-The script is then essentially the same as the "nosetests" script, including
-all of the usual Nose options. The test environment requires that a
-setup.cfg is locally present including various required options.
-
-Note that when using this runner, Nose's "coverage" plugin will not be
-able to provide coverage for SQLAlchemy itself, since SQLAlchemy is
-imported into sys.modules before coverage is started. The special
-script sqla_nose.py is provided as a top-level script which loads the
-plugin in a special (somewhat hacky) way so that coverage against
-SQLAlchemy itself is possible.
-
-"""
-import nose
-
-from .plugin.noseplugin import NoseSQLAlchemy
-
-
-def main():
- nose.main(addplugins=[NoseSQLAlchemy()])
-
-
-def setup_py_test():
- """Runner to use for the 'test_suite' entry of your setup.py.
-
- Prevents any name clash shenanigans from the command line
- argument "test" that the "setup.py test" command sends
- to nose.
-
- """
- nose.main(addplugins=[NoseSQLAlchemy()], argv=["runner"])
warnings.filterwarnings("error", category=sa_exc.SAWarning)
warnings.filterwarnings("error", category=DeprecationWarning)
+ # temporary to allow SQLAlchemy 1.1 to pass under python 3
+ warnings.filterwarnings(
+ "ignore", category=DeprecationWarning, message=".*formatargspec"
+ )
+
def assert_warnings(fn, warning_msgs, regex=False):
"""Assert that each of the given warnings are emitted by fn."""
from .pyfiles import load_python_file # noqa
from .pyfiles import pyc_file_from_path # noqa
from .pyfiles import template_to_file # noqa
-from .sqla_compat import sqla_09 # noqa
-from .sqla_compat import sqla_092 # noqa
-from .sqla_compat import sqla_094 # noqa
-from .sqla_compat import sqla_099 # noqa
-from .sqla_compat import sqla_100 # noqa
-from .sqla_compat import sqla_1010 # noqa
-from .sqla_compat import sqla_1014 # noqa
-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:
- raise CommandError("SQLAlchemy 0.9.0 or greater is required. ")
+if not sqla_110:
+ raise CommandError("SQLAlchemy 1.1.0 or greater is required. ")
import io
import sys
-if sys.version_info < (2, 7):
- raise NotImplementedError("Python 2.7 or greater is required.")
-
py27 = sys.version_info >= (2, 7)
py2k = sys.version_info.major < 3
py3k = sys.version_info.major >= 3
if py2k:
# in Py2K, the io.* package is awkward because it does not
# easily wrap the file type (e.g. sys.stdout) and I can't
- # figure out at all how to wrap StringIO.StringIO (used by nosetests)
+ # figure out at all how to wrap StringIO.StringIO
# and also might be user specified too. So create a full
# adapter.
_vers = tuple(
[_safe_int(x) for x in re.findall(r"(\d+|[abc]\d)", __version__)]
)
-sqla_09 = _vers >= (0, 9, 0)
-sqla_092 = _vers >= (0, 9, 2)
-sqla_094 = _vers >= (0, 9, 4)
-sqla_094 = _vers >= (0, 9, 4)
-sqla_099 = _vers >= (0, 9, 9)
-sqla_100 = _vers >= (1, 0, 0)
-sqla_105 = _vers >= (1, 0, 5)
-sqla_1010 = _vers >= (1, 0, 10)
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:
- AUTOINCREMENT_DEFAULT = "auto"
-else:
- AUTOINCREMENT_DEFAULT = True
+AUTOINCREMENT_DEFAULT = "auto"
def _table_for_constraint(constraint):
def _fk_spec(constraint):
- if sqla_100:
- source_columns = [
- constraint.columns[key].name for key in constraint.column_keys
- ]
- else:
- source_columns = [
- element.parent.name for element in constraint.elements
- ]
+ source_columns = [
+ constraint.columns[key].name for key in constraint.column_keys
+ ]
source_table = constraint.parent.name
source_schema = constraint.parent.schema
def _is_type_bound(constraint):
# this deals with SQLAlchemy #3260, don't copy CHECK constraints
# that will be generated by the type.
- if sqla_100:
- # new feature added for #3260
- return constraint._type_bound
- else:
- # old way, look at what we know Boolean/Enum to use
- return constraint._create_rule is not None and isinstance(
- getattr(constraint._create_rule, "target", None),
- sqltypes.SchemaType,
- )
+ # new feature added for #3260
+ return constraint._type_bound
def _find_columns(clause):
==========
.. changelog::
- :version: 1.0.12
+ :version: 1.1.0
:include_notes_from: unreleased
.. changelog::
--- /dev/null
+.. change::
+ :tags: change
+
+ Alembic 1.1 bumps the minimum version of SQLAlchemy to 1.1. As was the
+ case before, Python requirements remain at Python 2.7, or in the 3.x series
+ Python 3.4.
readme = os.path.join(os.path.dirname(__file__), "README.rst")
requires = [
- "SQLAlchemy>=0.9.0",
+ "SQLAlchemy>=1.1.0",
"Mako",
"python-editor>=0.3",
"python-dateutil",
eq_([elem.column.name for elem in diff[1].elements], target_columns)
if conditional_name is not None:
- if config.requirements.no_fk_names.enabled:
- eq_(diff[1].name, None)
- elif conditional_name == "servergenerated":
+ if conditional_name == "servergenerated":
fks = Inspector.from_engine(self.bind).get_foreign_keys(
source_table
)
-from alembic import util
from alembic.testing import exclusions
from alembic.testing.requirements import SuiteRequirements
from alembic.util import sqla_compat
)
)
- @property
- def no_fk_names(self):
- """foreign key constraints *never* have names in the DB"""
-
- return exclusions.only_if(
- lambda config: exclusions.against(config, "sqlite")
- and not util.sqla_100
- )
-
@property
def check_constraints_w_enforcement(self):
return exclusions.fails_on("mysql")
@property
def reflects_fk_options(self):
- return exclusions.only_on(
- [
- "postgresql",
- "mysql",
- lambda config: util.sqla_110
- and exclusions.against(config, "sqlite"),
- ]
- )
+ return exclusions.only_on(["postgresql", "mysql", "sqlite"])
@property
def fk_initially(self):
"""Target driver reflects the name of primary key constraints."""
return exclusions.fails_on_everything_except(
- "postgresql",
- "oracle",
- "mssql",
- "sybase",
- lambda config: (
- util.sqla_110 and exclusions.against(config, "sqlite")
- ),
+ "postgresql", "oracle", "mssql", "sybase", "sqlite"
)
@property
class AutogenerateVariantCompareTest(AutogenTest, TestBase):
__backend__ = True
- # 1.0.13 and lower fail on Postgresql due to variant / bigserial issue
- # #3739
-
- __requires__ = ("sqlalchemy_1014",)
-
@classmethod
def _get_db_schema(cls):
m = MetaData()
ops = self._fixture(m1, m2, return_ops=True)
assert "autoincrement" not in ops.ops[0].ops[0].kw
- @config.requirements.fail_before_sqla_110
def test_alter_column_autoincrement_nonpk_explicit_true(self):
m1 = MetaData()
m2 = MetaData()
set([diffs[0][1].name, diffs[1][1].name]), set(["xy_idx", "y_idx"])
)
- # this simply doesn't fully work before we had
- # effective deduping of indexes/uniques.
- @config.requirements.sqlalchemy_100
def test_drop_table_w_uq_constraint(self):
m1 = MetaData()
m2 = MetaData()
diffs = self._fixture(m1, m2, include_schemas=True)
eq_(diffs, [])
- @config.requirements.sqlalchemy_100
@config.requirements.btree_gist
def test_exclude_const_unchanged(self):
from sqlalchemy.dialects.postgresql import TSRANGE, ExcludeConstraint
idx = Index("test_lower_code_idx", cast(t.c.code, String))
op_obj = ops.CreateIndexOp.from_index(idx)
- if config.requirements.sqlalchemy_110.enabled:
- eq_ignore_whitespace(
- autogenerate.render_op_text(self.autogen_context, op_obj),
- "op.create_index('test_lower_code_idx', 'test', "
- "[sa.text(!U'CAST(code AS VARCHAR)')], unique=False)",
- )
- else:
- eq_ignore_whitespace(
- autogenerate.render_op_text(self.autogen_context, op_obj),
- "op.create_index('test_lower_code_idx', 'test', "
- "[sa.text(!U'CAST(code AS VARCHAR)')], unique=False)",
- )
+ eq_ignore_whitespace(
+ autogenerate.render_op_text(self.autogen_context, op_obj),
+ "op.create_index('test_lower_code_idx', 'test', "
+ "[sa.text(!U'CAST(code AS VARCHAR)')], unique=False)",
+ )
def test_render_add_index_desc(self):
m = MetaData()
"sa.PrimaryKeyConstraint('x'))",
)
- @config.requirements.fail_before_sqla_110
def test_render_table_w_autoincrement(self):
m = MetaData()
t = Table(
result,
"sa.Column('some_key', sa.Integer(), "
"nullable=True, "
- "comment=\"This is a john's comment\")",
+ 'comment="This is a john\'s comment")',
)
def test_render_col_autoinc_false_mysql(self):
"sa.Enum('one', 'two', 'three')",
)
- @config.requirements.sqlalchemy_099
def test_render_non_native_enum(self):
eq_ignore_whitespace(
autogenerate.render._repr_type(
"sa.Integer()",
)
- @config.requirements.sqlalchemy_110
def test_generic_array_type(self):
eq_ignore_whitespace(
"sa.ARRAY(sa.DateTime(timezone=True))",
)
- @config.requirements.sqlalchemy_110
def test_render_array_no_context(self):
uo = ops.UpgradeOps(
ops=[
class RenderNamingConventionTest(TestBase):
- __requires__ = ("sqlalchemy_094",)
-
def setUp(self):
convention = {
{"id": 6, "data": 5, "x": -2},
)
- @config.requirements.sqlalchemy_094
@config.requirements.unnamed_constraints
def test_drop_foreign_key(self):
bar = Table(
@classmethod
def setup_class(cls):
staging_env()
- if util.sqla_105:
- directives = "sqlalchemy.legacy_schema_aliasing=false"
- else:
- directives = ""
+ directives = "sqlalchemy.legacy_schema_aliasing=false"
cls.cfg = cfg = _no_sql_testing_config("mssql", directives)
cls.a, cls.b, cls.c = three_rev_fixture(cfg)
from sqlalchemy.sql import column
from sqlalchemy.sql import func
from sqlalchemy.sql import text
+from sqlalchemy.sql.schema import quoted_name
from alembic import op
from alembic.operations import ops
)
def test_add_column_schema_hard_quoting(self):
- from sqlalchemy.sql.schema import quoted_name
context = op_fixture("postgresql")
op.add_column(
)
def test_rename_table_schema_hard_quoting(self):
- from sqlalchemy.sql.schema import quoted_name
context = op_fixture("postgresql")
op.rename_table(
context.assert_('ALTER TABLE "some.schema".t1 RENAME TO t2')
def test_add_constraint_schema_hard_quoting(self):
- from sqlalchemy.sql.schema import quoted_name
context = op_fixture("postgresql")
op.create_check_constraint(
class AutoNamingConventionTest(TestBase):
- __requires__ = ("sqlalchemy_094",)
-
def test_add_check_constraint(self):
context = op_fixture(
naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
from sqlalchemy import types
from sqlalchemy.dialects.postgresql import ARRAY
from sqlalchemy.dialects.postgresql import BYTEA
+from sqlalchemy.dialects.postgresql import HSTORE
+from sqlalchemy.dialects.postgresql import JSON
+from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.engine.reflection import Inspector
from sqlalchemy.sql import column
from alembic.testing.fixtures import TestBase
-if util.sqla_09:
- from sqlalchemy.dialects.postgresql import JSON, JSONB, HSTORE
-
-
class PostgresqlOpTest(TestBase):
def test_rename_table_postgresql(self):
context = op_fixture("postgresql")
"WHERE locations.coordinates != Null"
)
- @config.requirements.fail_before_sqla_099
def test_create_index_postgresql_concurrently(self):
context = op_fixture("postgresql")
op.create_index(
"CREATE INDEX CONCURRENTLY geocoded ON locations (coordinates)"
)
- @config.requirements.fail_before_sqla_110
def test_drop_index_postgresql_concurrently(self):
context = op_fixture("postgresql")
op.drop_index("geocoded", "locations", postgresql_concurrently=True)
op.add_column("some_table", Column("q", Integer, primary_key=True))
context.assert_("ALTER TABLE some_table ADD COLUMN q SERIAL NOT NULL")
- @config.requirements.fail_before_sqla_100
def test_create_exclude_constraint(self):
context = op_fixture("postgresql")
op.create_exclude_constraint(
"WHERE (x > 5)"
)
- @config.requirements.fail_before_sqla_100
def test_create_exclude_constraint_quoted_literal(self):
context = op_fixture("postgresql")
op.create_exclude_constraint(
'("SomeColumn" WITH >) WHERE ("SomeColumn" > 5)'
)
- @config.requirements.fail_before_sqla_1010
def test_create_exclude_constraint_quoted_column(self):
context = op_fixture("postgresql")
op.create_exclude_constraint(
DateTime(), text("TIMEZONE('utc', CURRENT_TIMESTAMP)")
)
- @config.requirements.sqlalchemy_10
def test_compare_current_timestamp_fn_w_binds(self):
self._compare_default_roundtrip(
DateTime(), func.timezone("utc", func.current_timestamp())
in self.autogen_context.imports
)
- @config.requirements.sqlalchemy_110
def test_postgresql_hstore_subtypes(self):
eq_ignore_whitespace(
autogenerate.render._repr_type(HSTORE(), self.autogen_context),
in self.autogen_context.imports
)
- @config.requirements.sqlalchemy_110
def test_generic_array_type(self):
eq_ignore_whitespace(
"postgresql.ARRAY(foobar.MYVARCHAR)",
)
- @config.requirements.fail_before_sqla_100
def test_add_exclude_constraint(self):
from sqlalchemy.dialects.postgresql import ExcludeConstraint
"where=sa.text(!U'x != 2'), using='gist')",
)
- @config.requirements.fail_before_sqla_100
def test_add_exclude_constraint_case_sensitive(self):
from sqlalchemy.dialects.postgresql import ExcludeConstraint
"where=sa.text(!U'\"XColumn\" != 2'), using='gist')",
)
- @config.requirements.fail_before_sqla_100
def test_inline_exclude_constraint(self):
from sqlalchemy.dialects.postgresql import ExcludeConstraint
")",
)
- @config.requirements.fail_before_sqla_100
def test_inline_exclude_constraint_case_sensitive(self):
from sqlalchemy.dialects.postgresql import ExcludeConstraint
)
def test_json_type(self):
- if config.requirements.sqlalchemy_110.enabled:
- eq_ignore_whitespace(
- autogenerate.render._repr_type(JSON(), self.autogen_context),
- "postgresql.JSON(astext_type=sa.Text())",
- )
- else:
- eq_ignore_whitespace(
- autogenerate.render._repr_type(JSON(), self.autogen_context),
- "postgresql.JSON()",
- )
+ eq_ignore_whitespace(
+ autogenerate.render._repr_type(JSON(), self.autogen_context),
+ "postgresql.JSON(astext_type=sa.Text())",
+ )
def test_jsonb_type(self):
- if config.requirements.sqlalchemy_110.enabled:
- eq_ignore_whitespace(
- autogenerate.render._repr_type(JSONB(), self.autogen_context),
- "postgresql.JSONB(astext_type=sa.Text())",
- )
- else:
- eq_ignore_whitespace(
- autogenerate.render._repr_type(JSONB(), self.autogen_context),
- "postgresql.JSONB()",
- )
+ eq_ignore_whitespace(
+ autogenerate.render._repr_type(JSONB(), self.autogen_context),
+ "postgresql.JSONB(astext_type=sa.Text())",
+ )
[tox]
-# current mysqlclient fails with <=SQLA 0.9 on py3k due to
-# old unicode statements flag
-envlist = py{27,34,35,36,37}-sqla{10,11,12,13,master}, py{27}-sqla{079,084,09}
+envlist = py{27,34,35,36,37}-sqla{11,12,13,master}
SQLA_REPO = {env:SQLA_REPO:git+https://github.com/sqlalchemy/sqlalchemy.git}
deps=pytest!=3.9.1,!=3.9.2
pytest-xdist
mock
- sqla079: {[tox]SQLA_REPO}@rel_0_7_9
- sqla084: {[tox]SQLA_REPO}@rel_0_8_4
- sqla09: {[tox]SQLA_REPO}@rel_0_9
- sqla10: {[tox]SQLA_REPO}@rel_1_0
sqla11: {[tox]SQLA_REPO}@rel_1_1
sqla12: {[tox]SQLA_REPO}@rel_1_2
sqla13: {[tox]SQLA_REPO}@rel_1_3
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
