Add a formatter
How to add a new formatter to the fmt and lint goals.
In Pants, every formatter is (typically) also a linter, meaning that if you can run a tool with ./pants fmt, you can run the same tool in check-only mode with ./pants lint. Start by skimming Add a linter to familiarize yourself with how linters work.
This guide walks through each step of adding shfmt. See here for the final result.
This guide assumes that you are running a formatter that already exists outside of Pants as a stand-alone binary, such as running Black or Prettier.
If you are instead writing your own formatting logic inline, you can skip Step 1. In Step 4, you will not need to use Process.
1. Install your formatter
There are several ways for Pants to install your formatter. See Installing tools. This example will use ExternalTool because there is already a pre-compiled binary for shfmt.
You will also likely want to register some options, like --config, --skip, and --args. Options are registered through a Subsystem. If you are using ExternalTool, this is already a subclass of Subsystem. Otherwise, create a subclass of Subsystem. Then, set the class property options_scope to the name of the tool, e.g. "shfmt" or "prettier". Finally, add options via the class method register_options().
from pants.core.util_rules.external_tool import ExternalTool
from pants.engine.platform import Platform
from pants.option.custom_types import file_option, shell_str
class Shfmt(ExternalTool):
"""An autoformatter for shell scripts (https://github.com/mvdan/sh)."""
options_scope = "shfmt"
default_version = "v3.1.2"
default_known_versions = [
"v3.1.2|darwin|284674897e4f708a8b2e4b549af60ac01da648b9581dc7510c586ce41096edaa|3277184",
"v3.1.2|linux|c5794c1ac081f0028d60317454fe388068ab5af7740a83e393515170a7157dce|3043328",
]
@classmethod
def register_options(cls, register):
super().register_options(register)
register(
"--skip",
type=bool,
default=False,
help="Don't use shfmt when running `./pants fmt` or `./pants lint`.",
)
register(
"--args",
type=list,
member_type=shell_str,
help=(
"Arguments to pass directly to shfmt, e.g. `--shfmt-args='-i 2'`.'"
),
)
register(
"--config",
type=list,
member_type=file_option,
advanced=True,
help="Path to `.editorconfig` file. This must be relative to the build root.",
)
def generate_url(self, plat: Platform) -> str:
plat_str = "linux" if plat == Platform.linux else "darwin"
return (
f"https://github.com/mvdan/sh/releases/download/{self.version}/"
f"shfmt_{self.version}_{plat_str}_amd64"
)
def generate_exe(self, plat: Platform) -> str:
plat_str = "linux" if plat == Platform.linux else "darwin"
return f"./shfmt_{self.version}_{plat_str}_amd64"
Lastly, register your Subsystem with the engine:
from pants.engine.rules import collect_rules
...
def rules():
return collect_rules()
2. Set up LanguageFmtTargets and FmtRequest
Unlike linters and test runners, formatters cannot safely run in parallel because they might overwrite each other. Instead, Pants runs formatters sequentially.
However, it is safe to run formatters belonging to different languages in parallel, such as a Java formatter at the same time as a Python formatter. So, Pants groups formatters by languages via the type LanguageFmtTargets. Within each language, the formatters will run sequentially, but across languages, they may run in parallel.
Pants already defines PythonFmtTargets and PythonFmtRequest, which are both defined in pants.backend.python.lint.python_fmt.
To add support for a new language, copy the below code into a new file like bash_formatters.py, but change the names of the LanguageFmtTargets, LanguageFmtRequest, and the rule to match your new language. Update the LanguageFmtTargets.required_fields class property, and update the body of the rule to use your language.
from dataclasses import dataclass
from typing import Iterable, List, Type
from pants.core.goals.fmt import FmtResult, LanguageFmtResults, LanguageFmtTargets
from pants.core.goals.style_request import StyleRequest
from pants.core.util_rules.source_files import SourceFiles, SourceFilesRequest
from pants.engine.fs import Digest, Snapshot
from pants.engine.rules import Get, collect_rules, rule
from pants.engine.unions import UnionMembership, UnionRule, union
from examples.bash.target_types import BashSources
@dataclass(frozen=True)
class BashFmtTargets(LanguageFmtTargets):
required_fields = (BashSources,)
@union
class BashFmtRequest(StyleRequest):
pass
@rule
async def format_bash_targets(
bash_fmt_targets: BashFmtTargets, union_membership: UnionMembership
) -> LanguageFmtResults:
original_sources = await Get(
SourceFiles,
SourceFilesRequest(target[BashSources] for target in bash_fmt_targets.targets),
)
prior_formatter_result = original_sources.snapshot
results: List[FmtResult] = []
fmt_request_types: Iterable[Type[BashFmtRequest]] = union_membership.union_rules[
BashFmtRequest
]
for fmt_request_type in fmt_request_types:
result = await Get(
FmtResult,
BashFmtRequest,
fmt_request_type(
(
fmt_request_type.field_set_type.create(target)
for target in bash_fmt_targets.targets
),
prior_formatter_result=prior_formatter_result,
),
)
if result != FmtResult.noop():
results.append(result)
if result.did_change:
prior_formatter_result = await Get(Snapshot, Digest, result.output)
return LanguageFmtResults(
tuple(results),
input=original_sources.snapshot.digest,
output=prior_formatter_result.digest,
)
def rules():
return [*collect_rules(), UnionRule(LanguageFmtTargets, BashFmtTargets)]
Note that we use await Get(FmtResult, BashFmtRequest) in a for loop. Normally, you should use await MultiGet() instead of a for loop, but we must use a for loop here to run sequentially.
3. Set up a FieldSet and FmtRequest/LintRequest
As described in Rules and the Target API, a FieldSet is a way to tell Pants which Fields you care about targets having for your plugin to work.
Usually, you should add a subclass of the Sources field to the class property required_fields, such as BashSources or PythonSources. This means that your linter will run on any target with that sources field or a subclass of it.
Create a new dataclass that subclasses FieldSet:
from dataclasses import dataclass
from pants.engine.target import FieldSet
...
@dataclass(frozen=True)
class ShfmtFieldSet(FieldSet):
required_fields = (BashSources,)
sources: BashSources
Then, hook this up to a new subclass of both LintRequest and the FmtRequest you set up in step 2 for your language, such as BashFmtRequest or PythonFmtRequest:
from pants.core.goals.lint import LintRequest
...
class ShfmtRequest(BashFmtRequest, LintRequest):
field_set_type = ShfmtFieldSet
Finally, register your new LintRequest/FmtRequest with two UnionRules so that Pants knows your formatter exists:
from pants.engine.unions import UnionRule
...
def rules():
return [
*collect_rules(),
UnionRule(BashFmtRequest, ShfmtRequest),
UnionRule(LintRequest, ShfmtRequest),
]
4. Create fmt and lint rules
You will need rules for both fmt and lint. Both rules should take the LintRequest/FmtRequest from step 3 (e.g. ShfmtRequest) as a parameter. The fmt rule should return FmtResult, and the lint rule should return LintResults.
@rule(desc="Format with shfmt")
async def shfmt_fmt(request: ShfmtRequest, shfmt: Shfmt) -> FmtResult:
...
return FmtResult(...)
@rule(desc="Lint with shfmt")
async def shfmt_lint(request: ShfmtRequest, shfmt: Shfmt) -> LintResults:
...
return LintResults()
The fmt and lint rules will be very similar, except that a) the argv to your Process will be different, b) for lint, you should use await Get(FallibleProcessResult, Process) so that you tolerate failures, whereas fmt should use await Get(ProcessResult, Process). To avoid duplication between the fmt and lint rules, you should set up a helper setup rule, along with dataclasses for SetupRequest and Setup.
@dataclass(frozen=True)
class SetupRequest:
request: ShfmtRequest
check_only: bool
@dataclass(frozen=True)
class Setup:
process: Process
original_digest: Digest
@rule(level=LogLevel.DEBUG)
async def setup_shfmt(setup_request: SetupRequest, shfmt: Shfmt) -> Setup:
download_shfmt_request = Get(
DownloadedExternalTool,
ExternalToolRequest,
shfmt.get_request(Platform.current),
)
# If the user specified `--shfmt-config`, we must search for the file they specified with
# `PathGlobs` to include it in the `input_digest`. We error if the file cannot be found.
config_digest_request = Get(
Digest,
PathGlobs(
globs=[shfmt.options.config] if shfmt.options.config else [],
glob_match_error_behavior=GlobMatchErrorBehavior.error,
description_of_origin="the option `--shfmt-config`",
),
)
source_files_request = Get(
SourceFiles,
SourceFilesRequest(
field_set.sources for field_set in setup_request.request.field_sets
),
)
downloaded_shfmt, config_digest, source_files = await MultiGet(
download_shfmt_request, config_digest_request, source_files_request
)
# If we were given an input digest from a previous formatter for the source files, then we
# should use that input digest instead of the one we read from the filesystem.
source_files_snapshot = (
source_files.snapshot
if setup_request.request.prior_formatter_result is None
else setup_request.request.prior_formatter_result
)
input_digest = await Get(
Digest,
MergeDigests(
(source_files_snapshot.digest, downloaded_shfmt.digest, config_digest)
),
)
argv = [
downloaded_shfmt.exe,
"-d" if setup_request.check_only else "-w",
*shfmt.options.args,
*source_files_snapshot.files,
]
process = Process(
argv=argv,
input_digest=input_digest,
output_files=source_files_snapshot.files,
description=f"Run shfmt on {pluralize(len(setup_request.request.field_sets), 'file')}.",
level=LogLevel.DEBUG,
)
return Setup(process, original_digest=source_files_snapshot.digest)
@rule(desc="Format with shfmt", level=LogLevel.DEBUG)
async def shfmt_fmt(request: ShfmtRequest, shfmt: Shfmt) -> FmtResult:
if shfmt.options.skip:
return FmtResult.noop()
setup = await Get(Setup, SetupRequest(request, check_only=False))
result = await Get(ProcessResult, Process, setup.process)
return FmtResult.from_process_result(
result, original_digest=setup.original_digest, formatter_name="shfmt"
)
@rule(desc="Lint with shfmt", level=LogLevel.DEBUG)
async def shfmt_lint(request: ShfmtRequest, shfmt: Shfmt) -> LintResults:
if shfmt.options.skip:
return LintResults()
setup = await Get(Setup, SetupRequest(request, check_only=True))
result = await Get(FallibleProcessResult, Process, setup.process)
return LintResults(
[LintResult.from_fallible_process_result(result, linter_name="shfmt")]
)
The FmtRequest/LintRequest has a property called .field_sets, which stores a collection of the FieldSets defined in step 2. Each FieldSet corresponds to a single target. Pants will have already validated that there is at least one valid FieldSet, so you can expect ShfmtRequest.field_sets to have 1-n FieldSet instances.
If you have a --skip option, you should check if it was used at the beginning of your fmt and lint rules and, if so, to early return an empty LintResults() and FmtResult.noop().
Use Get(SourceFiles, SourceFilesRequest) to get all the sources you want to run your linter on. However, you should check if the FmtRequest.prior_formatter_result is set, and if so, use that value instead. This ensure that the result of any previous formatters is used, rather than the original source files.
If you used ExternalTool in step 1, you will use Get(DownloadedExternalTool, ExternalToolRequest) in the setup rule to install the tool.
If you have a --config option, you should use Get(Digest, PathGlobs) to find the config file and include it in the input_digest.
Use Get(Digest, MergeDigests) to combine the different inputs together, such as merging the source files, config file, and downloaded tool.
Finally, update your plugin's register.py to activate this file's rules. Note that we must register the rules added in Step 2, as well.
from bash import bash_formatters, shfmt
def rules():
return [*bash_formatters.rules(), *shfmt.rules()]
Now, when you run ./pants fmt :: or ./pants lint ::, your new formatter should run.
5. Add tests (optional)
Refer to Testing rules.