# See https://pre-commit.com/hooks.html for more hooks
repos:
- repo: https://github.com/python/black
- rev: 25.9.0
+ rev: 26.5.1
hooks:
- id: black
from . import op
from .runtime import plugins
-
__version__ = "1.18.6"
opts: List[Tuple[str, Any]] = []
if column.server_default:
- rendered = _render_server_default( # type:ignore[assignment]
+ rendered = _render_server_default( # type: ignore[assignment]
column.server_default, autogen_context
)
if rendered:
head=revisions,
branch_labels=branch_label,
splice=splice,
- **template_args, # type:ignore[arg-type]
+ **template_args, # type: ignore[arg-type]
)
from .util import compat
from .util.pyfiles import _preserving_path_as_str
-
log = logging.getLogger(__name__)
return None
return Path(self.toml_file_name)
- config_ini_section: str = None # type:ignore[assignment]
+ config_ini_section: str = None # type: ignore[assignment]
"""Name of the config file section to read basic configuration
from. Defaults to ``alembic``, that is the ``[alembic]`` section
of the .ini file. This value is modified using the ``-n/--name``
if arg in self._KWARGS_OPTS:
kwarg_opt = self._KWARGS_OPTS[arg]
args, opts = kwarg_opt[0:-1], kwarg_opt[-1]
- subparser.add_argument(*args, **opts) # type:ignore
+ subparser.add_argument(*args, **opts) # type: ignore
for arg in positional:
opts = self._POSITIONAL_OPTS.get(arg, {})
- subparser.add_argument(arg, **opts) # type:ignore
+ subparser.add_argument(arg, **opts) # type: ignore
def _inspect_function(self, fn: CommandFunction) -> tuple[Any, Any, str]:
spec = compat.inspect_getfullargspec(fn)
cls_(
table_name,
column_name,
- server_default, # type:ignore[arg-type]
+ server_default, # type: ignore[arg-type]
schema=schema,
existing_type=existing_type,
existing_server_default=existing_server_default,
index.table.append_column(Column(col, sqltypes.NullType))
self._exec(CreateIndex(index, **kw))
- def bulk_insert( # type:ignore[override]
+ def bulk_insert( # type: ignore[override]
self, table: Union[TableClause, Table], rows: List[dict], **kw: Any
) -> None:
if self.as_sql:
from ..util import sqla_compat
from ..util.sqla_compat import compiles
-
if TYPE_CHECKING:
from typing import Literal
self.kw = kw
@classmethod
- def from_constraint( # type:ignore[override]
+ def from_constraint( # type: ignore[override]
cls, constraint: ExcludeConstraint
) -> CreateExcludeConstraintOp:
constraint_table = sqla_compat._table_for_constraint(constraint)
"(%s, %r)"
% (
_render_potential_column(
- sqltext, # type:ignore[arg-type]
+ sqltext, # type: ignore[arg-type]
autogen_context,
),
opstring,
from .ops import MigrateOperation
from .ops import MigrationScript
-
__all__ = [
"AbstractOperations",
"Operations",
a particular constraint name is already converted.
"""
- return conv(name)
+ return conv(name) # type: ignore[no-any-return]
def inline_literal(
self, value: Union[str, int], type_: Optional[TypeEngine[Any]] = None
# pop named constraints for Boolean/Enum for rename
if (
isinstance(resolved_existing_type, SchemaEventTarget)
- and resolved_existing_type.name # type:ignore[attr-defined] # noqa E501
+ and resolved_existing_type.name # type: ignore[attr-defined] # noqa E501
):
self.named_constraints.pop(
- resolved_existing_type.name, # type:ignore[attr-defined] # noqa E501
+ resolved_existing_type.name, # type: ignore[attr-defined] # noqa E501
None,
)
# Operations.implementation_for(alter_column)
if isinstance(existing.type, SchemaEventTarget):
- existing.type._create_events = ( # type:ignore[attr-defined]
- existing.type.create_constraint # type:ignore[attr-defined] # noqa
+ existing.type._create_events = ( # type: ignore[attr-defined]
+ existing.type.create_constraint # type: ignore[attr-defined] # noqa
) = False
self.impl.cast_for_batch_migrate(
if (
"existing_type" in kw
and isinstance(kw["existing_type"], SchemaEventTarget)
- and kw["existing_type"].name # type:ignore[attr-defined]
+ and kw["existing_type"].name # type: ignore[attr-defined]
):
self.named_constraints.pop(
- kw["existing_type"].name, None # type:ignore[attr-defined]
+ kw["existing_type"].name, None # type: ignore[attr-defined]
)
def create_column_comment(self, column):
_migration_context: Optional[MigrationContext] = None
- config: Config = None # type:ignore[assignment]
+ config: Config = None # type: ignore[assignment]
"""An instance of :class:`.Config` representing the
configuration file contents as well as other variables
set programmatically within it."""
- script: ScriptDirectory = None # type:ignore[assignment]
+ script: ScriptDirectory = None # type: ignore[assignment]
"""An instance of :class:`.ScriptDirectory` which provides
programmatic access to version files within the ``versions/``
directory.
if "output_encoding" in opts:
self.output_buffer = EncodedIO(
opts.get("output_buffer")
- or sys.stdout, # type:ignore[arg-type]
+ or sys.stdout, # type: ignore[arg-type]
opts["output_encoding"],
)
else:
self.output_buffer = opts.get(
"output_buffer", sys.stdout
- ) # type:ignore[assignment] # noqa: E501
+ ) # type: ignore[assignment] # noqa: E501
self.transactional_ddl = transactional_ddl
def _in_connection_transaction(self) -> bool:
try:
- meth = self.connection.in_transaction # type:ignore[union-attr]
+ meth = self.connection.in_transaction # type: ignore[union-attr]
except AttributeError:
return False
else:
return (self.revision.revision,)
@property
- def from_revisions_no_deps( # type:ignore[override]
+ def from_revisions_no_deps( # type: ignore[override]
self,
) -> Tuple[str, ...]:
if self.is_upgrade:
return self.revision._normalized_down_revisions
@property
- def to_revisions_no_deps( # type:ignore[override]
+ def to_revisions_no_deps( # type: ignore[override]
self,
) -> Tuple[str, ...]:
if self.is_upgrade:
return self.to_
@property
- def from_revisions_no_deps( # type:ignore[override]
+ def from_revisions_no_deps( # type: ignore[override]
self,
) -> Tuple[str, ...]:
return self.from_
@property
- def to_revisions_no_deps( # type:ignore[override]
+ def to_revisions_no_deps( # type: ignore[override]
self,
) -> Tuple[str, ...]:
return self.to_
def env_file_fixture(txt):
dir_ = _join_path(_get_staging_directory(), "scripts")
- txt = (
- """
+ txt = """
from alembic import context
config = context.config
-"""
- + txt
- )
+""" + txt
path = _join_path(dir_, "env.py")
pyc_path = util.pyc_file_from_path(path)
sqlalchemy_future = future or ("future" in config.db.__class__.__module__)
- return _write_config_file(
- f"""
+ return _write_config_file(f"""
[alembic]
script_location = {dir_}
sqlalchemy.url = {url}
[formatter_generic]
format = %%(levelname)-5.5s [%%(name)s] %%(message)s
datefmt = %%H:%%M:%%S
- """
- )
+ """)
def _multi_dir_testing_config(sourceless=False, extra_version_location=""):
url = "sqlite:///%s/foo.db" % dir_
- return _write_config_file(
- f"""
+ return _write_config_file(f"""
[alembic]
script_location = {dir_}
sqlalchemy.url = {url}
[formatter_generic]
format = %%(levelname)-5.5s [%%(name)s] %%(message)s
datefmt = %%H:%%M:%%S
- """
- )
+ """)
def _no_sql_pyproject_config(dialect="postgresql", directives=""):
"""use a postgresql url with no host so that
connections guaranteed to fail"""
dir_ = _join_path(_get_staging_directory(), "scripts")
- return _write_config_file(
- f"""
+ return _write_config_file(f"""
[alembic]
script_location ={dir_}
sqlalchemy.url = {dialect}://
format = %%(levelname)-5.5s [%%(name)s] %%(message)s
datefmt = %%H:%%M:%%S
-"""
- )
+""")
def _write_toml_config(tomltext, initext):
for key, value in engines.items()
)
- return _write_config_file(
- f"""
+ return _write_config_file(f"""
[alembic]
script_location = {dir_}
sourceless = false
[formatter_generic]
format = %%(levelname)-5.5s [%%(name)s] %%(message)s
datefmt = %%H:%%M:%%S
- """
- )
+ """)
def _join_path(base: str, *more: str):
import os
import sys
-
# If extensions (or modules to document with autodoc) are in another directory,
# add these directories to sys.path here. If the directory is relative to the
# documentation root, use os.path.abspath to make it absolute, like shown here.
tests = [
"pytest>8,<10",
"pytest-xdist",
- "black==25.9.0", # for test_post_write.py
+ "black==26.5.1", # for test_post_write.py
"zimports", # for stub testing
"tzdata",
"junitparser"
"flake8-rst-docstrings",
"pydocstyle",
"pygments",
- "black==25.9.0"
+ "black==26.5.1"
]
mypy = [
from sqlalchemy.testing import provision
-
logging.basicConfig()
logging.getLogger(provision.__name__).setLevel(logging.INFO)
installs SQLAlchemy's testing plugin into the local environment.
"""
+
import os
import pytest
batch_op.drop_column('foo')
batch_op.add_column(Column('bar', String))
- """ # noqa: E501
- % a,
+ """ % a, # noqa: E501
)
yield go
batch_op.drop_column('foo')
batch_op.add_column(Column('bar', String))
- """
- % a,
+ """ % a,
)
yield go
def setUp(self):
self.conn = config.db.connect()
with self.conn.begin():
- self.conn.execute(
- text(
- """
+ self.conn.execute(text("""
create table foo(
id integer primary key,
data varchar(50),
x integer
)
- """
- )
- )
+ """))
context = MigrationContext.configure(self.conn)
self.op = op.Operations(context)
self.t1 = table("foo", column("id"), column("data"), column("x"))
@classmethod
def _setup_env_file(self):
- env_file_fixture(
- r"""
+ env_file_fixture(r"""
from sqlalchemy import MetaData, engine_from_config
target_metadata = MetaData()
finally:
connection.close()
engine.dispose()
-"""
- )
+""")
def _eq_cmd_output(self, buf, expected, env_token=False, currents=()):
script = ScriptDirectory.from_config(self.cfg)
@classmethod
def _setup_env_file(self):
- env_file_fixture(
- r"""
+ env_file_fixture(r"""
from sqlalchemy import MetaData, engine_from_config
target_metadata = MetaData()
finally:
connection.close()
engine.dispose()
-"""
- )
+""")
def _assert_env_token(self, buf, expected):
if expected:
clear_staging_env()
def _env_fixture(self, version_table_pk=True):
- env_file_fixture(
- """
+ env_file_fixture("""
from sqlalchemy import MetaData, engine_from_config
target_metadata = MetaData()
connection.close()
engine.dispose()
-"""
- % (version_table_pk,)
- )
+""" % (version_table_pk,))
def test_create_rev_plain_db_not_up_to_date(self):
self._env_fixture()
clear_staging_env()
def _env_fixture(self, version_table_pk=True):
- env_file_fixture(
- f"""
+ env_file_fixture(f"""
from sqlalchemy import MetaData, engine_from_config
target_metadata = MetaData()
connection.close()
engine.dispose()
-"""
- )
+""")
def test_check_no_changes(self):
self._env_fixture()
)
def _env_fixture(self):
- env_file_fixture(
- """
+ env_file_fixture("""
import re
from sqlalchemy import MetaData, engine_from_config
engine.dispose()
-"""
- )
+""")
class _StampTest:
"""
revision = '%s'
down_revision = None
-"""
- % a,
+""" % a,
)
script.generate_revision(b, None, refresh=True)
write_script(
"""
revision = '%s'
down_revision = '%s'
-"""
- % (b, a),
+""" % (b, a),
)
def tearDown(self):
root = pathlib.Path(_get_staging_directory())
with (root / "pyproject.toml").open("w") as file_:
- file_.write(
- """[tool.sometool]
-someconfig = 'bar'"""
- )
+ file_.write("""[tool.sometool]
+someconfig = 'bar'""")
yield config.Config(
self.cfg.config_file_name, toml_file=root / "pyproject.toml"
)
with open(cfg.toml_file_name, "r") as f:
file_content = f.read()
- assert file_content.startswith(
- """[tool.sometool]
-someconfig = 'bar'\n\n[tool.alembic]"""
- )
+ assert file_content.startswith("""[tool.sometool]
+someconfig = 'bar'\n\n[tool.alembic]""")
toml = compat.tomllib.loads(file_content)
eq_(
toml,
class FileConfigTest(TestBase):
def test_config_args(self):
- cfg = _write_config_file(
- """
+ cfg = _write_config_file("""
[alembic]
migrations = %(base_path)s/db/migrations
-"""
- )
+""")
test_cfg = config.Config(
cfg.config_file_name, config_args=dict(base_path="/home/alembic")
)
logger.addHandler(handler)
logger.setLevel(logging.INFO)
- cfg = _write_config_file(
- """
+ cfg = _write_config_file("""
[alembic]
script_location = %(base_path)s/db/migrations
-"""
- )
+""")
test_cfg = config.Config(
cfg.config_file_name, config_args=dict(base_path="/tmp")
)
def test_script_location(self, pyproject_only_env):
cfg = pyproject_only_env
with cfg._toml_file_path.open("wb") as file_:
- file_.write(
- rb"""
+ file_.write(rb"""
[tool.alembic]
script_location = "%(here)s/scripts"
-"""
- )
+""")
new_cfg = config.Config(
file_=cfg.config_file_name, toml_file=cfg._toml_file_path
cfg = pyproject_only_env
with cfg._toml_file_path.open("ba") as file_:
- file_.write(
- b"""
+ file_.write(b"""
version_locations = [
"%(here)s/foo/bar"
]
-"""
- )
+""")
if "toml_alembic_config" in cfg.__dict__:
cfg.__dict__.pop("toml_alembic_config")
cfg = pyproject_only_env
with cfg._toml_file_path.open("wb") as file_:
- file_.write(
- rb"""
+ file_.write(rb"""
[tool.alembic]
script_location = "%(here)s/scripts"
"%(here)s/path/to/python",
"c:\\some\\path"
]
-"""
- )
+""")
if "toml_alembic_config" in cfg.__dict__:
cfg.__dict__.pop("toml_alembic_config")
cfg = pyproject_only_env
with cfg._toml_file_path.open("wb") as file_:
- file_.write(
- rb"""
+ file_.write(rb"""
[tool.alembic]
script_location = "%(here)s/scripts"
executable = "%(here)s/.venv/bin/ruff"
options = "-l 79 REVISION_SCRIPT_FILENAME"
-"""
- )
+""")
if "toml_alembic_config" in cfg.__dict__:
cfg.__dict__.pop("toml_alembic_config")
cfg = pyproject_only_env
with cfg._toml_file_path.open("wb") as file_:
- file_.write(
- rb"""
+ file_.write(rb"""
[tool.alembic]
script_location = "%(here)s/scripts"
"two %(here)s three"
]
-"""
- )
+""")
if "toml_alembic_config" in cfg.__dict__:
cfg.__dict__.pop("toml_alembic_config")
else:
argtype.fail()
- file_.write(
- rf"""
+ file_.write(rf"""
[tool.alembic]
script_location = "%(here)s/scripts"
{config_option}
-"""
- )
+""")
if "toml_alembic_config" in cfg.__dict__:
cfg.__dict__.pop("toml_alembic_config")
else:
arg_type.fail()
- file_.write(
- rf"""
+ file_.write(rf"""
[tool.alembic]
script_location = "%(here)s/scripts"
{config_option}
-"""
- )
+""")
if "toml_alembic_config" in cfg.__dict__:
cfg.__dict__.pop("toml_alembic_config")
do some SQL thing with a % percent sign %
''')
-""".format(
- a_rev
- ),
+""".format(a_rev),
)
with capture_context_buffer(transactional_ddl=True) as buf:
command.upgrade(self.cfg, "arev", sql=True)
def downgrade():
pass
-"""
- % a_rev,
+""" % a_rev,
)
migration_fn = mock.MagicMock()
clear_staging_env()
def test_not_requires_connection(self):
- env_file_fixture(
- """
+ env_file_fixture("""
assert not context.requires_connection()
-"""
- )
+""")
command.upgrade(self.cfg, a, sql=True)
command.downgrade(self.cfg, "%s:%s" % (b, a), sql=True)
def test_requires_connection(self):
- env_file_fixture(
- """
+ env_file_fixture("""
assert context.requires_connection()
-"""
- )
+""")
command.upgrade(self.cfg, a)
command.downgrade(self.cfg, a)
def test_starting_rev_post_context(self):
- env_file_fixture(
- """
+ env_file_fixture("""
context.configure(dialect_name='sqlite', starting_rev='x')
assert context.get_starting_revision_argument() == 'x'
-"""
- )
+""")
command.upgrade(self.cfg, a, sql=True)
command.downgrade(self.cfg, "%s:%s" % (b, a), sql=True)
command.current(self.cfg)
command.stamp(self.cfg, a)
def test_starting_rev_pre_context(self):
- env_file_fixture(
- """
+ env_file_fixture("""
assert context.get_starting_revision_argument() == 'x'
-"""
- )
+""")
command.upgrade(self.cfg, "x:y", sql=True)
command.downgrade(self.cfg, "x:y", sql=True)
def test_starting_rev_pre_context_cmd_w_no_startrev(self):
- env_file_fixture(
- """
+ env_file_fixture("""
assert context.get_starting_revision_argument() == 'x'
-"""
- )
+""")
assert_raises_message(
util.CommandError,
"No starting revision argument is available.",
)
def test_starting_rev_current_pre_context(self):
- env_file_fixture(
- """
+ env_file_fixture("""
assert context.get_starting_revision_argument() is None
-"""
- )
+""")
assert_raises_message(
util.CommandError,
"No starting revision argument is available.",
)
def test_destination_rev_pre_context(self):
- env_file_fixture(
- """
+ env_file_fixture("""
assert context.get_revision_argument() == '%s'
-"""
- % b
- )
+""" % b)
command.upgrade(self.cfg, b, sql=True)
command.stamp(self.cfg, b, sql=True)
command.downgrade(self.cfg, "%s:%s" % (c, b), sql=True)
def test_destination_rev_pre_context_multihead(self):
d, e, f = multi_heads_fixture(self.cfg, a, b, c)
- env_file_fixture(
- """
+ env_file_fixture("""
assert set(context.get_revision_argument()) == set(('%s', '%s', '%s', ))
-"""
- % (f, e, c)
- )
+""" % (f, e, c))
command.upgrade(self.cfg, "heads", sql=True)
def test_destination_rev_post_context(self):
- env_file_fixture(
- """
+ env_file_fixture("""
context.configure(dialect_name='sqlite')
assert context.get_revision_argument() == '%s'
-"""
- % b
- )
+""" % b)
command.upgrade(self.cfg, b, sql=True)
command.downgrade(self.cfg, "%s:%s" % (c, b), sql=True)
command.stamp(self.cfg, b, sql=True)
def test_destination_rev_post_context_multihead(self):
d, e, f = multi_heads_fixture(self.cfg, a, b, c)
- env_file_fixture(
- """
+ env_file_fixture("""
context.configure(dialect_name='sqlite')
assert set(context.get_revision_argument()) == set(('%s', '%s', '%s', ))
-"""
- % (f, e, c)
- )
+""" % (f, e, c))
command.upgrade(self.cfg, "heads", sql=True)
def test_head_rev_pre_context(self):
- env_file_fixture(
- """
+ env_file_fixture("""
assert context.get_head_revision() == '%s'
assert context.get_head_revisions() == ('%s', )
-"""
- % (c, c)
- )
+""" % (c, c))
command.upgrade(self.cfg, b, sql=True)
command.downgrade(self.cfg, "%s:%s" % (b, a), sql=True)
command.stamp(self.cfg, b, sql=True)
def test_head_rev_pre_context_multihead(self):
d, e, f = multi_heads_fixture(self.cfg, a, b, c)
- env_file_fixture(
- """
+ env_file_fixture("""
assert set(context.get_head_revisions()) == set(('%s', '%s', '%s', ))
-"""
- % (e, f, c)
- )
+""" % (e, f, c))
command.upgrade(self.cfg, e, sql=True)
command.downgrade(self.cfg, "%s:%s" % (e, b), sql=True)
command.stamp(self.cfg, c, sql=True)
command.current(self.cfg)
def test_head_rev_post_context(self):
- env_file_fixture(
- """
+ env_file_fixture("""
context.configure(dialect_name='sqlite')
assert context.get_head_revision() == '%s'
assert context.get_head_revisions() == ('%s', )
-"""
- % (c, c)
- )
+""" % (c, c))
command.upgrade(self.cfg, b, sql=True)
command.downgrade(self.cfg, "%s:%s" % (b, a), sql=True)
command.stamp(self.cfg, b, sql=True)
def test_head_rev_post_context_multihead(self):
d, e, f = multi_heads_fixture(self.cfg, a, b, c)
- env_file_fixture(
- """
+ env_file_fixture("""
context.configure(dialect_name='sqlite')
assert set(context.get_head_revisions()) == set(('%s', '%s', '%s', ))
-"""
- % (e, f, c)
- )
+""" % (e, f, c))
command.upgrade(self.cfg, e, sql=True)
command.downgrade(self.cfg, "%s:%s" % (e, b), sql=True)
command.stamp(self.cfg, c, sql=True)
command.current(self.cfg)
def test_tag_pre_context(self):
- env_file_fixture(
- """
+ env_file_fixture("""
assert context.get_tag_argument() == 'hi'
-"""
- )
+""")
command.upgrade(self.cfg, b, sql=True, tag="hi")
command.downgrade(self.cfg, "%s:%s" % (b, a), sql=True, tag="hi")
def test_tag_pre_context_None(self):
- env_file_fixture(
- """
+ env_file_fixture("""
assert context.get_tag_argument() is None
-"""
- )
+""")
command.upgrade(self.cfg, b, sql=True)
command.downgrade(self.cfg, "%s:%s" % (b, a), sql=True)
def test_tag_cmd_arg(self):
- env_file_fixture(
- """
+ env_file_fixture("""
context.configure(dialect_name='sqlite')
assert context.get_tag_argument() == 'hi'
-"""
- )
+""")
command.upgrade(self.cfg, b, sql=True, tag="hi")
command.downgrade(self.cfg, "%s:%s" % (b, a), sql=True, tag="hi")
def test_tag_cfg_arg(self):
- env_file_fixture(
- """
+ env_file_fixture("""
context.configure(dialect_name='sqlite', tag='there')
assert context.get_tag_argument() == 'there'
-"""
- )
+""")
command.upgrade(self.cfg, b, sql=True, tag="hi")
command.downgrade(self.cfg, "%s:%s" % (b, a), sql=True, tag="hi")
def test_tag_None(self):
- env_file_fixture(
- """
+ env_file_fixture("""
context.configure(dialect_name='sqlite')
assert context.get_tag_argument() is None
-"""
- )
+""")
command.upgrade(self.cfg, b, sql=True)
command.downgrade(self.cfg, "%s:%s" % (b, a), sql=True)
def test_downgrade_wo_colon(self):
- env_file_fixture(
- """
+ env_file_fixture("""
context.configure(dialect_name='sqlite')
-"""
- )
+""")
assert_raises_message(
util.CommandError,
"downgrade with --sql requires <fromrev>:<torev>",
)
def test_upgrade_with_output_encoding(self):
- env_file_fixture(
- """
+ env_file_fixture("""
url = config.get_main_option('sqlalchemy.url')
context.configure(url=url, output_encoding='utf-8')
assert not context.requires_connection()
-"""
- )
+""")
command.upgrade(self.cfg, a, sql=True)
command.downgrade(self.cfg, "%s:%s" % (b, a), sql=True)
)
def test_starting_rev_pre_context_abbreviated(self):
- env_file_fixture(
- """
+ env_file_fixture("""
assert context.get_starting_revision_argument() == '%s'
-"""
- % b[0:4]
- )
+""" % b[0:4])
command.upgrade(self.cfg, "%s:%s" % (b[0:4], c), sql=True)
command.stamp(self.cfg, "%s:%s" % (b[0:4], c), sql=True)
command.downgrade(self.cfg, "%s:%s" % (b[0:4], a), sql=True)
def test_destination_rev_pre_context_abbreviated(self):
- env_file_fixture(
- """
+ env_file_fixture("""
assert context.get_revision_argument() == '%s'
-"""
- % b[0:4]
- )
+""" % b[0:4])
command.upgrade(self.cfg, "%s:%s" % (a, b[0:4]), sql=True)
command.stamp(self.cfg, b[0:4], sql=True)
command.downgrade(self.cfg, "%s:%s" % (c, b[0:4]), sql=True)
def test_starting_rev_context_runs_abbreviated(self):
- env_file_fixture(
- """
+ env_file_fixture("""
context.configure(dialect_name='sqlite')
context.run_migrations()
-"""
- )
+""")
command.upgrade(self.cfg, "%s:%s" % (b[0:4], c), sql=True)
command.downgrade(self.cfg, "%s:%s" % (b[0:4], a), sql=True)
def test_destination_rev_context_runs_abbreviated(self):
- env_file_fixture(
- """
+ env_file_fixture("""
context.configure(dialect_name='sqlite')
context.run_migrations()
-"""
- )
+""")
command.upgrade(self.cfg, "%s:%s" % (a, b[0:4]), sql=True)
command.stamp(self.cfg, b[0:4], sql=True)
command.downgrade(self.cfg, "%s:%s" % (c, b[0:4]), sql=True)
)
)
else:
- self.cfg = _no_sql_pyproject_config(
- directives="""
+ self.cfg = _no_sql_pyproject_config(directives="""
[[tool.alembic.post_write_hooks]]
name="hook1"
name="hook2"
type="hook2"
arg1="bar"
- """
- )
+ """)
rev = command.revision(self.cfg, message="x")
eq_(
if config.ini:
self.cfg = _no_sql_testing_config(directives="")
else:
- self.cfg = _no_sql_pyproject_config(
- directives="""
+ self.cfg = _no_sql_pyproject_config(directives="""
- """
- )
+ """)
command.revision(self.cfg, message="x")
)
)
else:
- self.cfg = _no_sql_pyproject_config(
- directives="""
+ self.cfg = _no_sql_pyproject_config(directives="""
[[tool.alembic.post_write_hooks]]
name="foo"
bar="somebar"
- """
- )
+ """)
assert_raises_message(
util.CommandError,
)
def test_module_config_missing(self):
- self.cfg = _no_sql_testing_config(
- directives=(
- """
+ self.cfg = _no_sql_testing_config(directives=("""
[post_write_hooks]
hooks = ruff
ruff.type = module
- """
- )
- )
+ """))
assert_raises_message(
util.CommandError,
"Key ruff.module is required for post write hook 'ruff'",
)
def test_module_not_found(self):
- self.cfg = _no_sql_testing_config(
- directives=(
- """
+ self.cfg = _no_sql_testing_config(directives=("""
[post_write_hooks]
hooks = ruff
ruff.type = module
ruff.module = ruff_not_found
- """
- )
- )
+ """))
assert_raises_message(
util.CommandError,
"Could not find module ruff_not_found",
from alembic.testing.fixtures import TestBase
from alembic.testing.suite._autogen_fixtures import AutogenFixtureTest
-
if True:
from alembic.autogenerate.compare.server_defaults import (
_render_server_default_for_compare,
def downgrade():
op.drop_table("sometable")
-"""
- % self.rid,
+""" % self.rid,
)
def _distinct_enum_script(self):
op.drop_table("sometable")
ENUM(name="pgenum").drop(op.get_bind(), checkfirst=False)
-"""
- % self.rid,
+""" % self.rid,
)
def test_offline_inline_enum_create(self):
@classmethod
def insert_data(cls, connection):
- connection.execute(
- text(
- """
+ connection.execute(text("""
insert into tab (col) values
('old data 1'),
('old data 2.1'),
('old data 3')
- """
- )
- )
+ """))
def test_inline_percent(self, connection, ops_context):
# TODO: here's the issue, you need to escape this.
def downgrade():
op.execute("DROP TABLE foo")
- """
- % a,
+ """ % a,
sourceless=self.sourceless,
)
def downgrade():
op.execute("DROP TABLE bar")
- """
- % (b, a),
+ """ % (b, a),
sourceless=self.sourceless,
)
def downgrade():
op.execute("DROP TABLE bat")
- """
- % (c, b),
+ """ % (c, b),
sourceless=self.sourceless,
)
@staticmethod
def _env_file_fixture():
- env_file_fixture(
- textwrap.dedent(
- """\
+ env_file_fixture(textwrap.dedent("""\
import alembic
from alembic import context
from sqlalchemy import engine_from_config, pool
run_migrations_offline()
else:
run_migrations_online()
- """
- )
- )
+ """))
def test_steps(self):
import alembic
def downgrade():
pass
-"""
- % (a,),
+""" % (a,),
)
script.generate_revision(b, "revision b", refresh=True)
write_script(
def downgrade():
pass
-"""
- % (b, a),
+""" % (b, a),
)
script.generate_revision(c, "revision c", refresh=True)
write_script(
def downgrade():
pass
-"""
- % (c, b),
+""" % (c, b),
)
return a, b, c
def test_noerr_transaction_opened_externally(self):
a, b, c = self._opened_transaction_fixture()
- env_file_fixture(
- """
+ env_file_fixture("""
from sqlalchemy import engine_from_config, pool
def run_migrations_online():
run_migrations_online()
-"""
- )
+""")
command.stamp(self.cfg, c)
write_script(
script,
self.a,
- (
- """# coding: utf-8
+ ("""# coding: utf-8
from __future__ import unicode_literals
revision = '%s'
down_revision = None
def downgrade():
op.execute("drôle de petite voix m’a réveillé")
-"""
- % self.a
- ),
+""" % self.a),
encoding="utf-8",
)
def downgrade():
op.execute("DROP TABLE foo")
- """
- % a,
+ """ % a,
)
script = ScriptDirectory.from_config(self.cfg)
path = script.get_revision(a).path
with open(path, "w") as fp:
- fp.write(
- """
+ fp.write("""
down_revision = None
from alembic import op
def downgrade():
op.execute("DROP TABLE foo")
-"""
- )
+""")
pyc_path = util.pyc_file_from_path(path)
if pyc_path is not None and os.access(pyc_path, os.F_OK):
os.unlink(pyc_path)
def downgrade():
op.execute("DROP TABLE foo")
- """
- % a,
+ """ % a,
sourceless=True,
)
def downgrade():
pass
-"""
- % (name, model, name),
+""" % (name, model, name),
)
def tearDown(self):
with open(rev.path) as handle:
result = handle.read()
- assert (
- (
- '''
+ assert ('''
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
sa.PrimaryKeyConstraint('id')
)
# ### end Alembic commands ###
-'''
- )
- in result
- )
+''') in result
class ScriptAccessorTest(TestBase):
clear_staging_env()
def test_autogen(self):
- self._write_metadata(
- """
+ self._write_metadata("""
import sqlalchemy as sa
m1 = sa.MetaData()
sa.Table('e2t1', m2, sa.Column('y', sa.Integer))
sa.Table('e3t1', m3, sa.Column('z', sa.Integer))
-"""
- )
+""")
rev = command.revision(
self.cfg, message="some message", autogenerate=True
eq_(template_args, {"x": "x1", "y": "y2", "z": "z1", "q": "q1"})
def test_tmpl_args_revision(self):
- env_file_fixture(
- """
+ env_file_fixture("""
context.configure(dialect_name='sqlite', template_args={"somearg":"somevalue"})
-"""
- )
- script_file_fixture(
- """
+""")
+ script_file_fixture("""
# somearg: ${somearg}
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
-"""
- )
+""")
command.revision(self.cfg, message="some rev")
script = ScriptDirectory.from_config(self.cfg)
assert "somearg: somevalue" in text
def test_bad_render(self):
- env_file_fixture(
- """
+ env_file_fixture("""
context.configure(dialect_name='sqlite', template_args={"somearg":"somevalue"})
-"""
- )
- script_file_fixture(
- """
+""")
+ script_file_fixture("""
<% z = x + y %>
-"""
- )
+""")
try:
command.revision(self.cfg, message="some rev")
def downgrade():
pass
-"""
- % (name, model, name),
+""" % (name, model, name),
)
def tearDown(self):
suffix = "..." if file_info.AddEllipsis and docs else ""
- func_text = textwrap.dedent(
- f"""
+ func_text = textwrap.dedent(f"""
{overload}
{contextmanager}
def {name}{argspec}: {"..." if not docs else ""}
{docs}
{suffix}
- """
- )
+ """)
return func_text
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
