Code Quality Tool: toml-based backend templating

src/python/pants/backend/adhoc/code_quality_tool.py

@@ -41,6 +41,8 @@
     StringSequenceField,
     Target,
     Targets,
+    WrappedTarget,
+    WrappedTargetRequest,
 )
 from pants.option.option_types import SkipOption
 from pants.option.subsystem import Subsystem
@@ -158,6 +160,7 @@ class CodeQualityToolTarget(Target):
 @dataclass(frozen=True)
 class CodeQualityToolAddressString:
     address: str
+    description_of_origin: str
 
 
 @dataclass(frozen=True)
@@ -178,12 +181,17 @@ async def find_code_quality_tool(request: CodeQualityToolAddressString) -> CodeQ
         AddressInput,
         AddressInput.parse(request.address, description_of_origin="code quality tool target"),
     )
-
     addresses = Addresses((tool_address,))
     addresses.expect_single()
 
-    tool_targets = await Get(Targets, Addresses, addresses)
-    target = tool_targets[0]
+    wrapped_target = await Get(
+        WrappedTarget,
+        WrappedTargetRequest(
+            address=tool_address,
+            description_of_origin=f"the configured `target` for Code quality tool {request.description_of_origin}",
+        ),
+    )
+    target = wrapped_target.target
     runnable_address_str = target[CodeQualityToolRunnableField].value
     if not runnable_address_str:
         raise Exception(f"Must supply a value for `runnable` for {request.address}.")
@@ -359,6 +367,9 @@ def rules(self) -> Iterable[Rule]:
         else:
             raise ValueError(f"Unsupported goal for code quality tool: {self.goal}")
 
+    def _tool_address(self):
+        return CodeQualityToolAddressString(address=self.target, description_of_origin=self.name)
+
     def _build_lint_rules(self) -> Iterable[Rule]:
         class CodeQualityToolInstance(Subsystem):
             options_scope = self.scope
