import importlib.machinery
import os
+from pathlib import Path
import shutil
import textwrap
def _get_staging_directory():
if provision.FOLLOWER_IDENT:
- return "scratch_%s" % provision.FOLLOWER_IDENT
+ return f"scratch_{provision.FOLLOWER_IDENT}"
else:
return "scratch"
def staging_env(create=True, template="generic", sourceless=False):
cfg = _testing_config()
if create:
- path = os.path.join(_get_staging_directory(), "scripts")
+ path = _join_path(_get_staging_directory(), "scripts")
assert not os.path.exists(path), (
"staging directory %s already exists; poor cleanup?" % path
)
"pep3147_everything",
), sourceless
make_sourceless(
- os.path.join(path, "env.py"),
+ _join_path(path, "env.py"),
"pep3147" if "pep3147" in sourceless else "simple",
)
def script_file_fixture(txt):
- dir_ = os.path.join(_get_staging_directory(), "scripts")
- path = os.path.join(dir_, "script.py.mako")
+ dir_ = _join_path(_get_staging_directory(), "scripts")
+ path = _join_path(dir_, "script.py.mako")
with open(path, "w") as f:
f.write(txt)
def env_file_fixture(txt):
- dir_ = os.path.join(_get_staging_directory(), "scripts")
+ dir_ = _join_path(_get_staging_directory(), "scripts")
txt = (
"""
from alembic import context
+ txt
)
- path = os.path.join(dir_, "env.py")
+ path = _join_path(dir_, "env.py")
pyc_path = util.pyc_file_from_path(path)
if pyc_path:
os.unlink(pyc_path)
def _sqlite_file_db(tempname="foo.db", future=False, scope=None, **options):
- dir_ = os.path.join(_get_staging_directory(), "scripts")
+ dir_ = _join_path(_get_staging_directory(), "scripts")
url = "sqlite:///%s/%s" % (dir_, tempname)
if scope and util.sqla_14:
options["scope"] = scope
def _sqlite_testing_config(sourceless=False, future=False):
- dir_ = os.path.join(_get_staging_directory(), "scripts")
- url = "sqlite:///%s/foo.db" % dir_
+ dir_ = _join_path(_get_staging_directory(), "scripts")
+ url = f"sqlite:///{dir_}/foo.db"
sqlalchemy_future = future or ("future" in config.db.__class__.__module__)
return _write_config_file(
- """
+ f"""
[alembic]
-script_location = %s
-sqlalchemy.url = %s
-sourceless = %s
-%s
+script_location = {dir_}
+sqlalchemy.url = {url}
+sourceless = {"true" if sourceless else "false"}
+{"sqlalchemy.future = true" if sqlalchemy_future else ""}
[loggers]
keys = root,sqlalchemy
format = %%(levelname)-5.5s [%%(name)s] %%(message)s
datefmt = %%H:%%M:%%S
"""
- % (
- dir_,
- url,
- "true" if sourceless else "false",
- "sqlalchemy.future = true" if sqlalchemy_future else "",
- )
)
def _multi_dir_testing_config(sourceless=False, extra_version_location=""):
- dir_ = os.path.join(_get_staging_directory(), "scripts")
+ dir_ = _join_path(_get_staging_directory(), "scripts")
sqlalchemy_future = "future" in config.db.__class__.__module__
url = "sqlite:///%s/foo.db" % dir_
return _write_config_file(
- """
+ f"""
[alembic]
-script_location = %s
-sqlalchemy.url = %s
-sqlalchemy.future = %s
-sourceless = %s
-version_locations = %%(here)s/model1/ %%(here)s/model2/ %%(here)s/model3/ %s
+script_location = {dir_}
+sqlalchemy.url = {url}
+sqlalchemy.future = {"true" if sqlalchemy_future else "false"}
+sourceless = {"true" if sourceless else "false"}
+version_locations = %(here)s/model1/ %(here)s/model2/ %(here)s/model3/ \
+{extra_version_location}
[loggers]
keys = root
format = %%(levelname)-5.5s [%%(name)s] %%(message)s
datefmt = %%H:%%M:%%S
"""
- % (
- dir_,
- url,
- "true" if sqlalchemy_future else "false",
- "true" if sourceless else "false",
- extra_version_location,
- )
)
def _no_sql_testing_config(dialect="postgresql", directives=""):
"""use a postgresql url with no host so that
connections guaranteed to fail"""
- dir_ = os.path.join(_get_staging_directory(), "scripts")
+ dir_ = _join_path(_get_staging_directory(), "scripts")
return _write_config_file(
- """
+ f"""
[alembic]
-script_location = %s
-sqlalchemy.url = %s://
-%s
+script_location ={dir_}
+sqlalchemy.url = {dialect}://
+{directives}
[loggers]
keys = root
datefmt = %%H:%%M:%%S
"""
- % (dir_, dialect, directives)
)
if not os.access(_get_staging_directory(), os.F_OK):
os.mkdir(_get_staging_directory())
- return Config(os.path.join(_get_staging_directory(), "test_alembic.ini"))
+ return Config(_join_path(_get_staging_directory(), "test_alembic.ini"))
def write_script(
script = Script._from_path(scriptdir, path)
old = scriptdir.revision_map.get_revision(script.revision)
if old.down_revision != script.down_revision:
- raise Exception(
- "Can't change down_revision " "on a refresh operation."
- )
+ raise Exception("Can't change down_revision on a refresh operation.")
scriptdir.revision_map.add_revision(script, _replace=True)
if sourceless:
write_script(
script,
a,
- """\
+ f"""\
"Rev A"
-revision = '%s'
+revision = '{a}'
down_revision = None
from alembic import op
def downgrade():
op.execute("DROP STEP 1")
-"""
- % a,
+""",
)
script.generate_revision(b, "revision b", refresh=True, head=a)
write_script(
script,
c,
- """\
+ f"""\
"Rev C"
-revision = '%s'
-down_revision = '%s'
+revision = '{c}'
+down_revision = '{b}'
from alembic import op
def downgrade():
op.execute("DROP STEP 3")
-"""
- % (c, b),
+""",
)
return a, b, c
write_script(
script,
d,
- """\
+ f"""\
"Rev D"
-revision = '%s'
-down_revision = '%s'
+revision = '{d}'
+down_revision = '{b}'
from alembic import op
def downgrade():
op.execute("DROP STEP 4")
-"""
- % (d, b),
+""",
)
script.generate_revision(
write_script(
script,
e,
- """\
+ f"""\
"Rev E"
-revision = '%s'
-down_revision = '%s'
+revision = '{e}'
+down_revision = '{d}'
from alembic import op
def downgrade():
op.execute("DROP STEP 5")
-"""
- % (e, d),
+""",
)
script.generate_revision(
write_script(
script,
f,
- """\
+ f"""\
"Rev F"
-revision = '%s'
-down_revision = '%s'
+revision = '{f}'
+down_revision = '{b}'
from alembic import op
def downgrade():
op.execute("DROP STEP 6")
-"""
- % (f, b),
+""",
)
return d, e, f
def _multidb_testing_config(engines):
"""alembic.ini fixture to work exactly with the 'multidb' template"""
- dir_ = os.path.join(_get_staging_directory(), "scripts")
+ dir_ = _join_path(_get_staging_directory(), "scripts")
sqlalchemy_future = "future" in config.db.__class__.__module__
databases = ", ".join(engines.keys())
engines = "\n\n".join(
- "[%s]\n" "sqlalchemy.url = %s" % (key, value.url)
+ f"[{key}]\nsqlalchemy.url = {value.url}"
for key, value in engines.items()
)
return _write_config_file(
- """
+ f"""
[alembic]
-script_location = %s
+script_location = {dir_}
sourceless = false
-sqlalchemy.future = %s
-databases = %s
+sqlalchemy.future = {"true" if sqlalchemy_future else "false"}
+databases = {databases}
-%s
+{engines}
[loggers]
keys = root
format = %%(levelname)-5.5s [%%(name)s] %%(message)s
datefmt = %%H:%%M:%%S
"""
- % (dir_, "true" if sqlalchemy_future else "false", databases, engines)
)
+
+
+def _join_path(base: str, *more: str):
+ return str(Path(base).joinpath(*more).as_posix())
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
