From: Mike Bayer Date: Thu, 4 Apr 2013 20:23:24 +0000 (-0400) Subject: Added ``output_encoding`` option to X-Git-Tag: rel_0_5_0~7 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=0b316ec069a5d8fa721613373a0e5581990bb460;p=thirdparty%2Fsqlalchemy%2Falembic.git Added ``output_encoding`` option to :meth:`.EnvironmentContext.configure`, used with ``--sql`` mode to apply an encoding to the output stream. #90 --- diff --git a/alembic/environment.py b/alembic/environment.py index 9d57a857..d6b47dfc 100644 --- a/alembic/environment.py +++ b/alembic/environment.py @@ -204,6 +204,7 @@ class EnvironmentContext(object): dialect_name=None, transactional_ddl=None, output_buffer=None, + output_encoding=None, starting_rev=None, tag=None, template_args=None, @@ -273,6 +274,11 @@ class EnvironmentContext(object): the :class:`.Config` object. The value here overrides that of the :class:`.Config` object. + :param output_encoding: when using ``--sql`` to generate SQL + scripts, apply this encoding to the string output. + + .. versionadded:: 0.5.0 + :param starting_rev: Override the "starting revision" argument when using ``--sql`` mode. :param tag: a string tag for usage by custom ``env.py`` scripts. @@ -471,6 +477,8 @@ class EnvironmentContext(object): opts["output_buffer"] = output_buffer elif self.config.output_buffer is not None: opts["output_buffer"] = self.config.output_buffer + if output_encoding is not None: + opts["output_encoding"] = output_encoding if starting_rev: opts['starting_rev'] = starting_rev if tag: diff --git a/alembic/migration.py b/alembic/migration.py index fd75c1db..850bbd14 100644 --- a/alembic/migration.py +++ b/alembic/migration.py @@ -4,6 +4,7 @@ from sqlalchemy import create_engine from alembic import ddl import sys from sqlalchemy.engine import url as sqla_url +import codecs import logging log = logging.getLogger(__name__) @@ -69,6 +70,10 @@ class MigrationContext(object): self._migrations_fn = opts.get('fn') self.as_sql = as_sql self.output_buffer = opts.get("output_buffer", sys.stdout) + if opts.get('output_encoding'): + self.output_buffer = codecs.getwriter( + opts['output_encoding'] + )(self.output_buffer) self._user_compare_type = opts.get('compare_type', False) self._user_compare_server_default = opts.get( diff --git a/docs/build/changelog.rst b/docs/build/changelog.rst index f60ac595..e758074b 100644 --- a/docs/build/changelog.rst +++ b/docs/build/changelog.rst @@ -6,6 +6,15 @@ Changelog .. changelog:: :version: 0.5.0 + .. change:: + :tags: feature + :tickets: 90 + + Added ``output_encoding`` option to + :meth:`.EnvironmentContext.configure`, + used with ``--sql`` mode to apply an encoding + to the output stream. + .. change:: :tags: feature :tickets: 93 diff --git a/tests/__init__.py b/tests/__init__.py index 268ac610..4d9a5b94 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -106,7 +106,7 @@ def capture_context_buffer(**kw): return buf def __exit__(self, *arg, **kwarg): - print buf.getvalue() + #print(buf.getvalue()) EnvironmentContext._default_opts = None return capture() diff --git a/tests/test_sql_script.py b/tests/test_sql_script.py index fd1fc7d8..3dcb7b2b 100644 --- a/tests/test_sql_script.py +++ b/tests/test_sql_script.py @@ -1,76 +1,115 @@ -from __future__ import with_statement +# coding: utf-8 + +from __future__ import with_statement, unicode_literals from tests import clear_staging_env, staging_env, \ _no_sql_testing_config, capture_context_buffer, \ - three_rev_fixture -from alembic import command + three_rev_fixture, write_script +from alembic import command, util +from alembic.script import ScriptDirectory +import unittest cfg = None a, b, c = None, None, None -def setup(): - global cfg, env - env = staging_env() - cfg = _no_sql_testing_config() - cfg.set_main_option('dialect_name', 'sqlite') - cfg.remove_main_option('url') - global a, b, c - a, b, c = three_rev_fixture(cfg) - -def teardown(): - clear_staging_env() - -def test_begin_comit(): - with capture_context_buffer(transactional_ddl=True) as buf: - command.upgrade(cfg, a, sql=True) - assert "BEGIN;" in buf.getvalue() - assert "COMMIT;" in buf.getvalue() - - with capture_context_buffer(transactional_ddl=False) as buf: - command.upgrade(cfg, a, sql=True) - assert "BEGIN;" not in buf.getvalue() - assert "COMMIT;" not in buf.getvalue() - -def test_version_from_none_insert(): - with capture_context_buffer() as buf: - command.upgrade(cfg, a, sql=True) - assert "CREATE TABLE alembic_version" in buf.getvalue() - assert "INSERT INTO alembic_version" in buf.getvalue() - assert "CREATE STEP 1" in buf.getvalue() - assert "CREATE STEP 2" not in buf.getvalue() - assert "CREATE STEP 3" not in buf.getvalue() - -def test_version_from_middle_update(): - with capture_context_buffer() as buf: - command.upgrade(cfg, "%s:%s" % (b, c), sql=True) - assert "CREATE TABLE alembic_version" not in buf.getvalue() - assert "UPDATE alembic_version" in buf.getvalue() - assert "CREATE STEP 1" not in buf.getvalue() - assert "CREATE STEP 2" not in buf.getvalue() - assert "CREATE STEP 3" in buf.getvalue() - -def test_version_to_none(): - with capture_context_buffer() as buf: - command.downgrade(cfg, "%s:base" % c, sql=True) - assert "CREATE TABLE alembic_version" not in buf.getvalue() - assert "INSERT INTO alembic_version" not in buf.getvalue() - assert "DROP TABLE alembic_version" in buf.getvalue() - assert "DROP STEP 3" in buf.getvalue() - assert "DROP STEP 2" in buf.getvalue() - assert "DROP STEP 1" in buf.getvalue() - -def test_version_to_middle(): - with capture_context_buffer() as buf: - command.downgrade(cfg, "%s:%s" % (c, a), sql=True) - assert "CREATE TABLE alembic_version" not in buf.getvalue() - assert "INSERT INTO alembic_version" not in buf.getvalue() - assert "DROP TABLE alembic_version" not in buf.getvalue() - assert "DROP STEP 3" in buf.getvalue() - assert "DROP STEP 2" in buf.getvalue() - assert "DROP STEP 1" not in buf.getvalue() - -def test_stamp(): - with capture_context_buffer() as buf: - command.stamp(cfg, "head", sql=True) - assert "UPDATE alembic_version SET version_num='%s';" % c in buf.getvalue() +class ThreeRevTest(unittest.TestCase): + + def setUp(self): + global cfg, env + env = staging_env() + cfg = _no_sql_testing_config() + cfg.set_main_option('dialect_name', 'sqlite') + cfg.remove_main_option('url') + global a, b, c + a, b, c = three_rev_fixture(cfg) + + def tearDown(self): + clear_staging_env() + + def test_begin_comit(self): + with capture_context_buffer(transactional_ddl=True) as buf: + command.upgrade(cfg, a, sql=True) + assert "BEGIN;" in buf.getvalue() + assert "COMMIT;" in buf.getvalue() + + with capture_context_buffer(transactional_ddl=False) as buf: + command.upgrade(cfg, a, sql=True) + assert "BEGIN;" not in buf.getvalue() + assert "COMMIT;" not in buf.getvalue() + + def test_version_from_none_insert(self): + with capture_context_buffer() as buf: + command.upgrade(cfg, a, sql=True) + assert "CREATE TABLE alembic_version" in buf.getvalue() + assert "INSERT INTO alembic_version" in buf.getvalue() + assert "CREATE STEP 1" in buf.getvalue() + assert "CREATE STEP 2" not in buf.getvalue() + assert "CREATE STEP 3" not in buf.getvalue() + + def test_version_from_middle_update(self): + with capture_context_buffer() as buf: + command.upgrade(cfg, "%s:%s" % (b, c), sql=True) + assert "CREATE TABLE alembic_version" not in buf.getvalue() + assert "UPDATE alembic_version" in buf.getvalue() + assert "CREATE STEP 1" not in buf.getvalue() + assert "CREATE STEP 2" not in buf.getvalue() + assert "CREATE STEP 3" in buf.getvalue() + + def test_version_to_none(self): + with capture_context_buffer() as buf: + command.downgrade(cfg, "%s:base" % c, sql=True) + assert "CREATE TABLE alembic_version" not in buf.getvalue() + assert "INSERT INTO alembic_version" not in buf.getvalue() + assert "DROP TABLE alembic_version" in buf.getvalue() + assert "DROP STEP 3" in buf.getvalue() + assert "DROP STEP 2" in buf.getvalue() + assert "DROP STEP 1" in buf.getvalue() + + def test_version_to_middle(self): + with capture_context_buffer() as buf: + command.downgrade(cfg, "%s:%s" % (c, a), sql=True) + assert "CREATE TABLE alembic_version" not in buf.getvalue() + assert "INSERT INTO alembic_version" not in buf.getvalue() + assert "DROP TABLE alembic_version" not in buf.getvalue() + assert "DROP STEP 3" in buf.getvalue() + assert "DROP STEP 2" in buf.getvalue() + assert "DROP STEP 1" not in buf.getvalue() + + def test_stamp(self): + with capture_context_buffer() as buf: + command.stamp(cfg, "head", sql=True) + assert "UPDATE alembic_version SET version_num='%s';" % c in buf.getvalue() + + +class EncodingTest(unittest.TestCase): + def setUp(self): + global cfg, env, a + env = staging_env() + cfg = _no_sql_testing_config() + cfg.set_main_option('dialect_name', 'sqlite') + cfg.remove_main_option('url') + a = util.rev_id() + script = ScriptDirectory.from_config(cfg) + script.generate_revision(a, "revision a", refresh=True) + write_script(script, a, """# coding: utf-8 +from __future__ import unicode_literals +revision = '%s' +down_revision = None + +from alembic import op + +def upgrade(): + op.execute("« S’il vous plaît…") + +def downgrade(): + op.execute("drôle de petite voix m’a réveillé") + +""".encode('utf-8') % a) + + def tearDown(self): + clear_staging_env() + def test_encode(self): + with capture_context_buffer(output_encoding='utf-8') as buf: + command.upgrade(cfg, a, sql=True) + assert "« S’il vous plaît…".encode("utf-8") in buf.getvalue()