@@ -378,7 +389,7 @@ async def partition_inputs(
             if subsystem.skip:
                 return Partitions()
 
-            cqt = await Get(CodeQualityTool, CodeQualityToolAddressString(address=self.target))
+            cqt = await Get(CodeQualityTool, CodeQualityToolAddressString, self._tool_address())
 
             matching_filepaths = FilespecMatcher(
                 includes=cqt.file_glob_include,
@@ -391,7 +402,7 @@ async def partition_inputs(
         async def run_code_quality(request: CodeQualityProcessingRequest.Batch) -> LintResult:
             sources_snapshot, code_quality_tool_runner = await MultiGet(
                 Get(Snapshot, PathGlobs(request.elements)),
-                Get(CodeQualityToolBatchRunner, CodeQualityToolAddressString(address=self.target)),
+                Get(CodeQualityToolBatchRunner, CodeQualityToolAddressString, self._tool_address()),
             )
 
             proc_result = await Get(
@@ -431,7 +442,7 @@ async def partition_inputs(
             if subsystem.skip:
                 return Partitions()
 
-            cqt = await Get(CodeQualityTool, CodeQualityToolAddressString(address=self.target))
+            cqt = await Get(CodeQualityTool, CodeQualityToolAddressString, self._tool_address())
 
             matching_filepaths = FilespecMatcher(
                 includes=cqt.file_glob_include,
@@ -445,7 +456,7 @@ async def run_code_quality(request: CodeQualityProcessingRequest.Batch) -> FmtRe
             sources_snapshot = request.snapshot
 
             code_quality_tool_runner = await Get(
-                CodeQualityToolBatchRunner, CodeQualityToolAddressString(address=self.target)
+                CodeQualityToolBatchRunner, CodeQualityToolAddressString, self._tool_address()
             )
 
             proc_result = await Get(
@@ -493,7 +504,7 @@ async def partition_inputs(
             if subsystem.skip:
                 return Partitions()
 
-            cqt = await Get(CodeQualityTool, CodeQualityToolAddressString(address=self.target))
+            cqt = await Get(CodeQualityTool, CodeQualityToolAddressString, self._tool_address())
 
             matching_filepaths = FilespecMatcher(
                 includes=cqt.file_glob_include,
@@ -507,7 +518,7 @@ async def run_code_quality(request: CodeQualityProcessingRequest.Batch) -> FixRe
             sources_snapshot = request.snapshot
 
             code_quality_tool_runner = await Get(
-                CodeQualityToolBatchRunner, CodeQualityToolAddressString(address=self.target)
+                CodeQualityToolBatchRunner, CodeQualityToolAddressString, self._tool_address()
             )
 
             proc_result = await Get(

src/python/pants/backend/experimental/adhoc/code_quality_tool_backend_template.py

@@ -0,0 +1,54 @@
+# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
+# Licensed under the Apache License, Version 2.0 (see LICENSE).
+from dataclasses import dataclass
+from typing import Any
+
+from pants.backend.adhoc import code_quality_tool, run_system_binary
+from pants.backend.adhoc.code_quality_tool import CodeQualityToolRuleBuilder, CodeQualityToolTarget
+from pants.backend.adhoc.target_types import SystemBinaryTarget
+from pants.base.exceptions import BackendConfigurationError
+from pants.core.util_rules import adhoc_process_support
+
+
+@dataclass
+class CodeQualityToolBackend:
+    rule_builder: CodeQualityToolRuleBuilder
+
+    def rules(self):
+        return [
+            *run_system_binary.rules(),
+            *adhoc_process_support.rules(),
+            *code_quality_tool.base_rules(),
+            *self.rule_builder.rules(),
+        ]
+
+    def target_types(self):
+        return [
+            SystemBinaryTarget,
+            CodeQualityToolTarget,
+        ]
+
+
+def _validate(kwargs: dict[str, Any]) -> dict[str, Any]:
+    required = ["goal", "target", "name"]
+    missing = [k for k in required if k not in kwargs]
+    if missing:
+        raise BackendConfigurationError(
+            f"Missing required keys {missing} for backend template {__name__}."
+            f" Supplied {dict(kwargs)}."
+        )
+
+    return {k: str(v) for k, v in kwargs.items()}
+
+
+def generate(backend_package_alias: str, kwargs: dict[str, Any]):
+    kwargs = _validate(kwargs)
+
+    rule_builder = CodeQualityToolRuleBuilder(
+        goal=kwargs["goal"],
+        target=kwargs["target"],
+        name=kwargs["name"],
+        scope=backend_package_alias,
+    )
+
+    return CodeQualityToolBackend(rule_builder)

src/python/pants/backend/experimental/adhoc/code_quality_tool_backend_template_integration_test.py

@@ -0,0 +1,57 @@
+# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
+# Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+from textwrap import dedent
+
+from pants.testutil.pants_integration_test import run_pants, setup_tmpdir
+
+
+def test_code_quality_tool_backend_generation() -> None:
+    sources = {
+        "src/BUILD": dedent(
+            """
+        python_source(name="bad_to_good", source="bad_to_good.py")
+
+        code_quality_tool(
+            name="bad_to_good_tool",
+            runnable=":bad_to_good",
+            file_glob_include=["**/*.py"],
+            file_glob_exclude=["**/bad_to_good.py"],
+        )
+        """
+        ),
+        "src/bad_to_good.py": dedent(
+            """
+        import sys
+
+        for fpath in sys.argv[1:]:
+            with open(fpath) as f:
+                contents = f.read()
+            if 'badcode' in contents:
+                with open(fpath, 'w') as f:
+                    f.write(contents.replace('badcode', 'goodcode'))
+            """
+        ),
+        "src/good_fmt.py": "thisisfine = 5\n",
+        "src/needs_repair.py": "badcode = 10\n",
+    }
+    with setup_tmpdir(sources) as tmpdir:
+        templated_backends = {
+            "badcodefixer": {
+                "template": "pants.backend.experimental.adhoc.code_quality_tool_backend_template",
+                "goal": "fix",
+                "target": f"{tmpdir}/src:bad_to_good_tool",
+                "name": "Bad to Good",
+            }
+        }
+        args = [
+            "--backend-packages=['pants.backend.python', 'badcodefixer']",
+            f"--templated-backends={templated_backends}",
+            f"--source-root-patterns=['{tmpdir}/src']",
+            "fix",
+            f"{tmpdir}/src::",
+        ]
+        result = run_pants(args)
+        assert "badcodefixer made changes" in result.stderr.strip()
+        with open(f"{tmpdir}/src/needs_repair.py") as fixed_file:
+            assert "goodcode = 10\n" == fixed_file.read()

src/python/pants/init/backend_templating.py

@@ -0,0 +1,20 @@
+# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
+# Licensed under the Apache License, Version 2.0 (see LICENSE).
+from dataclasses import dataclass
+from typing import Any, cast
+
+from pants.util.frozendict import FrozenDict
+
+
+@dataclass(frozen=True)
+class TemplatedBackendConfig:
+    template: str
+    kwargs: FrozenDict[str, Any]
+
+    @classmethod
+    def from_dict(cls, d: Any):
+        d = dict(d)
+        template = d.pop("template", None)
+        if not template:
+            raise ValueError('"template" is a required key for a templated backend')
+        return cls(template=cast(str, template), kwargs=FrozenDict(d))

src/python/pants/init/extension_loader.py

@@ -4,13 +4,14 @@
 import importlib
 import logging
 import traceback
-from typing import Dict, List, Optional
+from typing import Dict, List, Mapping, Optional
 
 from pkg_resources import Requirement, WorkingSet
 
 from pants.base.exceptions import BackendConfigurationError
 from pants.build_graph.build_configuration import BuildConfiguration
 from pants.goal.builtins import register_builtin_goals
+from pants.init.backend_templating import TemplatedBackendConfig
 from pants.util.ordered_set import FrozenOrderedSet
 
 logger = logging.getLogger(__name__)
@@ -33,6 +34,7 @@ def load_backends_and_plugins(
     working_set: WorkingSet,
     backends: List[str],
     bc_builder: Optional[BuildConfiguration.Builder] = None,
+    templated_backends: Optional[Mapping[str, TemplatedBackendConfig]] = None,
 ) -> BuildConfiguration:
     """Load named plugins and source backends.
 
@@ -42,7 +44,9 @@ def load_backends_and_plugins(
     :param bc_builder: The BuildConfiguration (for adding aliases).
     """
     bc_builder = bc_builder or BuildConfiguration.Builder()
-    load_build_configuration_from_source(bc_builder, backends)
+    load_build_configuration_from_source(
+        bc_builder, backends, templated_backends=templated_backends
+    )
     load_plugins(bc_builder, plugins, working_set)
     register_builtin_goals(bc_builder)
     return bc_builder.create()
@@ -112,7 +116,9 @@ def load_plugins(
 
 
 def load_build_configuration_from_source(
-    build_configuration: BuildConfiguration.Builder, backends: List[str]
+    build_configuration: BuildConfiguration.Builder,
+    backends: List[str],
+    templated_backends: Optional[Mapping[str, TemplatedBackendConfig]] = None,
 ) -> None:
     """Installs pants backend packages to provide BUILD file symbols and cli goals.
 
@@ -123,25 +129,60 @@ def load_build_configuration_from_source(
     """
     # NB: Backends added here must be explicit dependencies of this module.
     backend_packages = FrozenOrderedSet(["pants.core", "pants.backend.project_info", *backends])
+    templated_backends = templated_backends or {}
+
     for backend_package in backend_packages:
-        load_backend(build_configuration, backend_package)
+        load_backend(
+            build_configuration,
+            backend_package,
+            templating_config=templated_backends.get(backend_package),
+        )
 
 
-def load_backend(build_configuration: BuildConfiguration.Builder, backend_package: str) -> None:
+def _load_register_module(backend_package: str):
+    module_source = backend_package + ".register"
+    try:
+        module = importlib.import_module(module_source)
+    except ImportError as ex:
+        traceback.print_exc()
+        raise BackendConfigurationError(f"Failed to load the {module_source} backend: {ex!r}")
+    return module, module_source
+
+
+def _generate_backend_module_from_template(backend_package: str, cfg: TemplatedBackendConfig):
+    try:
+        backend_generator_module = importlib.import_module(cfg.template)
+    except ImportError as ex:
+        traceback.print_exc()
+        raise BackendConfigurationError(
+            f"Failed to load the {cfg.template} backend template module: {ex!r}"
+        )
+    module = backend_generator_module.generate(backend_package, cfg.kwargs)
+    module_source = f"{module} generated from {cfg.template}"
+    return module, module_source
+
+
+def load_backend(
+    build_configuration: BuildConfiguration.Builder,
+    backend_package: str,
+    templating_config: Optional[TemplatedBackendConfig] = None,
+) -> None:
     """Installs the given backend package into the build configuration.
 
     :param build_configuration: the BuildConfiguration to install the backend plugin into.
     :param backend_package: the package name containing the backend plugin register module that
-      provides the plugin entrypoints.
+      provides the plugin entrypoints. For templated backends, should be a unique alias for the backend.
+    :param templating_config: If supplied, the backend will be generated from this config instead
+      of a `register` module defined in backend_package
     :raises: :class:``pants.base.exceptions.BuildConfigurationError`` if there is a problem loading
       the build configuration.
     """
-    backend_module = backend_package + ".register"
-    try:
-        module = importlib.import_module(backend_module)
-    except ImportError as ex:
-        traceback.print_exc()
-        raise BackendConfigurationError(f"Failed to load the {backend_module} backend: {ex!r}")
+
+    module, module_source = (
+        _load_register_module(backend_package)
+        if not templating_config
+        else _generate_backend_module_from_template(backend_package, templating_config)
+    )
 
     def invoke_entrypoint(name: str):
         entrypoint = getattr(module, name, lambda: None)
@@ -150,7 +191,7 @@ def invoke_entrypoint(name: str):
         except TypeError as e:
             traceback.print_exc()
             raise BackendConfigurationError(
-                f"Entrypoint {name} in {backend_module} must be a zero-arg callable: {e!r}"
+                f"Entrypoint {name} in {module_source} must be a zero-arg callable: {e!r}"
             )
 
     target_types = invoke_entrypoint("target_types")

src/python/pants/init/options_initializer.py

@@ -27,7 +27,7 @@
 from pants.init.plugin_resolver import PluginResolver
 from pants.init.plugin_resolver import rules as plugin_resolver_rules
 from pants.option.errors import UnknownFlagsError
-from pants.option.global_options import DynamicRemoteOptions
+from pants.option.global_options import BootstrapOptions, DynamicRemoteOptions
 from pants.option.options import Options
 from pants.option.options_bootstrapper import OptionsBootstrapper
 from pants.util.requirements import parse_requirements_file
@@ -57,11 +57,14 @@ def _initialize_build_configuration(
             sys.path.append(path)
             pkg_resources.fixup_namespace_packages(path)
 
+    templated_backends = BootstrapOptions.parse_templated_backend_configs(bootstrap_options)
+
     # Load plugins and backends.
     return load_backends_and_plugins(
         bootstrap_options.plugins,
         working_set,
         bootstrap_options.backend_packages,
+        templated_backends=templated_backends,
     )