self.dialect = dialect
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 = {}
self.context_opts = context_opts
if transactional_ddl is not None:
self.transactional_ddl = transactional_ddl
+ if self.literal_binds:
+ if not self.as_sql:
+ raise util.CommandError(
+ "Can't use literal_binds setting without as_sql mode")
+
@classmethod
def get_by_dialect(cls, dialect):
return _impls[dialect.name]
if multiparams or params:
# TODO: coverage
raise Exception("Execution arguments not allowed with as_sql")
+
+ if self.literal_binds and not isinstance(
+ construct, schema.DDLElement):
+ compile_kw = dict(compile_kwargs={"literal_binds": True})
+ else:
+ compile_kw = {}
+
self.static_output(text_type(
- construct.compile(dialect=self.dialect)
+ construct.compile(dialect=self.dialect, **compile_kw)
).replace("\t", " ").strip() + self.command_terminator)
else:
conn = self.connection
compare_type=False,
compare_server_default=False,
render_item=None,
+ literal_binds=False,
upgrade_token="upgrades",
downgrade_token="downgrades",
alembic_module_prefix="op.",
object.
:param output_encoding: when using ``--sql`` to generate SQL
scripts, apply this encoding to the string output.
+ :param literal_binds: when using ``--sql`` to generate SQL
+ scripts, pass through the ``literal_binds`` flag to the compiler
+ so that any literal values that would ordinarily be bound
+ parameters are converted to plain strings.
+
+ .. warning:: Dialects can typically only handle simple datatypes
+ like strings and numbers for auto-literal generation. Datatypes
+ like dates, intervals, and others may still require manual
+ formatting, typically using :meth:`.Operations.inline_literal`.
+
+ .. note:: the ``literal_binds`` flag is ignored on SQLAlchemy
+ versions prior to 0.8 where this feature is not supported.
+
+ .. versionadded:: 0.7.6
+
+ .. seealso::
+
+ :meth:`.Operations.inline_literal`
:param starting_rev: Override the "starting revision" argument
when using ``--sql`` mode.
opts['sqlalchemy_module_prefix'] = sqlalchemy_module_prefix
opts['alembic_module_prefix'] = alembic_module_prefix
opts['user_module_prefix'] = user_module_prefix
+ opts['literal_binds'] = literal_binds
if render_item is not None:
opts['render_item'] = render_item
if compare_type is not None:
from contextlib import contextmanager
from sqlalchemy import MetaData, Table, Column, String, literal_column
-from sqlalchemy import create_engine
+from sqlalchemy.engine.strategies import MockEngineStrategy
from sqlalchemy.engine import url as sqla_url
from .compat import callable, EncodedIO
def dump(construct, *multiparams, **params):
self.impl._exec(construct)
- return create_engine("%s://" % self.dialect.name,
- strategy="mock", executor=dump)
+ return MockEngineStrategy.MockConnection(self.dialect, dump)
@property
def bind(self):
See :meth:`.execute` for an example usage of
:meth:`.inline_literal`.
+ The environment can also be configured to attempt to render
+ "literal" values inline automatically, for those simple types
+ that are supported by the dialect; see
+ :paramref:`.EnvironmentContext.configure.literal_binds` for this
+ more recently added feature.
+
:param value: The value to render. Strings, integers, and simple
numerics should be supported. Other types like boolean,
dates, etc. may or may not be supported yet by various
from the Python type of the value itself, as well as
based on the context in which the value is used.
+ .. seealso::
+
+ :paramref:`.EnvironmentContext.configure.literal_binds`
+
"""
return impl._literal_bindparam(None, value, type_=type_)
"""
url = config.get_main_option("sqlalchemy.url")
- context.configure(url=url, target_metadata=target_metadata)
+ context.configure(
+ url=url, target_metadata=target_metadata, literal_binds=True)
with context.begin_transaction():
context.run_migrations()
logger.info("Writing output to %s" % file_)
with open(file_, 'w') as buffer:
context.configure(url=rec['url'], output_buffer=buffer,
- target_metadata=target_metadata.get(name))
+ target_metadata=target_metadata.get(name),
+ literal_binds=True)
with context.begin_transaction():
context.run_migrations(engine_name=name)
"""
context.configure(
- url=meta.engine.url, target_metadata=target_metadata)
+ url=meta.engine.url, target_metadata=target_metadata,
+ literal_binds=True)
with context.begin_transaction():
context.run_migrations()
yield buf
-def op_fixture(dialect='default', as_sql=False, naming_convention=None):
- impl = _impls[dialect]
-
- class Impl(impl):
-
- def __init__(self, dialect, as_sql):
- 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
- if as_sql and self.dialect.name != 'default':
- # act similarly to MigrationContext
- def dump(construct, *multiparams, **params):
- self._exec(construct)
-
- self.connection = create_engine(
- "%s://" % self.dialect.name,
- strategy="mock", executor=dump)
-
- else:
- self.connection = mock.Mock(dialect=dialect)
-
- def _exec(self, construct, *args, **kw):
- if isinstance(construct, string_types):
- construct = text(construct)
- assert construct.supports_execution
- sql = text_type(construct.compile(dialect=self.dialect))
- sql = re.sub(r'[\n\t]', '', sql)
- self.assertion.append(
- sql
- )
+def op_fixture(
+ dialect='default', as_sql=False,
+ naming_convention=None, literal_binds=False):
opts = {}
if naming_convention:
"sqla 0.9.2 or greater")
opts['target_metadata'] = MetaData(naming_convention=naming_convention)
- class ctx(MigrationContext):
+ class buffer_(object):
+ def __init__(self):
+ self.lines = []
+
+ def write(self, msg):
+ msg = msg.strip()
+ msg = re.sub(r'[\n\t]', '', msg)
+ if as_sql:
+ # the impl produces soft tabs,
+ # so search for blocks of 4 spaces
+ msg = re.sub(r' ', '', msg)
+ msg = re.sub('\;\n*$', '', msg)
+
+ self.lines.append(msg)
+
+ def flush(self):
+ pass
- 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
+ buf = buffer_()
+ class ctx(MigrationContext):
def clear_assertions(self):
- self.impl.assertion[:] = []
+ buf.lines[:] = []
def assert_(self, *sql):
# TODO: make this more flexible about
# whitespace and such
- eq_(self.impl.assertion, list(sql))
+ eq_(buf.lines, list(sql))
def assert_contains(self, sql):
- for stmt in self.impl.assertion:
+ for stmt in buf.lines:
if sql in stmt:
return
else:
assert False, "Could not locate fragment %r in %r" % (
sql,
- self.impl.assertion
+ buf.lines
)
- context = ctx(dialect, as_sql)
+
+ if as_sql:
+ opts['as_sql'] = as_sql
+ if literal_binds:
+ opts['literal_binds'] = literal_binds
+ ctx_dialect = _get_dialect(dialect)
+ if not as_sql:
+ def execute(stmt, *multiparam, **param):
+ if isinstance(stmt, string_types):
+ stmt = text(stmt)
+ assert stmt.supports_execution
+ sql = text_type(stmt.compile(dialect=ctx_dialect))
+
+ buf.write(sql)
+
+ connection = mock.Mock(dialect=ctx_dialect, execute=execute)
+ else:
+ opts['output_buffer'] = buf
+ connection = None
+ context = ctx(
+ ctx_dialect,
+ connection,
+ opts)
+
alembic.op._proxy = Operations(context)
return context
-
.. changelog::
:version: 0.7.6
+ .. change::
+ :tags: feature, operations
+ :tickets: 255
+
+ Added a new option
+ :paramref:`.EnvironmentContext.configure.literal_binds`, which
+ will pass the ``literal_binds`` flag into the compilation of SQL
+ constructs when using "offline" mode. This has the effect that
+ SQL objects like inserts, updates, deletes as well as textual
+ statements sent using ``text()`` will be compiled such that the dialect
+ will attempt to render literal values "inline" automatically.
+ Only a subset of types is typically supported; the
+ :meth:`.Operations.inline_literal` construct remains as the construct
+ used to force a specific literal representation of a value.
+ The :paramref:`.EnvironmentContext.configure.literal_binds` flag
+ is added to the "offline" section of the ``env.py`` files generated
+ in new environments.
+
.. change::
:tags: bug, batch
:tickets: 289
# doesn't have an IDENTITY column
context.assert_(
'SET IDENTITY_INSERT ins_table ON',
+ 'GO',
"INSERT INTO ins_table (id, v1, v2) "
"VALUES (1, 'row v1', 'row v5')",
+ 'GO',
"INSERT INTO ins_table (id, v1, v2) "
"VALUES (2, 'row v2', 'row v6')",
+ 'GO',
"INSERT INTO ins_table (id, v1, v2) "
"VALUES (3, 'row v3', 'row v7')",
+ 'GO',
"INSERT INTO ins_table (id, v1, v2) "
"VALUES (4, 'row v4', 'row v8')",
- 'SET IDENTITY_INSERT ins_table OFF'
+ 'GO',
+ 'SET IDENTITY_INSERT ins_table OFF',
+ 'GO',
)
def test_bulk_insert_from_new_table(self):
context = op_fixture('mysql')
op.drop_constraint('primary', 't1', type_='primary')
context.assert_(
- "ALTER TABLE t1 DROP PRIMARY KEY "
+ "ALTER TABLE t1 DROP PRIMARY KEY"
)
def test_drop_unique(self):
context = op_fixture('mssql')
op.drop_index('ik_test', tablename='t1')
context.assert_("DROP INDEX ik_test ON 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
+ from sqlalchemy import String, Integer
+
+ account = table('account',
+ column('name', String),
+ column('id', Integer)
+ )
+ op.execute(
+ account.update().
+ where(account.c.name == op.inline_literal('account 1')).
+ values({'name': op.inline_literal('account 2')})
+ )
+ op.execute(text("update table set foo=:bar").bindparams(bar='bat'))
+ context.assert_(
+ "UPDATE account SET name='account 2' "
+ "WHERE account.name = 'account 1'",
+ "update table set foo='bat'"
+ )
+
+ def test_create_table_literal_binds(self):
+ context = op_fixture(as_sql=True, literal_binds=True)
+
+ op.create_table(
+ "some_table",
+ Column('id', Integer, primary_key=True),
+ Column('st_id', Integer, ForeignKey('some_table.id'))
+ )
+
+ context.assert_(
+ "CREATE TABLE some_table (id INTEGER NOT NULL, st_id INTEGER, "
+ "PRIMARY KEY (id), FOREIGN KEY(st_id) REFERENCES some_table (id))"
+ )
\ No newline at end of file
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
