We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Coarsened target calculation for `pylint` pulls in more files than it should on v2.15.0a0
**Describe the bug**

While testing `v2.15.0a0`, I saw that `./pants lint` would consistently freeze / be OOM killed. Through `py-spy` and logging I found that:

* Much time was being spent merging together source digests when setting up `pylint` runs
* The number of source digests in each of ^^^ setups was much larger than expected (tens of thousands of files for an input batch size of a few hundred)
* `pylint` batches of different sizes always ended up with the same number of source digests

While looking through the `pylint` changes in v2.15, I found that coarsened target calculation is currently running in the "partitioner" rule (see [here](https://github.com/pantsbuild/pants/blob/main/src/python/pants/backend/python/lint/pylint/rules.py#L89-L92)). This will result in too many targets being associated with each `pylint` batch, because the partitions returned by that rule are re-batched into smaller chunks according to `[lint].batch_size`, and there's no support in the re-batching logic for subsetting the partition metadata.

We should push the calculation of coarsened targets into the "runner" rule for `pylint`, so we only compute & hydrate the transitive dependencies that are relevant for the specific inputs used in each batch.

**Pants version**

v2.15.0a0

**OS**

Both

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/python/pants/backend/python/lint/pylint/rules.py`
Content:
```
1 # Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2 # Licensed under the Apache License, Version 2.0 (see LICENSE).
3 
4 from __future__ import annotations
5 
6 from dataclasses import dataclass
7 from typing import Tuple
8 
9 from pants.backend.python.lint.pylint.subsystem import (
10     Pylint,
11     PylintFieldSet,
12     PylintFirstPartyPlugins,
13 )
14 from pants.backend.python.subsystems.setup import PythonSetup
15 from pants.backend.python.util_rules import pex_from_targets
16 from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints
17 from pants.backend.python.util_rules.partition import (
18     _partition_by_interpreter_constraints_and_resolve,
19 )
20 from pants.backend.python.util_rules.pex import (
21     Pex,
22     PexRequest,
23     VenvPex,
24     VenvPexProcess,
25     VenvPexRequest,
26 )
27 from pants.backend.python.util_rules.pex_from_targets import RequirementsPexRequest
28 from pants.backend.python.util_rules.python_sources import (
29     PythonSourceFiles,
30     PythonSourceFilesRequest,
31 )
32 from pants.core.goals.lint import REPORT_DIR, LintResult, LintTargetsRequest, Partitions
33 from pants.core.util_rules.config_files import ConfigFiles, ConfigFilesRequest
34 from pants.core.util_rules.partitions import Partition
35 from pants.engine.fs import CreateDigest, Digest, Directory, MergeDigests, RemovePrefix
36 from pants.engine.process import FallibleProcessResult
37 from pants.engine.rules import Get, MultiGet, collect_rules, rule
38 from pants.engine.target import CoarsenedTargets, CoarsenedTargetsRequest
39 from pants.util.logging import LogLevel
40 from pants.util.strutil import pluralize
41 
42 
43 @dataclass(frozen=True)
44 class PartitionMetadata:
45     coarsened_targets: CoarsenedTargets
46     # NB: These are the same across every element in a partition
47     resolve_description: str | None
48     interpreter_constraints: InterpreterConstraints
49 
50     @property
51     def description(self) -> str:
52         ics = str(sorted(str(c) for c in self.interpreter_constraints))
53         return f"{self.resolve_description}, {ics}" if self.resolve_description else ics
54 
55 
56 class PylintRequest(LintTargetsRequest):
57     field_set_type = PylintFieldSet
58     tool_subsystem = Pylint
59 
60 
61 def generate_argv(field_sets: tuple[PylintFieldSet, ...], pylint: Pylint) -> Tuple[str, ...]:
62     args = []
63     if pylint.config is not None:
64         args.append(f"--rcfile={pylint.config}")
65     args.append("--jobs={pants_concurrency}")
66     args.extend(pylint.args)
67     args.extend(field_set.source.file_path for field_set in field_sets)
68     return tuple(args)
69 
70 
71 @rule(desc="Determine if necessary to partition Pylint input", level=LogLevel.DEBUG)
72 async def partition_pylint(
73     request: PylintRequest.PartitionRequest[PylintFieldSet],
74     pylint: Pylint,
75     python_setup: PythonSetup,
76     first_party_plugins: PylintFirstPartyPlugins,
77 ) -> Partitions[PylintFieldSet, PartitionMetadata]:
78     if pylint.skip:
79         return Partitions()
80 
81     first_party_ics = InterpreterConstraints.create_from_compatibility_fields(
82         first_party_plugins.interpreter_constraints_fields, python_setup
83     )
84 
85     resolve_and_interpreter_constraints_to_field_sets = (
86         _partition_by_interpreter_constraints_and_resolve(request.field_sets, python_setup)
87     )
88 
89     coarsened_targets = await Get(
90         CoarsenedTargets,
91         CoarsenedTargetsRequest(field_set.address for field_set in request.field_sets),
92     )
93     coarsened_targets_by_address = coarsened_targets.by_address()
94 
95     return Partitions(
96         Partition(
97             tuple(field_sets),
98             PartitionMetadata(
99                 CoarsenedTargets(
100                     coarsened_targets_by_address[field_set.address] for field_set in field_sets
101                 ),
102                 resolve if len(python_setup.resolves) > 1 else None,
103                 InterpreterConstraints.merge((interpreter_constraints, first_party_ics)),
104             ),
105         )
106         for (
107             resolve,
108             interpreter_constraints,
109         ), field_sets, in resolve_and_interpreter_constraints_to_field_sets.items()
110     )
111 
112 
113 @rule(desc="Lint using Pylint", level=LogLevel.DEBUG)
114 async def run_pylint(
115     request: PylintRequest.Batch[PylintFieldSet, PartitionMetadata],
116     pylint: Pylint,
117     first_party_plugins: PylintFirstPartyPlugins,
118 ) -> LintResult:
119     assert request.partition_metadata is not None
120 
121     requirements_pex_get = Get(
122         Pex,
123         RequirementsPexRequest(
124             (target.address for target in request.partition_metadata.coarsened_targets.closure()),
125             # NB: These constraints must be identical to the other PEXes. Otherwise, we risk using
126             # a different version for the requirements than the other two PEXes, which can result
127             # in a PEX runtime error about missing dependencies.
128             hardcoded_interpreter_constraints=request.partition_metadata.interpreter_constraints,
129         ),
130     )
131 
132     pylint_pex_get = Get(
133         Pex,
134         PexRequest,
135         pylint.to_pex_request(
136             interpreter_constraints=request.partition_metadata.interpreter_constraints,
137             extra_requirements=first_party_plugins.requirement_strings,
138         ),
139     )
140 
141     sources_get = Get(
142         PythonSourceFiles,
143         PythonSourceFilesRequest(request.partition_metadata.coarsened_targets.closure()),
144     )
145     # Ensure that the empty report dir exists.
146     report_directory_digest_get = Get(Digest, CreateDigest([Directory(REPORT_DIR)]))
147 
148     (pylint_pex, requirements_pex, sources, report_directory,) = await MultiGet(
149         pylint_pex_get,
150         requirements_pex_get,
151         sources_get,
152         report_directory_digest_get,
153     )
154 
155     pylint_runner_pex, config_files = await MultiGet(
156         Get(
157             VenvPex,
158             VenvPexRequest(
159                 PexRequest(
160                     output_filename="pylint_runner.pex",
161                     interpreter_constraints=request.partition_metadata.interpreter_constraints,
162                     main=pylint.main,
163                     internal_only=True,
164                     pex_path=[pylint_pex, requirements_pex],
165                 ),
166                 # TODO(John Sirois): Remove this (change to the default of symlinks) when we can
167                 #  upgrade to a version of Pylint with https://github.com/PyCQA/pylint/issues/1470
168                 #  resolved.
169                 site_packages_copies=True,
170             ),
171         ),
172         Get(
173             ConfigFiles,
174             ConfigFilesRequest,
175             pylint.config_request(sources.source_files.snapshot.dirs),
176         ),
177     )
178 
179     pythonpath = list(sources.source_roots)
180     if first_party_plugins:
181         pythonpath.append(first_party_plugins.PREFIX)
182 
183     input_digest = await Get(
184         Digest,
185         MergeDigests(
186             (
187                 config_files.snapshot.digest,
188                 first_party_plugins.sources_digest,
189                 sources.source_files.snapshot.digest,
190                 report_directory,
191             )
192         ),
193     )
194 
195     result = await Get(
196         FallibleProcessResult,
197         VenvPexProcess(
198             pylint_runner_pex,
199             argv=generate_argv(request.elements, pylint),
200             input_digest=input_digest,
201             output_directories=(REPORT_DIR,),
202             extra_env={"PEX_EXTRA_SYS_PATH": ":".join(pythonpath)},
203             concurrency_available=len(request.elements),
204             description=f"Run Pylint on {pluralize(len(request.elements), 'target')}.",
205             level=LogLevel.DEBUG,
206         ),
207     )
208     report = await Get(Digest, RemovePrefix(result.output_digest, REPORT_DIR))
209     return LintResult.create(request, result, report=report)
210 
211 
212 def rules():
213     return [
214         *collect_rules(),
215         *PylintRequest.rules(),
216         *pex_from_targets.rules(),
217     ]
218
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/python/pants/backend/python/lint/pylint/rules.py b/src/python/pants/backend/python/lint/pylint/rules.py
--- a/src/python/pants/backend/python/lint/pylint/rules.py
+++ b/src/python/pants/backend/python/lint/pylint/rules.py
@@ -118,10 +118,20 @@
 ) -> LintResult:
     assert request.partition_metadata is not None
 
+    # The coarsened targets in the incoming request are for all targets in the request's original
+    # partition. Since the core `lint` logic re-batches inputs according to `[lint].batch_size`,
+    # this could be many more targets than are actually needed to lint the specific batch of files
+    # received by this rule. Subset the CTs one more time here to only those that are relevant.
+    all_coarsened_targets_by_address = request.partition_metadata.coarsened_targets.by_address()
+    coarsened_targets = CoarsenedTargets(
+        all_coarsened_targets_by_address[field_set.address] for field_set in request.elements
+    )
+    coarsened_closure = tuple(coarsened_targets.closure())
+
     requirements_pex_get = Get(
         Pex,
         RequirementsPexRequest(
-            (target.address for target in request.partition_metadata.coarsened_targets.closure()),
+            (target.address for target in coarsened_closure),
             # NB: These constraints must be identical to the other PEXes. Otherwise, we risk using
             # a different version for the requirements than the other two PEXes, which can result
             # in a PEX runtime error about missing dependencies.
@@ -140,7 +150,7 @@
 
     sources_get = Get(
         PythonSourceFiles,
-        PythonSourceFilesRequest(request.partition_metadata.coarsened_targets.closure()),
+        PythonSourceFilesRequest(coarsened_closure),
     )
     # Ensure that the empty report dir exists.
     report_directory_digest_get = Get(Digest, CreateDigest([Directory(REPORT_DIR)]))

{"golden_diff": "diff --git a/src/python/pants/backend/python/lint/pylint/rules.py b/src/python/pants/backend/python/lint/pylint/rules.py\n--- a/src/python/pants/backend/python/lint/pylint/rules.py\n+++ b/src/python/pants/backend/python/lint/pylint/rules.py\n@@ -118,10 +118,20 @@\n ) -> LintResult:\n     assert request.partition_metadata is not None\n \n+    # The coarsened targets in the incoming request are for all targets in the request's original\n+    # partition. Since the core `lint` logic re-batches inputs according to `[lint].batch_size`,\n+    # this could be many more targets than are actually needed to lint the specific batch of files\n+    # received by this rule. Subset the CTs one more time here to only those that are relevant.\n+    all_coarsened_targets_by_address = request.partition_metadata.coarsened_targets.by_address()\n+    coarsened_targets = CoarsenedTargets(\n+        all_coarsened_targets_by_address[field_set.address] for field_set in request.elements\n+    )\n+    coarsened_closure = tuple(coarsened_targets.closure())\n+\n     requirements_pex_get = Get(\n         Pex,\n         RequirementsPexRequest(\n-            (target.address for target in request.partition_metadata.coarsened_targets.closure()),\n+            (target.address for target in coarsened_closure),\n             # NB: These constraints must be identical to the other PEXes. Otherwise, we risk using\n             # a different version for the requirements than the other two PEXes, which can result\n             # in a PEX runtime error about missing dependencies.\n@@ -140,7 +150,7 @@\n \n     sources_get = Get(\n         PythonSourceFiles,\n-        PythonSourceFilesRequest(request.partition_metadata.coarsened_targets.closure()),\n+        PythonSourceFilesRequest(coarsened_closure),\n     )\n     # Ensure that the empty report dir exists.\n     report_directory_digest_get = Get(Digest, CreateDigest([Directory(REPORT_DIR)]))\n", "issue": "Coarsened target calculation for `pylint` pulls in more files than it should on v2.15.0a0\n**Describe the bug**\r\n\r\nWhile testing `v2.15.0a0`, I saw that `./pants lint` would consistently freeze / be OOM killed. Through `py-spy` and logging I found that:\r\n\r\n* Much time was being spent merging together source digests when setting up `pylint` runs\r\n* The number of source digests in each of ^^^ setups was much larger than expected (tens of thousands of files for an input batch size of a few hundred)\r\n* `pylint` batches of different sizes always ended up with the same number of source digests\r\n\r\nWhile looking through the `pylint` changes in v2.15, I found that coarsened target calculation is currently running in the \"partitioner\" rule (see [here](https://github.com/pantsbuild/pants/blob/main/src/python/pants/backend/python/lint/pylint/rules.py#L89-L92)). This will result in too many targets being associated with each `pylint` batch, because the partitions returned by that rule are re-batched into smaller chunks according to `[lint].batch_size`, and there's no support in the re-batching logic for subsetting the partition metadata.\r\n\r\nWe should push the calculation of coarsened targets into the \"runner\" rule for `pylint`, so we only compute & hydrate the transitive dependencies that are relevant for the specific inputs used in each batch.\r\n\r\n**Pants version**\r\n\r\nv2.15.0a0\r\n\r\n**OS**\r\n\r\nBoth\n", "before_files": [{"content": "# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).\n# Licensed under the Apache License, Version 2.0 (see LICENSE).\n\nfrom __future__ import annotations\n\nfrom dataclasses import dataclass\nfrom typing import Tuple\n\nfrom pants.backend.python.lint.pylint.subsystem import (\n    Pylint,\n    PylintFieldSet,\n    PylintFirstPartyPlugins,\n)\nfrom pants.backend.python.subsystems.setup import PythonSetup\nfrom pants.backend.python.util_rules import pex_from_targets\nfrom pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints\nfrom pants.backend.python.util_rules.partition import (\n    _partition_by_interpreter_constraints_and_resolve,\n)\nfrom pants.backend.python.util_rules.pex import (\n    Pex,\n    PexRequest,\n    VenvPex,\n    VenvPexProcess,\n    VenvPexRequest,\n)\nfrom pants.backend.python.util_rules.pex_from_targets import RequirementsPexRequest\nfrom pants.backend.python.util_rules.python_sources import (\n    PythonSourceFiles,\n    PythonSourceFilesRequest,\n)\nfrom pants.core.goals.lint import REPORT_DIR, LintResult, LintTargetsRequest, Partitions\nfrom pants.core.util_rules.config_files import ConfigFiles, ConfigFilesRequest\nfrom pants.core.util_rules.partitions import Partition\nfrom pants.engine.fs import CreateDigest, Digest, Directory, MergeDigests, RemovePrefix\nfrom pants.engine.process import FallibleProcessResult\nfrom pants.engine.rules import Get, MultiGet, collect_rules, rule\nfrom pants.engine.target import CoarsenedTargets, CoarsenedTargetsRequest\nfrom pants.util.logging import LogLevel\nfrom pants.util.strutil import pluralize\n\n\n@dataclass(frozen=True)\nclass PartitionMetadata:\n    coarsened_targets: CoarsenedTargets\n    # NB: These are the same across every element in a partition\n    resolve_description: str | None\n    interpreter_constraints: InterpreterConstraints\n\n    @property\n    def description(self) -> str:\n        ics = str(sorted(str(c) for c in self.interpreter_constraints))\n        return f\"{self.resolve_description}, {ics}\" if self.resolve_description else ics\n\n\nclass PylintRequest(LintTargetsRequest):\n    field_set_type = PylintFieldSet\n    tool_subsystem = Pylint\n\n\ndef generate_argv(field_sets: tuple[PylintFieldSet, ...], pylint: Pylint) -> Tuple[str, ...]:\n    args = []\n    if pylint.config is not None:\n        args.append(f\"--rcfile={pylint.config}\")\n    args.append(\"--jobs={pants_concurrency}\")\n    args.extend(pylint.args)\n    args.extend(field_set.source.file_path for field_set in field_sets)\n    return tuple(args)\n\n\n@rule(desc=\"Determine if necessary to partition Pylint input\", level=LogLevel.DEBUG)\nasync def partition_pylint(\n    request: PylintRequest.PartitionRequest[PylintFieldSet],\n    pylint: Pylint,\n    python_setup: PythonSetup,\n    first_party_plugins: PylintFirstPartyPlugins,\n) -> Partitions[PylintFieldSet, PartitionMetadata]:\n    if pylint.skip:\n        return Partitions()\n\n    first_party_ics = InterpreterConstraints.create_from_compatibility_fields(\n        first_party_plugins.interpreter_constraints_fields, python_setup\n    )\n\n    resolve_and_interpreter_constraints_to_field_sets = (\n        _partition_by_interpreter_constraints_and_resolve(request.field_sets, python_setup)\n    )\n\n    coarsened_targets = await Get(\n        CoarsenedTargets,\n        CoarsenedTargetsRequest(field_set.address for field_set in request.field_sets),\n    )\n    coarsened_targets_by_address = coarsened_targets.by_address()\n\n    return Partitions(\n        Partition(\n            tuple(field_sets),\n            PartitionMetadata(\n                CoarsenedTargets(\n                    coarsened_targets_by_address[field_set.address] for field_set in field_sets\n                ),\n                resolve if len(python_setup.resolves) > 1 else None,\n                InterpreterConstraints.merge((interpreter_constraints, first_party_ics)),\n            ),\n        )\n        for (\n            resolve,\n            interpreter_constraints,\n        ), field_sets, in resolve_and_interpreter_constraints_to_field_sets.items()\n    )\n\n\n@rule(desc=\"Lint using Pylint\", level=LogLevel.DEBUG)\nasync def run_pylint(\n    request: PylintRequest.Batch[PylintFieldSet, PartitionMetadata],\n    pylint: Pylint,\n    first_party_plugins: PylintFirstPartyPlugins,\n) -> LintResult:\n    assert request.partition_metadata is not None\n\n    requirements_pex_get = Get(\n        Pex,\n        RequirementsPexRequest(\n            (target.address for target in request.partition_metadata.coarsened_targets.closure()),\n            # NB: These constraints must be identical to the other PEXes. Otherwise, we risk using\n            # a different version for the requirements than the other two PEXes, which can result\n            # in a PEX runtime error about missing dependencies.\n            hardcoded_interpreter_constraints=request.partition_metadata.interpreter_constraints,\n        ),\n    )\n\n    pylint_pex_get = Get(\n        Pex,\n        PexRequest,\n        pylint.to_pex_request(\n            interpreter_constraints=request.partition_metadata.interpreter_constraints,\n            extra_requirements=first_party_plugins.requirement_strings,\n        ),\n    )\n\n    sources_get = Get(\n        PythonSourceFiles,\n        PythonSourceFilesRequest(request.partition_metadata.coarsened_targets.closure()),\n    )\n    # Ensure that the empty report dir exists.\n    report_directory_digest_get = Get(Digest, CreateDigest([Directory(REPORT_DIR)]))\n\n    (pylint_pex, requirements_pex, sources, report_directory,) = await MultiGet(\n        pylint_pex_get,\n        requirements_pex_get,\n        sources_get,\n        report_directory_digest_get,\n    )\n\n    pylint_runner_pex, config_files = await MultiGet(\n        Get(\n            VenvPex,\n            VenvPexRequest(\n                PexRequest(\n                    output_filename=\"pylint_runner.pex\",\n                    interpreter_constraints=request.partition_metadata.interpreter_constraints,\n                    main=pylint.main,\n                    internal_only=True,\n                    pex_path=[pylint_pex, requirements_pex],\n                ),\n                # TODO(John Sirois): Remove this (change to the default of symlinks) when we can\n                #  upgrade to a version of Pylint with https://github.com/PyCQA/pylint/issues/1470\n                #  resolved.\n                site_packages_copies=True,\n            ),\n        ),\n        Get(\n            ConfigFiles,\n            ConfigFilesRequest,\n            pylint.config_request(sources.source_files.snapshot.dirs),\n        ),\n    )\n\n    pythonpath = list(sources.source_roots)\n    if first_party_plugins:\n        pythonpath.append(first_party_plugins.PREFIX)\n\n    input_digest = await Get(\n        Digest,\n        MergeDigests(\n            (\n                config_files.snapshot.digest,\n                first_party_plugins.sources_digest,\n                sources.source_files.snapshot.digest,\n                report_directory,\n            )\n        ),\n    )\n\n    result = await Get(\n        FallibleProcessResult,\n        VenvPexProcess(\n            pylint_runner_pex,\n            argv=generate_argv(request.elements, pylint),\n            input_digest=input_digest,\n            output_directories=(REPORT_DIR,),\n            extra_env={\"PEX_EXTRA_SYS_PATH\": \":\".join(pythonpath)},\n            concurrency_available=len(request.elements),\n            description=f\"Run Pylint on {pluralize(len(request.elements), 'target')}.\",\n            level=LogLevel.DEBUG,\n        ),\n    )\n    report = await Get(Digest, RemovePrefix(result.output_digest, REPORT_DIR))\n    return LintResult.create(request, result, report=report)\n\n\ndef rules():\n    return [\n        *collect_rules(),\n        *PylintRequest.rules(),\n        *pex_from_targets.rules(),\n    ]\n", "path": "src/python/pants/backend/python/lint/pylint/rules.py"}], "after_files": [{"content": "# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).\n# Licensed under the Apache License, Version 2.0 (see LICENSE).\n\nfrom __future__ import annotations\n\nfrom dataclasses import dataclass\nfrom typing import Tuple\n\nfrom pants.backend.python.lint.pylint.subsystem import (\n    Pylint,\n    PylintFieldSet,\n    PylintFirstPartyPlugins,\n)\nfrom pants.backend.python.subsystems.setup import PythonSetup\nfrom pants.backend.python.util_rules import pex_from_targets\nfrom pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints\nfrom pants.backend.python.util_rules.partition import (\n    _partition_by_interpreter_constraints_and_resolve,\n)\nfrom pants.backend.python.util_rules.pex import (\n    Pex,\n    PexRequest,\n    VenvPex,\n    VenvPexProcess,\n    VenvPexRequest,\n)\nfrom pants.backend.python.util_rules.pex_from_targets import RequirementsPexRequest\nfrom pants.backend.python.util_rules.python_sources import (\n    PythonSourceFiles,\n    PythonSourceFilesRequest,\n)\nfrom pants.core.goals.lint import REPORT_DIR, LintResult, LintTargetsRequest, Partitions\nfrom pants.core.util_rules.config_files import ConfigFiles, ConfigFilesRequest\nfrom pants.core.util_rules.partitions import Partition\nfrom pants.engine.fs import CreateDigest, Digest, Directory, MergeDigests, RemovePrefix\nfrom pants.engine.process import FallibleProcessResult\nfrom pants.engine.rules import Get, MultiGet, collect_rules, rule\nfrom pants.engine.target import CoarsenedTargets, CoarsenedTargetsRequest\nfrom pants.util.logging import LogLevel\nfrom pants.util.strutil import pluralize\n\n\n@dataclass(frozen=True)\nclass PartitionMetadata:\n    coarsened_targets: CoarsenedTargets\n    # NB: These are the same across every element in a partition\n    resolve_description: str | None\n    interpreter_constraints: InterpreterConstraints\n\n    @property\n    def description(self) -> str:\n        ics = str(sorted(str(c) for c in self.interpreter_constraints))\n        return f\"{self.resolve_description}, {ics}\" if self.resolve_description else ics\n\n\nclass PylintRequest(LintTargetsRequest):\n    field_set_type = PylintFieldSet\n    tool_subsystem = Pylint\n\n\ndef generate_argv(field_sets: tuple[PylintFieldSet, ...], pylint: Pylint) -> Tuple[str, ...]:\n    args = []\n    if pylint.config is not None:\n        args.append(f\"--rcfile={pylint.config}\")\n    args.append(\"--jobs={pants_concurrency}\")\n    args.extend(pylint.args)\n    args.extend(field_set.source.file_path for field_set in field_sets)\n    return tuple(args)\n\n\n@rule(desc=\"Determine if necessary to partition Pylint input\", level=LogLevel.DEBUG)\nasync def partition_pylint(\n    request: PylintRequest.PartitionRequest[PylintFieldSet],\n    pylint: Pylint,\n    python_setup: PythonSetup,\n    first_party_plugins: PylintFirstPartyPlugins,\n) -> Partitions[PylintFieldSet, PartitionMetadata]:\n    if pylint.skip:\n        return Partitions()\n\n    first_party_ics = InterpreterConstraints.create_from_compatibility_fields(\n        first_party_plugins.interpreter_constraints_fields, python_setup\n    )\n\n    resolve_and_interpreter_constraints_to_field_sets = (\n        _partition_by_interpreter_constraints_and_resolve(request.field_sets, python_setup)\n    )\n\n    coarsened_targets = await Get(\n        CoarsenedTargets,\n        CoarsenedTargetsRequest(field_set.address for field_set in request.field_sets),\n    )\n    coarsened_targets_by_address = coarsened_targets.by_address()\n\n    return Partitions(\n        Partition(\n            tuple(field_sets),\n            PartitionMetadata(\n                CoarsenedTargets(\n                    coarsened_targets_by_address[field_set.address] for field_set in field_sets\n                ),\n                resolve if len(python_setup.resolves) > 1 else None,\n                InterpreterConstraints.merge((interpreter_constraints, first_party_ics)),\n            ),\n        )\n        for (\n            resolve,\n            interpreter_constraints,\n        ), field_sets, in resolve_and_interpreter_constraints_to_field_sets.items()\n    )\n\n\n@rule(desc=\"Lint using Pylint\", level=LogLevel.DEBUG)\nasync def run_pylint(\n    request: PylintRequest.Batch[PylintFieldSet, PartitionMetadata],\n    pylint: Pylint,\n    first_party_plugins: PylintFirstPartyPlugins,\n) -> LintResult:\n    assert request.partition_metadata is not None\n\n    # The coarsened targets in the incoming request are for all targets in the request's original\n    # partition. Since the core `lint` logic re-batches inputs according to `[lint].batch_size`,\n    # this could be many more targets than are actually needed to lint the specific batch of files\n    # received by this rule. Subset the CTs one more time here to only those that are relevant.\n    all_coarsened_targets_by_address = request.partition_metadata.coarsened_targets.by_address()\n    coarsened_targets = CoarsenedTargets(\n        all_coarsened_targets_by_address[field_set.address] for field_set in request.elements\n    )\n    coarsened_closure = tuple(coarsened_targets.closure())\n\n    requirements_pex_get = Get(\n        Pex,\n        RequirementsPexRequest(\n            (target.address for target in coarsened_closure),\n            # NB: These constraints must be identical to the other PEXes. Otherwise, we risk using\n            # a different version for the requirements than the other two PEXes, which can result\n            # in a PEX runtime error about missing dependencies.\n            hardcoded_interpreter_constraints=request.partition_metadata.interpreter_constraints,\n        ),\n    )\n\n    pylint_pex_get = Get(\n        Pex,\n        PexRequest,\n        pylint.to_pex_request(\n            interpreter_constraints=request.partition_metadata.interpreter_constraints,\n            extra_requirements=first_party_plugins.requirement_strings,\n        ),\n    )\n\n    sources_get = Get(\n        PythonSourceFiles,\n        PythonSourceFilesRequest(coarsened_closure),\n    )\n    # Ensure that the empty report dir exists.\n    report_directory_digest_get = Get(Digest, CreateDigest([Directory(REPORT_DIR)]))\n\n    (pylint_pex, requirements_pex, sources, report_directory,) = await MultiGet(\n        pylint_pex_get,\n        requirements_pex_get,\n        sources_get,\n        report_directory_digest_get,\n    )\n\n    pylint_runner_pex, config_files = await MultiGet(\n        Get(\n            VenvPex,\n            VenvPexRequest(\n                PexRequest(\n                    output_filename=\"pylint_runner.pex\",\n                    interpreter_constraints=request.partition_metadata.interpreter_constraints,\n                    main=pylint.main,\n                    internal_only=True,\n                    pex_path=[pylint_pex, requirements_pex],\n                ),\n                # TODO(John Sirois): Remove this (change to the default of symlinks) when we can\n                #  upgrade to a version of Pylint with https://github.com/PyCQA/pylint/issues/1470\n                #  resolved.\n                site_packages_copies=True,\n            ),\n        ),\n        Get(\n            ConfigFiles,\n            ConfigFilesRequest,\n            pylint.config_request(sources.source_files.snapshot.dirs),\n        ),\n    )\n\n    pythonpath = list(sources.source_roots)\n    if first_party_plugins:\n        pythonpath.append(first_party_plugins.PREFIX)\n\n    input_digest = await Get(\n        Digest,\n        MergeDigests(\n            (\n                config_files.snapshot.digest,\n                first_party_plugins.sources_digest,\n                sources.source_files.snapshot.digest,\n                report_directory,\n            )\n        ),\n    )\n\n    result = await Get(\n        FallibleProcessResult,\n        VenvPexProcess(\n            pylint_runner_pex,\n            argv=generate_argv(request.elements, pylint),\n            input_digest=input_digest,\n            output_directories=(REPORT_DIR,),\n            extra_env={\"PEX_EXTRA_SYS_PATH\": \":\".join(pythonpath)},\n            concurrency_available=len(request.elements),\n            description=f\"Run Pylint on {pluralize(len(request.elements), 'target')}.\",\n            level=LogLevel.DEBUG,\n        ),\n    )\n    report = await Get(Digest, RemovePrefix(result.output_digest, REPORT_DIR))\n    return LintResult.create(request, result, report=report)\n\n\ndef rules():\n    return [\n        *collect_rules(),\n        *PylintRequest.rules(),\n        *pex_from_targets.rules(),\n    ]\n", "path": "src/python/pants/backend/python/lint/pylint/rules.py"}]}