template_args = {}
imports = set()
if autogenerate:
+ util.requires_07("autogenerate")
def retrieve_migrations(rev):
if script._get_rev(rev) is not script._get_rev("head"):
raise util.CommandError("Target database is not up to date.")
self._start_from_rev = starting_rev
self.impl = ddl.DefaultImpl.get_by_dialect(dialect)(
- dialect, connection, self.as_sql,
+ dialect, self.connection, self.as_sql,
transactional_ddl,
self.output_buffer
)
from sqlalchemy.ext.compiler import compiles
from sqlalchemy import schema
from alembic.ddl import base
+from alembic import util
from sqlalchemy import types as sqltypes
class ImplMeta(type):
transactional_ddl = False
- def __init__(self, dialect, connection, as_sql, transactional_ddl, output_buffer):
+ def __init__(self, dialect, connection, as_sql,
+ transactional_ddl, output_buffer):
self.dialect = dialect
self.connection = connection
self.as_sql = as_sql
self.output_buffer = output_buffer
+ self.memo = {}
if transactional_ddl is not None:
self.transactional_ddl = transactional_ddl
def static_output(self, text):
self.output_buffer.write(text + "\n\n")
+ @property
+ def bind(self):
+ return self.connection
+
def _exec(self, construct, *args, **kw):
if isinstance(construct, basestring):
construct = text(construct)
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)
self._exec(schema.CreateTable(table))
+ if util.sqla_07:
+ table.dispatch.after_create(table, self.connection,
+ checkfirst=False,
+ _ddl_runner=self)
for index in table.indexes:
self._exec(schema.CreateIndex(index))
Column('description', NVARCHAR(200))
)
+ :param name: Name of the table
+ :param \*columns: collection of :class:`~sqlalchemy.schema.Column` objects within
+ the table, as well as optional :class:`~sqlalchemy.schema.Constraint` objects
+ and :class:`~.sqlalchemy.schema.Index` objects.
+ :param emit_events: if ``True``, emit ``before_create`` and ``after_create``
+ events when the table is being created. In particular, the Postgresql ENUM
+ type will emit a CREATE TYPE within these events.
+ :param \**kw: Other keyword arguments are passed to the underlying
+ :class:`.Table` object created for the command.
+
"""
-
get_impl().create_table(
_table(name, *columns, **kw)
)
import random
import uuid
-
class CommandError(Exception):
pass
+from sqlalchemy import __version__
+_vers = tuple([int(x) for x in __version__.split(".")])
+sqla_06 = _vers > (0, 6)
+sqla_07 = _vers > (0, 7)
+if not sqla_06:
+ raise CommandError(
+ "SQLAlchemy 0.6 or greater is required. "
+ "Version 0.7 or above required for full featureset.")
+
+def requires_07(feature):
+ if not sqla_07:
+ raise CommandError(
+ "The %s feature requires "
+ "SQLAlchemy 0.7 or greater."
+ % feature
+ )
try:
width = int(os.environ['COLUMNS'])
except (KeyError, ValueError):
import ConfigParser
from nose import SkipTest
from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.util import decorator
staging_directory = os.path.join(os.path.dirname(__file__), 'scratch')
files_directory = os.path.join(os.path.dirname(__file__), 'files')
_engs[name] = eng
return eng
+@decorator
+def requires_07(fn, *arg, **kw):
+ if not util.sqla_07:
+ raise SkipTest("SQLAlchemy 0.7 required")
+ return fn(*arg, **kw)
+
_dialects = {}
def _get_dialect(name):
if name is None or name == 'default':
self.assertion = []
self.dialect = dialect
self.as_sql = as_sql
-
+ # TODO: this might need to
+ # be more like a real connection
+ # as tests get more involved
+ self.connection = None
def _exec(self, construct, *args, **kw):
if isinstance(construct, basestring):
construct = text(construct)
from sqlalchemy.types import NULLTYPE
from alembic import autogenerate, context
from unittest import TestCase
-from tests import staging_env, sqlite_db, clear_staging_env, eq_, eq_ignore_whitespace
+from tests import staging_env, sqlite_db, clear_staging_env, eq_, \
+ eq_ignore_whitespace, requires_07
def _model_one():
m = MetaData()
class AutogenerateDiffTest(TestCase):
@classmethod
+ @requires_07
def setup_class(cls):
staging_env()
cls.bind = sqlite_db()
"""test individual directives"""
@classmethod
+ @requires_07
def setup_class(cls):
context._context_opts['sqlalchemy_module_prefix'] = 'sa.'
-from tests import op_fixture, db_for_dialect, eq_, staging_env, clear_staging_env
+from tests import op_fixture, db_for_dialect, eq_, staging_env, \
+ clear_staging_env, no_sql_testing_config,\
+ capture_context_buffer, requires_07
from unittest import TestCase
from sqlalchemy import DateTime, MetaData, Table, Column, text, Integer, String
from sqlalchemy.engine.reflection import Inspector
-from alembic import context
+from alembic import context, command, util
+from alembic.script import ScriptDirectory
+
+class PGOfflineEnumTest(TestCase):
+ @requires_07
+ def setUp(self):
+ env = staging_env()
+ self.cfg = cfg = no_sql_testing_config()
+
+ self.rid = rid = util.rev_id()
+
+ self.script = script = ScriptDirectory.from_config(cfg)
+ script.generate_rev(rid, None, refresh=True)
+
+ def _inline_enum_script(self):
+ self.script.write(self.rid, """
+down_revision = None
+
+from alembic.op import *
+from sqlalchemy.dialects.postgresql import ENUM
+from sqlalchemy import Column
+
+def upgrade():
+ create_table("sometable",
+ Column("data", ENUM("one", "two", "three", name="pgenum"))
+ )
+
+def downgrade():
+ drop_table("sometable")
+""")
+
+ def _distinct_enum_script(self):
+ self.script.write(self.rid, """
+down_revision = None
+
+from alembic.op import *
+from sqlalchemy.dialects.postgresql import ENUM
+from sqlalchemy import Column
+
+def upgrade():
+ enum = ENUM("one", "two", "three", name="pgenum", create_type=False)
+ enum.create(get_bind(), checkfirst=False)
+ create_table("sometable",
+ Column("data", enum)
+ )
+
+def downgrade():
+ drop_table("sometable")
+ ENUM(name="pgenum").drop(get_bind(), checkfirst=False)
+
+""")
+
+ def tearDown(self):
+ clear_staging_env()
+
+ def test_offline_inline_enum_create(self):
+ self._inline_enum_script()
+ with capture_context_buffer() as buf:
+ command.upgrade(self.cfg, self.rid, sql=True)
+ assert "CREATE TYPE pgenum AS ENUM ('one','two','three')" in buf.getvalue()
+ assert "CREATE TABLE sometable (\n data pgenum\n)" in buf.getvalue()
+
+ def test_offline_inline_enum_drop(self):
+ self._inline_enum_script()
+ with capture_context_buffer() as buf:
+ command.downgrade(self.cfg, "%s:base" % self.rid, sql=True)
+ assert "DROP TABLE sometable" in buf.getvalue()
+ # no drop since we didn't emit events
+ assert "DROP TYPE pgenum" not in buf.getvalue()
+
+ def test_offline_distinct_enum_create(self):
+ self._distinct_enum_script()
+ with capture_context_buffer() as buf:
+ command.upgrade(self.cfg, self.rid, sql=True)
+ assert "CREATE TYPE pgenum AS ENUM ('one','two','three')" in buf.getvalue()
+ assert "CREATE TABLE sometable (\n data pgenum\n)" in buf.getvalue()
+
+ def test_offline_distinct_enum_drop(self):
+ self._distinct_enum_script()
+ with capture_context_buffer() as buf:
+ command.downgrade(self.cfg, "%s:base" % self.rid, sql=True)
+ assert "DROP TABLE sometable" in buf.getvalue()
+ assert "DROP TYPE pgenum" in buf.getvalue()
+
+
class PostgresqlDefaultCompareTest(TestCase):
@classmethod
assert self._compare_default(
t, t2, t2.c.somecol, alternate
) is expected
-# t.create(self.bind)
-# insp = Inspector.from_engine(self.bind)
-# cols = insp.get_columns("test")
-# ctx = context.get_context()
-# assert ctx.impl.compare_server_default(
-# cols[0],
-# t2.c.somecol,
-# alternate) is expected
def _compare_default(
self,
-from tests import clear_staging_env, staging_env, no_sql_testing_config, sqlite_db, eq_, ne_, capture_context_buffer, three_rev_fixture
+from tests import clear_staging_env, staging_env, \
+ no_sql_testing_config, sqlite_db, eq_, ne_, capture_context_buffer, \
+ three_rev_fixture
from alembic import command, util
def setup():
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
