from __future__ import annotations
+import importlib.util
import os
import shlex
import subprocess
return cmdline_options_list
-@register("console_scripts")
-def console_scripts(
- path: str, options: dict, ignore_output: bool = False
-) -> None:
+def _get_required_option(options: dict, name: str) -> str:
try:
- entrypoint_name = options["entrypoint"]
+ return options[name]
except KeyError as ke:
raise util.CommandError(
- f"Key {options['_hook_name']}.entrypoint is required for post "
+ f"Key {options['_hook_name']}.{name} is required for post "
f"write hook {options['_hook_name']!r}"
) from ke
+
+
+def _run_hook(
+ path: str, options: dict, ignore_output: bool, command: List[str]
+) -> None:
+ cwd: Optional[str] = options.get("cwd", None)
+ cmdline_options_str = options.get("options", "")
+ cmdline_options_list = _parse_cmdline_options(cmdline_options_str, path)
+
+ kw: Dict[str, Any] = {}
+ if ignore_output:
+ kw["stdout"] = kw["stderr"] = subprocess.DEVNULL
+
+ subprocess.run([*command, *cmdline_options_list], cwd=cwd, **kw)
+
+
+@register("console_scripts")
+def console_scripts(
+ path: str, options: dict, ignore_output: bool = False
+) -> None:
+ entrypoint_name = _get_required_option(options, "entrypoint")
for entry in compat.importlib_metadata_get("console_scripts"):
if entry.name == entrypoint_name:
impl: Any = entry
raise util.CommandError(
f"Could not find entrypoint console_scripts.{entrypoint_name}"
)
- cwd: Optional[str] = options.get("cwd", None)
- cmdline_options_str = options.get("options", "")
- cmdline_options_list = _parse_cmdline_options(cmdline_options_str, path)
-
- kw: Dict[str, Any] = {}
- if ignore_output:
- kw["stdout"] = kw["stderr"] = subprocess.DEVNULL
- subprocess.run(
- [
- sys.executable,
- "-c",
- f"import {impl.module}; {impl.module}.{impl.attr}()",
- ]
- + cmdline_options_list,
- cwd=cwd,
- **kw,
- )
+ command = [
+ sys.executable,
+ "-c",
+ f"import {impl.module}; {impl.module}.{impl.attr}()",
+ ]
+ _run_hook(path, options, ignore_output, command)
@register("exec")
def exec_(path: str, options: dict, ignore_output: bool = False) -> None:
- try:
- executable = options["executable"]
- except KeyError as ke:
- raise util.CommandError(
- f"Key {options['_hook_name']}.executable is required for post "
- f"write hook {options['_hook_name']!r}"
- ) from ke
- cwd: Optional[str] = options.get("cwd", None)
- cmdline_options_str = options.get("options", "")
- cmdline_options_list = _parse_cmdline_options(cmdline_options_str, path)
+ executable = _get_required_option(options, "executable")
+ _run_hook(path, options, ignore_output, command=[executable])
- kw: Dict[str, Any] = {}
- if ignore_output:
- kw["stdout"] = kw["stderr"] = subprocess.DEVNULL
- subprocess.run(
- [
- executable,
- *cmdline_options_list,
- ],
- cwd=cwd,
- **kw,
- )
+@register("module")
+def module(path: str, options: dict, ignore_output: bool = False) -> None:
+ module_name = _get_required_option(options, "module")
+
+ if importlib.util.find_spec(module_name) is None:
+ raise util.CommandError(f"Could not find module {module_name}")
+
+ command = [sys.executable, "-m", module_name]
+ _run_hook(path, options, ignore_output, command)
# black.entrypoint = black
# black.options = -l 79 REVISION_SCRIPT_FILENAME
-# lint with attempts to fix using "ruff" - use the exec runner, execute a binary
+# lint with attempts to fix using "ruff" - use the module runner, against the "ruff" module
+# hooks = ruff
+# ruff.type = module
+# ruff.module = ruff
+# ruff.options = check --fix REVISION_SCRIPT_FILENAME
+
+# Alternatively, use the exec runner to execute a binary found on your PATH
# hooks = ruff
# ruff.type = exec
-# ruff.executable = %(here)s/.venv/bin/ruff
+# ruff.executable = ruff
# ruff.options = check --fix REVISION_SCRIPT_FILENAME
# Logging configuration. This is also consumed by the user-maintained
# black.entrypoint = black
# black.options = -l 79 REVISION_SCRIPT_FILENAME
-# lint with attempts to fix using "ruff" - use the exec runner, execute a binary
+# lint with attempts to fix using "ruff" - use the module runner, against the "ruff" module
+# hooks = ruff
+# ruff.type = module
+# ruff.module = ruff
+# ruff.options = check --fix REVISION_SCRIPT_FILENAME
+
+# Alternatively, use the exec runner to execute a binary found on your PATH
# hooks = ruff
# ruff.type = exec
-# ruff.executable = %(here)s/.venv/bin/ruff
+# ruff.executable = ruff
# ruff.options = check --fix REVISION_SCRIPT_FILENAME
# Logging configuration. This is also consumed by the user-maintained
# black.entrypoint = black
# black.options = -l 79 REVISION_SCRIPT_FILENAME
-# lint with attempts to fix using "ruff" - use the exec runner, execute a binary
+# lint with attempts to fix using "ruff" - use the module runner, against the "ruff" module
+# hooks = ruff
+# ruff.type = module
+# ruff.module = ruff
+# ruff.options = check --fix REVISION_SCRIPT_FILENAME
+
+# Alternatively, use the exec runner to execute a binary found on your PATH
# hooks = ruff
# ruff.type = exec
-# ruff.executable = %(here)s/.venv/bin/ruff
+# ruff.executable = ruff
# ruff.options = check --fix REVISION_SCRIPT_FILENAME
# Logging configuration. This is also consumed by the user-maintained
# options = "-l 79 REVISION_SCRIPT_FILENAME"
#
# [[tool.alembic.post_write_hooks]]
-# lint with attempts to fix using "ruff" - use the exec runner,
-# execute a binary
+# lint with attempts to fix using "ruff" - use the module runner, against the "ruff" module
+# name = "ruff"
+# type = "module"
+# module = "ruff"
+# options = "check --fix REVISION_SCRIPT_FILENAME"
+#
+# [[tool.alembic.post_write_hooks]]
+# Alternatively, use the exec runner to execute a binary found on your PATH
# name = "ruff"
# type = "exec"
-# executable = "%(here)s/.venv/bin/ruff"
+# executable = "ruff"
# options = "check --fix REVISION_SCRIPT_FILENAME"
# options = "-l 79 REVISION_SCRIPT_FILENAME"
#
# [[tool.alembic.post_write_hooks]]
-# lint with attempts to fix using "ruff" - use the exec runner,
-# execute a binary
+# lint with attempts to fix using "ruff" - use the module runner, against the "ruff" module
+# name = "ruff"
+# type = "module"
+# module = "ruff"
+# options = "check --fix REVISION_SCRIPT_FILENAME"
+#
+# [[tool.alembic.post_write_hooks]]
+# Alternatively, use the exec runner to execute a binary found on your PATH
# name = "ruff"
# type = "exec"
-# executable = "%(here)s/.venv/bin/ruff"
+# executable = "ruff"
# options = "check --fix REVISION_SCRIPT_FILENAME"
configuration for the ``"black"`` post write hook, which includes:
* ``type`` - this is the type of hook we are running. Alembic includes
- two hook runners: ``"console_scripts"``, which is specifically a
- Python function that uses ``subprocess.run()`` to invoke a separate
- Python script against the revision file; and ``"exec"``, which uses
- ``subprocess.run()`` to execute an arbitrary binary. For a custom-written
- hook function, this configuration variable would refer to the name under
- which the custom hook was registered; see the next section for an example.
+ three hook runners:
+
+ * ``"console_scripts"``, which is specifically a Python function that uses
+ ``subprocess.run()`` to invoke a separate Python script against the revision file;
+ * ``"exec"``, which uses ``subprocess.run()`` to execute an arbitrary binary; and
+ * ``"module"``, which uses ``subprocess.run()`` to invoke a Python module directly.
+
+ For a custom-written hook function, this configuration variable would
+ refer to the name under which the custom hook was registered; see the
+ next section for an example.
.. versionadded:: 1.12 added new ``exec`` runner
+.. versionadded:: 1.16.3 added new ``module`` runner
+
+
The following configuration option is specific to the ``"console_scripts"``
hook runner:
# black.entrypoint = black
# black.options = -l 79 REVISION_SCRIPT_FILENAME
- # lint with attempts to fix using "ruff" - use the exec runner, execute a binary
+ # lint with attempts to fix using "ruff" - use the module runner, against the "ruff" module
+ # hooks = ruff
+ # ruff.type = module
+ # ruff.module = ruff
+ # ruff.options = check --fix REVISION_SCRIPT_FILENAME
+
+ # Alternatively, use the exec runner to execute a binary found on your PATH
# hooks = ruff
# ruff.type = exec
- # ruff.executable = %(here)s/.venv/bin/ruff
+ # ruff.executable = ruff
# ruff.options = check --fix REVISION_SCRIPT_FILENAME
# Logging configuration. This is also consumed by the user-maintained
--- /dev/null
+.. change::
+ :tags: usecase, autogenerate
+ :tickets: 1686
+
+ Add "module" post-write hook. This hook type is almost identical to the
+ console_scripts hook, except it's running `python -m black` instead of
+ using black's console_script. It is mainly useful for tools without
+ console scripts (e.g. ruff), but has semantics closer to the
+ console_scripts hook in that it finds the ruff module available to the
+ running interpreter instead of finding an executable by path. Pull request
+ courtesy Frazer McLean.
\ No newline at end of file
+import importlib.util
from pathlib import Path
import sys
self._run_ruff_with_config(
input_config, expected_additional_arguments_fn, cwd="/path/to/cwd"
)
+
+ def test_module_config_missing(self):
+ 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'",
+ command.revision,
+ self.cfg,
+ message="x",
+ )
+
+ def test_module_not_found(self):
+ 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",
+ command.revision,
+ self.cfg,
+ message="x",
+ )
+
+ def _run_black_module_with_config(
+ self,
+ input_config,
+ expected_additional_arguments_fn,
+ cwd=None,
+ use_toml=False,
+ ):
+ if use_toml:
+ self.cfg = _no_sql_pyproject_config(directives=input_config)
+ else:
+ self.cfg = _no_sql_testing_config(directives=input_config)
+
+ black_module_spec = importlib.util.find_spec("black")
+
+ importlib_util_find_spec = mock.Mock(return_value=black_module_spec)
+ with (
+ mock.patch(
+ "importlib.util.find_spec",
+ importlib_util_find_spec,
+ ),
+ mock.patch(
+ "alembic.script.write_hooks.subprocess"
+ ) as mock_subprocess,
+ ):
+ rev = command.revision(self.cfg, message="x")
+
+ eq_(importlib_util_find_spec.mock_calls, [mock.call("black")])
+ eq_(
+ mock_subprocess.mock_calls,
+ [
+ mock.call.run(
+ [
+ sys.executable,
+ "-m",
+ "black",
+ ]
+ + expected_additional_arguments_fn(rev.path),
+ cwd=cwd,
+ )
+ ],
+ )
+
+ @testing.variation("config", ["ini", "toml"])
+ def test_module(self, config):
+ if config.ini:
+ input_config = """
+[post_write_hooks]
+hooks = black
+black.type = module
+black.module = black
+black.options = -l 79
+ """
+ else:
+ input_config = """
+ [[tool.alembic.post_write_hooks]]
+ name = "black"
+ type = "module"
+ module = "black"
+ options = "-l 79"
+ """
+
+ def expected_additional_arguments_fn(rev_path):
+ return [rev_path, "-l", "79"]
+
+ self._run_black_module_with_config(
+ input_config,
+ expected_additional_arguments_fn,
+ use_toml=config.toml,
+ )
+
+ @combinations(True, False, argnames="posix")
+ @testing.variation("config", ["ini", "toml"])
+ def test_module_filename_interpolation(self, posix, config):
+ if config.ini:
+ input_config = """
+[post_write_hooks]
+hooks = black
+black.type = module
+black.module = black
+black.options = arg1 REVISION_SCRIPT_FILENAME 'multi-word arg' \
+ --flag1='REVISION_SCRIPT_FILENAME'
+ """
+ else:
+ input_config = """
+[[tool.alembic.post_write_hooks]]
+name = "black"
+type = "module"
+module = "black"
+options = "arg1 REVISION_SCRIPT_FILENAME 'multi-word arg' --flag1='REVISION_SCRIPT_FILENAME'"
+""" # noqa: E501
+
+ def expected_additional_arguments_fn(rev_path):
+ if compat.is_posix:
+ return [
+ "arg1",
+ rev_path,
+ "multi-word arg",
+ "--flag1=" + rev_path,
+ ]
+ else:
+ return [
+ "arg1",
+ rev_path,
+ "'multi-word arg'",
+ "--flag1='%s'" % rev_path,
+ ]
+
+ with mock.patch("alembic.util.compat.is_posix", posix):
+ self._run_black_module_with_config(
+ input_config,
+ expected_additional_arguments_fn,
+ use_toml=config.toml,
+ )
+
+ def test_module_path_in_config(self):
+ input_config = """
+[post_write_hooks]
+hooks = black
+black.type = module
+black.module = black
+black.options = arg1 REVISION_SCRIPT_FILENAME --config %(here)s/pyproject.toml
+ """
+
+ def expected_additional_arguments_fn(rev_path):
+ return [
+ "arg1",
+ rev_path,
+ "--config",
+ Path(_get_staging_directory(), "pyproject.toml")
+ .absolute()
+ .as_posix(),
+ ]
+
+ self._run_black_module_with_config(
+ input_config, expected_additional_arguments_fn
+ )
+
+ def test_module_black_with_cwd(self):
+ input_config = """
+[post_write_hooks]
+hooks = black
+black.type = module
+black.module = black
+black.cwd = /path/to/cwd
+ """
+
+ def expected_additional_arguments_fn(rev_path):
+ return [rev_path]
+
+ self._run_black_module_with_config(
+ input_config, expected_additional_arguments_fn, cwd="/path/to/cwd"
+ )
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
