We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add `update_env` to `process_execution::local`.
Replaces any `{chroot}` placeholders in the requested process env with the sandbox workdir.

This reflects the behavior for interactive processes executed with the `run` goal.

--- 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/core/goals/publish.py`
Content:
```
1 # Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
2 # Licensed under the Apache License, Version 2.0 (see LICENSE).
3 """Goal for publishing packaged targets to any repository or registry etc.
4 
5 Plugins implement the publish protocol that provides this goal with the processes to run in order to
6 publish the artifacts.
7 
8 The publish protocol consists of defining two union members and one rule, returning the processes to
9 run. See the doc for the corresponding classses in this module for details on the classes to define.
10 
11 Example rule:
12 
13     @rule
14     async def publish_example(request: PublishToMyRepoRequest, ...) -> PublishProcesses:
15       # Create `InteractiveProcess` instances as required by the `request`.
16       return PublishProcesses(...)
17 """
18 
19 
20 from __future__ import annotations
21 
22 import collections
23 import json
24 import logging
25 from abc import ABCMeta
26 from dataclasses import asdict, dataclass, field, is_dataclass, replace
27 from itertools import chain
28 from typing import ClassVar, Generic, Type, TypeVar
29 
30 from typing_extensions import final
31 
32 from pants.core.goals.package import BuiltPackage, PackageFieldSet
33 from pants.engine.addresses import Address
34 from pants.engine.collection import Collection
35 from pants.engine.console import Console
36 from pants.engine.goal import Goal, GoalSubsystem
37 from pants.engine.process import InteractiveProcess, InteractiveProcessResult
38 from pants.engine.rules import Effect, Get, MultiGet, collect_rules, goal_rule, rule
39 from pants.engine.target import (
40     FieldSet,
41     ImmutableValue,
42     NoApplicableTargetsBehavior,
43     TargetRootsToFieldSets,
44     TargetRootsToFieldSetsRequest,
45 )
46 from pants.engine.unions import UnionMembership, UnionRule, union
47 from pants.option.option_types import StrOption
48 from pants.util.frozendict import FrozenDict
49 
50 logger = logging.getLogger(__name__)
51 
52 
53 _F = TypeVar("_F", bound=FieldSet)
54 
55 
56 class PublishOutputData(FrozenDict[str, ImmutableValue]):
57     pass
58 
59 
60 @union
61 @dataclass(frozen=True)
62 class PublishRequest(Generic[_F]):
63     """Implement a union member subclass of this union class along with a PublishFieldSet subclass
64     that appoints that member subclass in order to receive publish requests for targets compatible
65     with the field set.
66 
67     The `packages` hold all artifacts produced for a given target to be published.
68 
69     Example:
70 
71         PublishToMyRepoRequest(PublishRequest):
72           pass
73 
74         PublishToMyRepoFieldSet(PublishFieldSet):
75           publish_request_type = PublishToMyRepoRequest
76 
77           # Standard FieldSet semantics from here on:
78           required_fields = (MyRepositories,)
79           ...
80     """
81 
82     field_set: _F
83     packages: tuple[BuiltPackage, ...]
84 
85 
86 _T = TypeVar("_T", bound=PublishRequest)
87 
88 
89 @union
90 @dataclass(frozen=True)
91 class PublishFieldSet(Generic[_T], FieldSet, metaclass=ABCMeta):
92     """FieldSet for PublishRequest.
93 
94     Union members may list any fields required to fulfill the instantiation of the
95     `PublishProcesses` result of the publish rule.
96     """
97 
98     # Subclasses must provide this, to a union member (subclass) of `PublishRequest`.
99     publish_request_type: ClassVar[Type[_T]]
100 
101     @final
102     def _request(self, packages: tuple[BuiltPackage, ...]) -> _T:
103         """Internal helper for the core publish goal."""
104         return self.publish_request_type(field_set=self, packages=packages)
105 
106     @final
107     @classmethod
108     def rules(cls) -> tuple[UnionRule, ...]:
109         """Helper method for registering the union members."""
110         return (
111             UnionRule(PublishFieldSet, cls),
112             UnionRule(PublishRequest, cls.publish_request_type),
113         )
114 
115     def get_output_data(self) -> PublishOutputData:
116         return PublishOutputData({"target": self.address})
117 
118 
119 @dataclass(frozen=True)
120 class PublishPackages:
121     """Processes to run in order to publish the named artifacts.
122 
123     The `names` should list all artifacts being published by the `process` command.
124 
125     The `process` may be `None`, indicating that it will not be published. This will be logged as
126     `skipped`. If the process returns a non zero exit code, it will be logged as `failed`.
127 
128     The `description` may be a reason explaining why the publish was skipped, or identifying which
129     repository the artifacts are published to.
130     """
131 
132     names: tuple[str, ...]
133     process: InteractiveProcess | None = None
134     description: str | None = None
135     data: PublishOutputData = field(default_factory=PublishOutputData)
136 
137     def get_output_data(self, **extra_data) -> PublishOutputData:
138         return PublishOutputData(
139             {
140                 "names": self.names,
141                 **self.data,
142                 **extra_data,
143             }
144         )
145 
146 
147 class PublishProcesses(Collection[PublishPackages]):
148     """Collection of what processes to run for all built packages.
149 
150     This is returned from implementing rules in response to a PublishRequest.
151 
152     Depending on the capabilities of the publishing tool, the work may be partitioned based on
153     number of artifacts and/or repositories to publish to.
154     """
155 
156 
157 @dataclass(frozen=True)
158 class PublishProcessesRequest:
159     """Internal request taking all field sets for a target and turning it into a `PublishProcesses`
160     collection (via registered publish plugins)."""
161 
162     package_field_sets: tuple[PackageFieldSet, ...]
163     publish_field_sets: tuple[PublishFieldSet, ...]
164 
165 
166 class PublishSubsystem(GoalSubsystem):
167     name = "publish"
168     help = "Publish deliverables (assets, distributions, images, etc)."
169 
170     @classmethod
171     def activated(cls, union_membership: UnionMembership) -> bool:
172         return PackageFieldSet in union_membership and PublishFieldSet in union_membership
173 
174     output = StrOption(
175         "--output",
176         default=None,
177         help="Filename for JSON structured publish information.",
178     )
179 
180 
181 class Publish(Goal):
182     subsystem_cls = PublishSubsystem
183 
184 
185 @goal_rule
186 async def run_publish(console: Console, publish: PublishSubsystem) -> Publish:
187     target_roots_to_package_field_sets, target_roots_to_publish_field_sets = await MultiGet(
188         Get(
189             TargetRootsToFieldSets,
190             TargetRootsToFieldSetsRequest(
191                 PackageFieldSet,
192                 goal_description="",
193                 # Don't warn/error here because it's already covered by `PublishFieldSet`.
194                 no_applicable_targets_behavior=NoApplicableTargetsBehavior.ignore,
195             ),
196         ),
197         Get(
198             TargetRootsToFieldSets,
199             TargetRootsToFieldSetsRequest(
200                 PublishFieldSet,
201                 goal_description="the `publish` goal",
202                 no_applicable_targets_behavior=NoApplicableTargetsBehavior.warn,
203             ),
204         ),
205     )
206 
207     # Only keep field sets that both package something, and have something to publish.
208     targets = set(target_roots_to_package_field_sets.targets).intersection(
209         set(target_roots_to_publish_field_sets.targets)
210     )
211 
212     if not targets:
213         return Publish(exit_code=0)
214 
215     # Build all packages and request the processes to run for each field set.
216     processes = await MultiGet(
217         Get(
218             PublishProcesses,
219             PublishProcessesRequest(
220                 target_roots_to_package_field_sets.mapping[tgt],
221                 target_roots_to_publish_field_sets.mapping[tgt],
222             ),
223         )
224         for tgt in targets
225     )
226 
227     # Run all processes interactively.
228     exit_code: int = 0
229     outputs: list[PublishOutputData] = []
230     results: list[str] = []
231 
232     for pub in chain.from_iterable(processes):
233         if not pub.process:
234             sigil = console.sigil_skipped()
235             status = "skipped"
236             if pub.description:
237                 status += f" {pub.description}"
238             for name in pub.names:
239                 results.append(f"{sigil} {name} {status}.")
240             outputs.append(pub.get_output_data(published=False, status=status))
241             continue
242 
243         res = await Effect(InteractiveProcessResult, InteractiveProcess, pub.process)
244         if res.exit_code == 0:
245             sigil = console.sigil_succeeded()
246             status = "published"
247             prep = "to"
248         else:
249             sigil = console.sigil_failed()
250             status = "failed"
251             prep = "for"
252             exit_code = res.exit_code
253 
254         if pub.description:
255             status += f" {prep} {pub.description}"
256 
257         for name in pub.names:
258             results.append(f"{sigil} {name} {status}.")
259 
260         outputs.append(
261             pub.get_output_data(
262                 exit_code=res.exit_code,
263                 published=res.exit_code == 0,
264                 status=status,
265             )
266         )
267 
268     console.print_stderr("")
269     if not results:
270         sigil = console.sigil_skipped()
271         console.print_stderr(f"{sigil} Nothing published.")
272 
273     # We collect all results to the end, so all output from the interactive processes are done,
274     # before printing the results.
275     for line in results:
276         console.print_stderr(line)
277 
278     # Log structured output
279     output_data = json.dumps(outputs, cls=_PublishJsonEncoder, indent=2, sort_keys=True)
280     logger.debug(f"Publish result data:\n{output_data}")
281     if publish.output:
282         with open(publish.output, mode="w") as fd:
283             fd.write(output_data)
284 
285     return Publish(exit_code)
286 
287 
288 class _PublishJsonEncoder(json.JSONEncoder):
289     safe_to_str_types = (Address,)
290 
291     def default(self, o):
292         """Return a serializable object for o."""
293         if is_dataclass(o):
294             return asdict(o)
295         if isinstance(o, collections.abc.Mapping):
296             return dict(o)
297         if isinstance(o, collections.abc.Sequence):
298             return list(o)
299         try:
300             return super().default(o)
301         except TypeError:
302             return str(o)
303 
304 
305 @rule
306 async def package_for_publish(request: PublishProcessesRequest) -> PublishProcesses:
307     packages = await MultiGet(
308         Get(BuiltPackage, PackageFieldSet, field_set) for field_set in request.package_field_sets
309     )
310 
311     for pkg in packages:
312         for artifact in pkg.artifacts:
313             if artifact.relpath:
314                 logger.info(f"Packaged {artifact.relpath}")
315             elif artifact.extra_log_lines:
316                 logger.info(str(artifact.extra_log_lines[0]))
317 
318     publish = await MultiGet(
319         Get(
320             PublishProcesses,
321             PublishRequest,
322             field_set._request(packages),
323         )
324         for field_set in request.publish_field_sets
325     )
326 
327     # Flatten and dress each publish processes collection with data about its origin.
328     publish_processes = [
329         replace(
330             publish_process,
331             data=PublishOutputData({**publish_process.data, **field_set.get_output_data()}),
332         )
333         for processes, field_set in zip(publish, request.publish_field_sets)
334         for publish_process in processes
335     ]
336 
337     return PublishProcesses(publish_processes)
338 
339 
340 def rules():
341     return collect_rules()
342
```
--- 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/core/goals/publish.py b/src/python/pants/core/goals/publish.py
--- a/src/python/pants/core/goals/publish.py
+++ b/src/python/pants/core/goals/publish.py
@@ -240,6 +240,7 @@
             outputs.append(pub.get_output_data(published=False, status=status))
             continue
 
+        logger.debug(f"Execute {pub.process}")
         res = await Effect(InteractiveProcessResult, InteractiveProcess, pub.process)
         if res.exit_code == 0:
             sigil = console.sigil_succeeded()

{"golden_diff": "diff --git a/src/python/pants/core/goals/publish.py b/src/python/pants/core/goals/publish.py\n--- a/src/python/pants/core/goals/publish.py\n+++ b/src/python/pants/core/goals/publish.py\n@@ -240,6 +240,7 @@\n             outputs.append(pub.get_output_data(published=False, status=status))\n             continue\n \n+        logger.debug(f\"Execute {pub.process}\")\n         res = await Effect(InteractiveProcessResult, InteractiveProcess, pub.process)\n         if res.exit_code == 0:\n             sigil = console.sigil_succeeded()\n", "issue": "Add `update_env` to `process_execution::local`.\nReplaces any `{chroot}` placeholders in the requested process env with the sandbox workdir.\r\n\r\nThis reflects the behavior for interactive processes executed with the `run` goal.\n", "before_files": [{"content": "# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).\n# Licensed under the Apache License, Version 2.0 (see LICENSE).\n\"\"\"Goal for publishing packaged targets to any repository or registry etc.\n\nPlugins implement the publish protocol that provides this goal with the processes to run in order to\npublish the artifacts.\n\nThe publish protocol consists of defining two union members and one rule, returning the processes to\nrun. See the doc for the corresponding classses in this module for details on the classes to define.\n\nExample rule:\n\n    @rule\n    async def publish_example(request: PublishToMyRepoRequest, ...) -> PublishProcesses:\n      # Create `InteractiveProcess` instances as required by the `request`.\n      return PublishProcesses(...)\n\"\"\"\n\n\nfrom __future__ import annotations\n\nimport collections\nimport json\nimport logging\nfrom abc import ABCMeta\nfrom dataclasses import asdict, dataclass, field, is_dataclass, replace\nfrom itertools import chain\nfrom typing import ClassVar, Generic, Type, TypeVar\n\nfrom typing_extensions import final\n\nfrom pants.core.goals.package import BuiltPackage, PackageFieldSet\nfrom pants.engine.addresses import Address\nfrom pants.engine.collection import Collection\nfrom pants.engine.console import Console\nfrom pants.engine.goal import Goal, GoalSubsystem\nfrom pants.engine.process import InteractiveProcess, InteractiveProcessResult\nfrom pants.engine.rules import Effect, Get, MultiGet, collect_rules, goal_rule, rule\nfrom pants.engine.target import (\n    FieldSet,\n    ImmutableValue,\n    NoApplicableTargetsBehavior,\n    TargetRootsToFieldSets,\n    TargetRootsToFieldSetsRequest,\n)\nfrom pants.engine.unions import UnionMembership, UnionRule, union\nfrom pants.option.option_types import StrOption\nfrom pants.util.frozendict import FrozenDict\n\nlogger = logging.getLogger(__name__)\n\n\n_F = TypeVar(\"_F\", bound=FieldSet)\n\n\nclass PublishOutputData(FrozenDict[str, ImmutableValue]):\n    pass\n\n\n@union\n@dataclass(frozen=True)\nclass PublishRequest(Generic[_F]):\n    \"\"\"Implement a union member subclass of this union class along with a PublishFieldSet subclass\n    that appoints that member subclass in order to receive publish requests for targets compatible\n    with the field set.\n\n    The `packages` hold all artifacts produced for a given target to be published.\n\n    Example:\n\n        PublishToMyRepoRequest(PublishRequest):\n          pass\n\n        PublishToMyRepoFieldSet(PublishFieldSet):\n          publish_request_type = PublishToMyRepoRequest\n\n          # Standard FieldSet semantics from here on:\n          required_fields = (MyRepositories,)\n          ...\n    \"\"\"\n\n    field_set: _F\n    packages: tuple[BuiltPackage, ...]\n\n\n_T = TypeVar(\"_T\", bound=PublishRequest)\n\n\n@union\n@dataclass(frozen=True)\nclass PublishFieldSet(Generic[_T], FieldSet, metaclass=ABCMeta):\n    \"\"\"FieldSet for PublishRequest.\n\n    Union members may list any fields required to fulfill the instantiation of the\n    `PublishProcesses` result of the publish rule.\n    \"\"\"\n\n    # Subclasses must provide this, to a union member (subclass) of `PublishRequest`.\n    publish_request_type: ClassVar[Type[_T]]\n\n    @final\n    def _request(self, packages: tuple[BuiltPackage, ...]) -> _T:\n        \"\"\"Internal helper for the core publish goal.\"\"\"\n        return self.publish_request_type(field_set=self, packages=packages)\n\n    @final\n    @classmethod\n    def rules(cls) -> tuple[UnionRule, ...]:\n        \"\"\"Helper method for registering the union members.\"\"\"\n        return (\n            UnionRule(PublishFieldSet, cls),\n            UnionRule(PublishRequest, cls.publish_request_type),\n        )\n\n    def get_output_data(self) -> PublishOutputData:\n        return PublishOutputData({\"target\": self.address})\n\n\n@dataclass(frozen=True)\nclass PublishPackages:\n    \"\"\"Processes to run in order to publish the named artifacts.\n\n    The `names` should list all artifacts being published by the `process` command.\n\n    The `process` may be `None`, indicating that it will not be published. This will be logged as\n    `skipped`. If the process returns a non zero exit code, it will be logged as `failed`.\n\n    The `description` may be a reason explaining why the publish was skipped, or identifying which\n    repository the artifacts are published to.\n    \"\"\"\n\n    names: tuple[str, ...]\n    process: InteractiveProcess | None = None\n    description: str | None = None\n    data: PublishOutputData = field(default_factory=PublishOutputData)\n\n    def get_output_data(self, **extra_data) -> PublishOutputData:\n        return PublishOutputData(\n            {\n                \"names\": self.names,\n                **self.data,\n                **extra_data,\n            }\n        )\n\n\nclass PublishProcesses(Collection[PublishPackages]):\n    \"\"\"Collection of what processes to run for all built packages.\n\n    This is returned from implementing rules in response to a PublishRequest.\n\n    Depending on the capabilities of the publishing tool, the work may be partitioned based on\n    number of artifacts and/or repositories to publish to.\n    \"\"\"\n\n\n@dataclass(frozen=True)\nclass PublishProcessesRequest:\n    \"\"\"Internal request taking all field sets for a target and turning it into a `PublishProcesses`\n    collection (via registered publish plugins).\"\"\"\n\n    package_field_sets: tuple[PackageFieldSet, ...]\n    publish_field_sets: tuple[PublishFieldSet, ...]\n\n\nclass PublishSubsystem(GoalSubsystem):\n    name = \"publish\"\n    help = \"Publish deliverables (assets, distributions, images, etc).\"\n\n    @classmethod\n    def activated(cls, union_membership: UnionMembership) -> bool:\n        return PackageFieldSet in union_membership and PublishFieldSet in union_membership\n\n    output = StrOption(\n        \"--output\",\n        default=None,\n        help=\"Filename for JSON structured publish information.\",\n    )\n\n\nclass Publish(Goal):\n    subsystem_cls = PublishSubsystem\n\n\n@goal_rule\nasync def run_publish(console: Console, publish: PublishSubsystem) -> Publish:\n    target_roots_to_package_field_sets, target_roots_to_publish_field_sets = await MultiGet(\n        Get(\n            TargetRootsToFieldSets,\n            TargetRootsToFieldSetsRequest(\n                PackageFieldSet,\n                goal_description=\"\",\n                # Don't warn/error here because it's already covered by `PublishFieldSet`.\n                no_applicable_targets_behavior=NoApplicableTargetsBehavior.ignore,\n            ),\n        ),\n        Get(\n            TargetRootsToFieldSets,\n            TargetRootsToFieldSetsRequest(\n                PublishFieldSet,\n                goal_description=\"the `publish` goal\",\n                no_applicable_targets_behavior=NoApplicableTargetsBehavior.warn,\n            ),\n        ),\n    )\n\n    # Only keep field sets that both package something, and have something to publish.\n    targets = set(target_roots_to_package_field_sets.targets).intersection(\n        set(target_roots_to_publish_field_sets.targets)\n    )\n\n    if not targets:\n        return Publish(exit_code=0)\n\n    # Build all packages and request the processes to run for each field set.\n    processes = await MultiGet(\n        Get(\n            PublishProcesses,\n            PublishProcessesRequest(\n                target_roots_to_package_field_sets.mapping[tgt],\n                target_roots_to_publish_field_sets.mapping[tgt],\n            ),\n        )\n        for tgt in targets\n    )\n\n    # Run all processes interactively.\n    exit_code: int = 0\n    outputs: list[PublishOutputData] = []\n    results: list[str] = []\n\n    for pub in chain.from_iterable(processes):\n        if not pub.process:\n            sigil = console.sigil_skipped()\n            status = \"skipped\"\n            if pub.description:\n                status += f\" {pub.description}\"\n            for name in pub.names:\n                results.append(f\"{sigil} {name} {status}.\")\n            outputs.append(pub.get_output_data(published=False, status=status))\n            continue\n\n        res = await Effect(InteractiveProcessResult, InteractiveProcess, pub.process)\n        if res.exit_code == 0:\n            sigil = console.sigil_succeeded()\n            status = \"published\"\n            prep = \"to\"\n        else:\n            sigil = console.sigil_failed()\n            status = \"failed\"\n            prep = \"for\"\n            exit_code = res.exit_code\n\n        if pub.description:\n            status += f\" {prep} {pub.description}\"\n\n        for name in pub.names:\n            results.append(f\"{sigil} {name} {status}.\")\n\n        outputs.append(\n            pub.get_output_data(\n                exit_code=res.exit_code,\n                published=res.exit_code == 0,\n                status=status,\n            )\n        )\n\n    console.print_stderr(\"\")\n    if not results:\n        sigil = console.sigil_skipped()\n        console.print_stderr(f\"{sigil} Nothing published.\")\n\n    # We collect all results to the end, so all output from the interactive processes are done,\n    # before printing the results.\n    for line in results:\n        console.print_stderr(line)\n\n    # Log structured output\n    output_data = json.dumps(outputs, cls=_PublishJsonEncoder, indent=2, sort_keys=True)\n    logger.debug(f\"Publish result data:\\n{output_data}\")\n    if publish.output:\n        with open(publish.output, mode=\"w\") as fd:\n            fd.write(output_data)\n\n    return Publish(exit_code)\n\n\nclass _PublishJsonEncoder(json.JSONEncoder):\n    safe_to_str_types = (Address,)\n\n    def default(self, o):\n        \"\"\"Return a serializable object for o.\"\"\"\n        if is_dataclass(o):\n            return asdict(o)\n        if isinstance(o, collections.abc.Mapping):\n            return dict(o)\n        if isinstance(o, collections.abc.Sequence):\n            return list(o)\n        try:\n            return super().default(o)\n        except TypeError:\n            return str(o)\n\n\n@rule\nasync def package_for_publish(request: PublishProcessesRequest) -> PublishProcesses:\n    packages = await MultiGet(\n        Get(BuiltPackage, PackageFieldSet, field_set) for field_set in request.package_field_sets\n    )\n\n    for pkg in packages:\n        for artifact in pkg.artifacts:\n            if artifact.relpath:\n                logger.info(f\"Packaged {artifact.relpath}\")\n            elif artifact.extra_log_lines:\n                logger.info(str(artifact.extra_log_lines[0]))\n\n    publish = await MultiGet(\n        Get(\n            PublishProcesses,\n            PublishRequest,\n            field_set._request(packages),\n        )\n        for field_set in request.publish_field_sets\n    )\n\n    # Flatten and dress each publish processes collection with data about its origin.\n    publish_processes = [\n        replace(\n            publish_process,\n            data=PublishOutputData({**publish_process.data, **field_set.get_output_data()}),\n        )\n        for processes, field_set in zip(publish, request.publish_field_sets)\n        for publish_process in processes\n    ]\n\n    return PublishProcesses(publish_processes)\n\n\ndef rules():\n    return collect_rules()\n", "path": "src/python/pants/core/goals/publish.py"}], "after_files": [{"content": "# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).\n# Licensed under the Apache License, Version 2.0 (see LICENSE).\n\"\"\"Goal for publishing packaged targets to any repository or registry etc.\n\nPlugins implement the publish protocol that provides this goal with the processes to run in order to\npublish the artifacts.\n\nThe publish protocol consists of defining two union members and one rule, returning the processes to\nrun. See the doc for the corresponding classses in this module for details on the classes to define.\n\nExample rule:\n\n    @rule\n    async def publish_example(request: PublishToMyRepoRequest, ...) -> PublishProcesses:\n      # Create `InteractiveProcess` instances as required by the `request`.\n      return PublishProcesses(...)\n\"\"\"\n\n\nfrom __future__ import annotations\n\nimport collections\nimport json\nimport logging\nfrom abc import ABCMeta\nfrom dataclasses import asdict, dataclass, field, is_dataclass, replace\nfrom itertools import chain\nfrom typing import ClassVar, Generic, Type, TypeVar\n\nfrom typing_extensions import final\n\nfrom pants.core.goals.package import BuiltPackage, PackageFieldSet\nfrom pants.engine.addresses import Address\nfrom pants.engine.collection import Collection\nfrom pants.engine.console import Console\nfrom pants.engine.goal import Goal, GoalSubsystem\nfrom pants.engine.process import InteractiveProcess, InteractiveProcessResult\nfrom pants.engine.rules import Effect, Get, MultiGet, collect_rules, goal_rule, rule\nfrom pants.engine.target import (\n    FieldSet,\n    ImmutableValue,\n    NoApplicableTargetsBehavior,\n    TargetRootsToFieldSets,\n    TargetRootsToFieldSetsRequest,\n)\nfrom pants.engine.unions import UnionMembership, UnionRule, union\nfrom pants.option.option_types import StrOption\nfrom pants.util.frozendict import FrozenDict\n\nlogger = logging.getLogger(__name__)\n\n\n_F = TypeVar(\"_F\", bound=FieldSet)\n\n\nclass PublishOutputData(FrozenDict[str, ImmutableValue]):\n    pass\n\n\n@union\n@dataclass(frozen=True)\nclass PublishRequest(Generic[_F]):\n    \"\"\"Implement a union member subclass of this union class along with a PublishFieldSet subclass\n    that appoints that member subclass in order to receive publish requests for targets compatible\n    with the field set.\n\n    The `packages` hold all artifacts produced for a given target to be published.\n\n    Example:\n\n        PublishToMyRepoRequest(PublishRequest):\n          pass\n\n        PublishToMyRepoFieldSet(PublishFieldSet):\n          publish_request_type = PublishToMyRepoRequest\n\n          # Standard FieldSet semantics from here on:\n          required_fields = (MyRepositories,)\n          ...\n    \"\"\"\n\n    field_set: _F\n    packages: tuple[BuiltPackage, ...]\n\n\n_T = TypeVar(\"_T\", bound=PublishRequest)\n\n\n@union\n@dataclass(frozen=True)\nclass PublishFieldSet(Generic[_T], FieldSet, metaclass=ABCMeta):\n    \"\"\"FieldSet for PublishRequest.\n\n    Union members may list any fields required to fulfill the instantiation of the\n    `PublishProcesses` result of the publish rule.\n    \"\"\"\n\n    # Subclasses must provide this, to a union member (subclass) of `PublishRequest`.\n    publish_request_type: ClassVar[Type[_T]]\n\n    @final\n    def _request(self, packages: tuple[BuiltPackage, ...]) -> _T:\n        \"\"\"Internal helper for the core publish goal.\"\"\"\n        return self.publish_request_type(field_set=self, packages=packages)\n\n    @final\n    @classmethod\n    def rules(cls) -> tuple[UnionRule, ...]:\n        \"\"\"Helper method for registering the union members.\"\"\"\n        return (\n            UnionRule(PublishFieldSet, cls),\n            UnionRule(PublishRequest, cls.publish_request_type),\n        )\n\n    def get_output_data(self) -> PublishOutputData:\n        return PublishOutputData({\"target\": self.address})\n\n\n@dataclass(frozen=True)\nclass PublishPackages:\n    \"\"\"Processes to run in order to publish the named artifacts.\n\n    The `names` should list all artifacts being published by the `process` command.\n\n    The `process` may be `None`, indicating that it will not be published. This will be logged as\n    `skipped`. If the process returns a non zero exit code, it will be logged as `failed`.\n\n    The `description` may be a reason explaining why the publish was skipped, or identifying which\n    repository the artifacts are published to.\n    \"\"\"\n\n    names: tuple[str, ...]\n    process: InteractiveProcess | None = None\n    description: str | None = None\n    data: PublishOutputData = field(default_factory=PublishOutputData)\n\n    def get_output_data(self, **extra_data) -> PublishOutputData:\n        return PublishOutputData(\n            {\n                \"names\": self.names,\n                **self.data,\n                **extra_data,\n            }\n        )\n\n\nclass PublishProcesses(Collection[PublishPackages]):\n    \"\"\"Collection of what processes to run for all built packages.\n\n    This is returned from implementing rules in response to a PublishRequest.\n\n    Depending on the capabilities of the publishing tool, the work may be partitioned based on\n    number of artifacts and/or repositories to publish to.\n    \"\"\"\n\n\n@dataclass(frozen=True)\nclass PublishProcessesRequest:\n    \"\"\"Internal request taking all field sets for a target and turning it into a `PublishProcesses`\n    collection (via registered publish plugins).\"\"\"\n\n    package_field_sets: tuple[PackageFieldSet, ...]\n    publish_field_sets: tuple[PublishFieldSet, ...]\n\n\nclass PublishSubsystem(GoalSubsystem):\n    name = \"publish\"\n    help = \"Publish deliverables (assets, distributions, images, etc).\"\n\n    @classmethod\n    def activated(cls, union_membership: UnionMembership) -> bool:\n        return PackageFieldSet in union_membership and PublishFieldSet in union_membership\n\n    output = StrOption(\n        \"--output\",\n        default=None,\n        help=\"Filename for JSON structured publish information.\",\n    )\n\n\nclass Publish(Goal):\n    subsystem_cls = PublishSubsystem\n\n\n@goal_rule\nasync def run_publish(console: Console, publish: PublishSubsystem) -> Publish:\n    target_roots_to_package_field_sets, target_roots_to_publish_field_sets = await MultiGet(\n        Get(\n            TargetRootsToFieldSets,\n            TargetRootsToFieldSetsRequest(\n                PackageFieldSet,\n                goal_description=\"\",\n                # Don't warn/error here because it's already covered by `PublishFieldSet`.\n                no_applicable_targets_behavior=NoApplicableTargetsBehavior.ignore,\n            ),\n        ),\n        Get(\n            TargetRootsToFieldSets,\n            TargetRootsToFieldSetsRequest(\n                PublishFieldSet,\n                goal_description=\"the `publish` goal\",\n                no_applicable_targets_behavior=NoApplicableTargetsBehavior.warn,\n            ),\n        ),\n    )\n\n    # Only keep field sets that both package something, and have something to publish.\n    targets = set(target_roots_to_package_field_sets.targets).intersection(\n        set(target_roots_to_publish_field_sets.targets)\n    )\n\n    if not targets:\n        return Publish(exit_code=0)\n\n    # Build all packages and request the processes to run for each field set.\n    processes = await MultiGet(\n        Get(\n            PublishProcesses,\n            PublishProcessesRequest(\n                target_roots_to_package_field_sets.mapping[tgt],\n                target_roots_to_publish_field_sets.mapping[tgt],\n            ),\n        )\n        for tgt in targets\n    )\n\n    # Run all processes interactively.\n    exit_code: int = 0\n    outputs: list[PublishOutputData] = []\n    results: list[str] = []\n\n    for pub in chain.from_iterable(processes):\n        if not pub.process:\n            sigil = console.sigil_skipped()\n            status = \"skipped\"\n            if pub.description:\n                status += f\" {pub.description}\"\n            for name in pub.names:\n                results.append(f\"{sigil} {name} {status}.\")\n            outputs.append(pub.get_output_data(published=False, status=status))\n            continue\n\n        logger.debug(f\"Execute {pub.process}\")\n        res = await Effect(InteractiveProcessResult, InteractiveProcess, pub.process)\n        if res.exit_code == 0:\n            sigil = console.sigil_succeeded()\n            status = \"published\"\n            prep = \"to\"\n        else:\n            sigil = console.sigil_failed()\n            status = \"failed\"\n            prep = \"for\"\n            exit_code = res.exit_code\n\n        if pub.description:\n            status += f\" {prep} {pub.description}\"\n\n        for name in pub.names:\n            results.append(f\"{sigil} {name} {status}.\")\n\n        outputs.append(\n            pub.get_output_data(\n                exit_code=res.exit_code,\n                published=res.exit_code == 0,\n                status=status,\n            )\n        )\n\n    console.print_stderr(\"\")\n    if not results:\n        sigil = console.sigil_skipped()\n        console.print_stderr(f\"{sigil} Nothing published.\")\n\n    # We collect all results to the end, so all output from the interactive processes are done,\n    # before printing the results.\n    for line in results:\n        console.print_stderr(line)\n\n    # Log structured output\n    output_data = json.dumps(outputs, cls=_PublishJsonEncoder, indent=2, sort_keys=True)\n    logger.debug(f\"Publish result data:\\n{output_data}\")\n    if publish.output:\n        with open(publish.output, mode=\"w\") as fd:\n            fd.write(output_data)\n\n    return Publish(exit_code)\n\n\nclass _PublishJsonEncoder(json.JSONEncoder):\n    safe_to_str_types = (Address,)\n\n    def default(self, o):\n        \"\"\"Return a serializable object for o.\"\"\"\n        if is_dataclass(o):\n            return asdict(o)\n        if isinstance(o, collections.abc.Mapping):\n            return dict(o)\n        if isinstance(o, collections.abc.Sequence):\n            return list(o)\n        try:\n            return super().default(o)\n        except TypeError:\n            return str(o)\n\n\n@rule\nasync def package_for_publish(request: PublishProcessesRequest) -> PublishProcesses:\n    packages = await MultiGet(\n        Get(BuiltPackage, PackageFieldSet, field_set) for field_set in request.package_field_sets\n    )\n\n    for pkg in packages:\n        for artifact in pkg.artifacts:\n            if artifact.relpath:\n                logger.info(f\"Packaged {artifact.relpath}\")\n            elif artifact.extra_log_lines:\n                logger.info(str(artifact.extra_log_lines[0]))\n\n    publish = await MultiGet(\n        Get(\n            PublishProcesses,\n            PublishRequest,\n            field_set._request(packages),\n        )\n        for field_set in request.publish_field_sets\n    )\n\n    # Flatten and dress each publish processes collection with data about its origin.\n    publish_processes = [\n        replace(\n            publish_process,\n            data=PublishOutputData({**publish_process.data, **field_set.get_output_data()}),\n        )\n        for processes, field_set in zip(publish, request.publish_field_sets)\n        for publish_process in processes\n    ]\n\n    return PublishProcesses(publish_processes)\n\n\ndef rules():\n    return collect_rules()\n", "path": "src/python/pants/core/goals/publish.py"}]}