Allow pants export --resolve to take targets

docs/notes/2.33.x.md

@@ -69,6 +69,13 @@ All version of [Ruff](https://docs.astral.sh/ruff/) from [0.15.6](https://github
 
 Fixed a bug where `--sync`-ing a Pex lockfile with legacy metadata headers failed.
 
+The `export` goal now accepts target specs alongside `--resolve`. When targets are provided, the exported virtualenv contains only the third-party requirements transitively needed by those targets instead of the entire resolve. When no targets are provided, the existing full-resolve behavior is unchanged.
+
+```
+# Export only the packages needed by my_service (not the entire resolve)
+pants export --resolve=python-default src/python/my_service:my_service
+```
+
 #### Protobuf
 
 Upgraded the default version of `buf` to v1.69.0 (previously v1.3.0).

src/python/pants/backend/python/goals/export.py

@@ -14,7 +14,12 @@
 
 from pants.backend.python.subsystems.python_tool_base import PythonToolBase
 from pants.backend.python.subsystems.setup import PythonSetup
-from pants.backend.python.target_types import PexLayout, PythonResolveField, PythonSourceField
+from pants.backend.python.target_types import (
+    PexLayout,
+    PythonRequirementsField,
+    PythonResolveField,
+    PythonSourceField,
+)
 from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints
 from pants.backend.python.util_rules.local_dists_pep660 import (
     EditableLocalDistsRequest,
@@ -28,7 +33,12 @@
 )
 from pants.backend.python.util_rules.pex_cli import PexPEX
 from pants.backend.python.util_rules.pex_environment import PexEnvironment, PythonExecutable
-from pants.backend.python.util_rules.pex_requirements import EntireLockfile, Lockfile
+from pants.backend.python.util_rules.pex_requirements import (
+    EntireLockfile,
+    Lockfile,
+    PexRequirements,
+    Resolve,
+)
 from pants.core.goals.export import (
     Export,
     ExportError,
@@ -41,15 +51,21 @@
 from pants.core.goals.resolves import ExportableTool
 from pants.core.util_rules.source_files import SourceFiles
 from pants.core.util_rules.stripped_source_files import strip_source_roots
+from pants.engine.addresses import Addresses
 from pants.engine.engine_aware import EngineAwareParameter
 from pants.engine.fs import CreateDigest, FileContent
-from pants.engine.internals.graph import hydrate_sources
+from pants.engine.internals.graph import hydrate_sources, transitive_targets
 from pants.engine.internals.native_engine import EMPTY_DIGEST, AddPrefix, Digest, MergeDigests
 from pants.engine.internals.selectors import concurrently
 from pants.engine.intrinsics import add_prefix, create_digest, digest_to_snapshot, merge_digests
 from pants.engine.process import Process, ProcessCacheScope, execute_process_or_raise
 from pants.engine.rules import collect_rules, implicitly, rule
-from pants.engine.target import AllTargets, HydrateSourcesRequest, SourcesField
+from pants.engine.target import (
+    AllTargets,
+    HydrateSourcesRequest,
+    SourcesField,
+    TransitiveTargetsRequest,
+)
 from pants.engine.unions import UnionMembership, UnionRule
 from pants.option.option_types import EnumOption, StrListOption
 from pants.util.strutil import path_safe, softwrap
@@ -65,6 +81,7 @@ class ExportVenvsRequest(ExportRequest):
 @dataclass(frozen=True)
 class _ExportVenvForResolveRequest(EngineAwareParameter):
     resolve: str
+    addresses: Addresses = dataclasses.field(default_factory=Addresses)
 
 
 class PythonResolveExportFormat(Enum):
@@ -537,11 +554,42 @@ async def export_virtualenv_for_resolve(
     else:
         editable_local_dists_digest = None
 
+    if request.addresses:
+        transitive_tgts = await transitive_targets(
+            TransitiveTargetsRequest(request.addresses), **implicitly()
+        )
+        wrong_resolve = [
+            tgt.address
+            for tgt in transitive_tgts.roots
+            if tgt.has_field(PythonResolveField)
+            and tgt[PythonResolveField].normalized_value(python_setup) != resolve
+        ]
+        if wrong_resolve:
+            raise ExportError(
+                f"The following targets do not belong to the resolve `{resolve}` "
+                f"and cannot be used to filter its export:\n"
+                + "\n".join(f"  {addr}" for addr in wrong_resolve)
+                + f"\nEnsure all targets belong to the `{resolve}` resolve."
+            )
+        req_strings = PexRequirements.req_strings_from_requirement_fields(
+            tgt[PythonRequirementsField]
+            for tgt in transitive_tgts.closure
+            if tgt.has_field(PythonRequirementsField)
+            and tgt[PythonResolveField].normalized_value(python_setup) == resolve
+        )
+        requirements: PexRequirements | EntireLockfile = PexRequirements(
+            req_strings,
+            from_superset=Resolve(resolve, use_entire_lockfile=False),
+            description_of_origin=f"the export of resolve `{resolve}`",
+        )
+    else:
+        requirements = EntireLockfile(lockfile)
+
     pex_request = PexRequest(
         description=f"Build pex for resolve `{resolve}`",
         output_filename=f"{path_safe(resolve)}.pex",
         internal_only=True,
-        requirements=EntireLockfile(lockfile),
+        requirements=requirements,
         sources=editable_local_dists_digest,
         python=python,
         # Packed layout should lead to the best performance in this use case.
@@ -578,8 +626,11 @@ async def export_virtualenvs(
     request: ExportVenvsRequest,
     export_subsys: ExportSubsystem,
 ) -> ExportResults:
+    addresses = Addresses(tgt.address for tgt in request.targets)
     maybe_venvs = await concurrently(
-        export_virtualenv_for_resolve(_ExportVenvForResolveRequest(resolve), **implicitly())
+        export_virtualenv_for_resolve(
+            _ExportVenvForResolveRequest(resolve, addresses), **implicitly()
+        )
         for resolve in export_subsys.options.resolve
     )
     return ExportResults(mv.result for mv in maybe_venvs if mv.result is not None)

src/python/pants/backend/python/goals/export_test.py

@@ -15,13 +15,15 @@
 from pants.backend.python.macros.python_artifact import PythonArtifact
 from pants.backend.python.target_types import (
     PythonDistribution,
+    PythonRequirementsField,
     PythonRequirementTarget,
     PythonResolveField,
     PythonSourceField,
     PythonSourcesGeneratorTarget,
 )
 from pants.backend.python.util_rules import local_dists_pep660, pex_from_targets
-from pants.base.specs import RawSpecs
+from pants.backend.python.util_rules.pex_requirements import PexRequirements
+from pants.base.specs import AddressLiteralSpec, RawSpecs
 from pants.core.goals.export import ExportResults
 from pants.core.util_rules import distdir
 from pants.engine.fs import CreateDigest
@@ -35,6 +37,8 @@
     SingleSourceField,
     Target,
     Targets,
+    TransitiveTargets,
+    TransitiveTargetsRequest,
 )
 from pants.engine.unions import UnionRule
 from pants.testutil.rule_runner import PYTHON_BOOTSTRAP_ENV, RuleRunner
@@ -61,6 +65,7 @@ def rule_runner() -> RuleRunner:
             *isort_subsystem.rules(),  # add a tool that we can try exporting
             QueryRule(Targets, [RawSpecs]),
             QueryRule(ExportResults, [ExportVenvsRequest]),
+            QueryRule(TransitiveTargets, [TransitiveTargetsRequest]),
         ],
         target_types=[PythonRequirementTarget, PythonSourcesGeneratorTarget, PythonDistribution],
         objects={"python_artifact": PythonArtifact, "parametrize": Parametrize},
@@ -190,6 +195,159 @@ def test_export_tool(rule_runner: RuleRunner) -> None:
     assert "isort" in result.description
 
 
+def test_export_venv_filtered_by_targets(rule_runner: RuleRunner) -> None:
+    """When target specs are provided, only their transitive requirements should be exported.
+
+    Dependency graph:
+      A (python_sources) -> B, C
+      B (python_sources) -> req_x
+      C (python_sources) -> req_z
+      req_x (python_requirement x==1.0) -> req_y
+      req_y (python_requirement y==1.0)
+      req_z (python_requirement z==1.0)
+
+    Expected req_strings:
+      export A  -> {x==1.0, y==1.0, z==1.0}
+      export B  -> {x==1.0, y==1.0}
+      export C  -> {z==1.0}
+
+    The req_strings are verified directly via TransitiveTargets (the same logic the export rule
+    uses). The export plumbing is verified with target D (a standalone source target with no
+    python_requirement deps), so req_strings is empty — this exercises the early-return path in
+    _setup_pex_requirements without requiring a PEX-native JSON lockfile.
+    """
+    vinfo = sys.version_info
+    current_interpreter = f"{vinfo.major}.{vinfo.minor}.{vinfo.micro}"
+    rule_runner.write_files(
+        {
+            "src/a/__init__.py": "",
+            "src/a/BUILD": dedent(
+                """\
+                python_sources(name='A', resolve='r', dependencies=['src/b:B', 'src/c:C'])
+                """
+            ),
+            "src/b/__init__.py": "",
+            "src/b/BUILD": dedent(
+                """\
+                python_sources(name='B', resolve='r', dependencies=[':req_x'])
+                python_requirement(name='req_x', requirements=['x==1.0'], resolve='r', dependencies=[':req_y'])
+                python_requirement(name='req_y', requirements=['y==1.0'], resolve='r')
+                """
+            ),
+            "src/c/__init__.py": "",
+            "src/c/BUILD": dedent(
+                """\
+                python_sources(name='C', resolve='r', dependencies=[':req_z'])
+                python_requirement(name='req_z', requirements=['z==1.0'], resolve='r')
+                """
+            ),
+            # Target D has no python_requirement deps — its req_strings is empty, so PEX
+            # subsetting is not invoked and the plain-text lockfile is accepted.
+            "src/d/__init__.py": "",
+            "src/d/BUILD": "python_sources(name='D', resolve='r')",
+            # Plain-text lockfile works here because the export plumbing assertion uses
+            # target D (empty req_strings → early return before PEX JSON parsing).
+            "lock.txt": "x==1.0\ny==1.0\nz==1.0",
+        }
+    )
+    rule_runner.set_options(
+        [
+            *pants_args_for_python_lockfiles,
+            f"--python-interpreter-constraints=['=={current_interpreter}']",
+            "--python-resolves={'r': 'lock.txt'}",
+            "--export-resolve=r",
+            "--source-root-patterns=['src']",
+        ],
+        env_inherit={"PATH", "PYENV_ROOT"},
+    )
+
+    def get_targets(*specs: str) -> Targets:
+        return rule_runner.request(
+            Targets,
+            [
+                RawSpecs(
+                    address_literals=tuple(
+                        AddressLiteralSpec(path_component=p, target_component=t)
+                        for s in specs
+                        for p, t in [s.rsplit(":", 1)]
+                    ),
+                    description_of_origin="test",
+                )
+            ],
+        )
+
+    def req_strings_for(targets: Targets) -> set[str]:
+        tt = rule_runner.request(
+            TransitiveTargets,
+            [TransitiveTargetsRequest([t.address for t in targets])],
+        )
+        return set(
+            PexRequirements.req_strings_from_requirement_fields(
+                tgt[PythonRequirementsField]
+                for tgt in tt.closure
+                if tgt.has_field(PythonRequirementsField)
+            )
+        )
+
+    targets_A = get_targets("src/a:A")
+    targets_B = get_targets("src/b:B")
+    targets_C = get_targets("src/c:C")
+
+    # Verify transitive-closure req_strings: this is the core filtering logic.
+    assert req_strings_for(targets_A) == {"x==1.0", "y==1.0", "z==1.0"}
+    assert req_strings_for(targets_B) == {"x==1.0", "y==1.0"}
+    assert req_strings_for(targets_C) == {"z==1.0"}
+
+    # Combining B and C covers the full set of requirements — same as what no-target filtering
+    # would export (the union of all transitive deps across the resolve).
+    targets_B_and_C = get_targets("src/b:B", "src/c:C")
+    assert req_strings_for(targets_B_and_C) == {"x==1.0", "y==1.0", "z==1.0"}
+
+    # Verify the export plumbing: target D has no transitive python_requirement deps so
+    # req_strings is empty and the early-return path is taken (no PEX JSON lockfile needed).
+    targets_D = get_targets("src/d:D")
+    assert req_strings_for(targets_D) == set()
+    all_results = rule_runner.request(ExportResults, [ExportVenvsRequest(targets=targets_D)])
+    assert len(all_results) == 1
+    result = all_results[0]
+    assert result.resolve == "r"
+    assert result.reldir == f"python/virtualenvs/r/{current_interpreter}"
+
+
+def test_export_venv_filtered_wrong_resolve(rule_runner: RuleRunner) -> None:
+    """Passing a target from the wrong resolve should raise a clear error."""
+    vinfo = sys.version_info
+    current_interpreter = f"{vinfo.major}.{vinfo.minor}.{vinfo.micro}"
+    rule_runner.write_files(
+        {
+            "src/a/__init__.py": "",
+            "src/a/BUILD": "python_sources(name='A', resolve='r1')",
+            "lock.txt": "",
+        }
+    )
+    rule_runner.set_options(
+        [
+            *pants_args_for_python_lockfiles,
+            f"--python-interpreter-constraints=['=={current_interpreter}']",
+            "--python-resolves={'r1': 'lock.txt', 'r2': 'lock.txt'}",
+            "--export-resolve=r2",
+            "--source-root-patterns=['src']",
+        ],
+        env_inherit={"PATH", "PYENV_ROOT"},
+    )
+    targets = rule_runner.request(
+        Targets,
+        [
+            RawSpecs(
+                address_literals=(AddressLiteralSpec("src/a", "A"),),
+                description_of_origin="test",
+            )
+        ],
+    )
+    with pytest.raises(Exception, match="do not belong to the resolve `r2`"):
+        rule_runner.request(ExportResults, [ExportVenvsRequest(targets=targets)])
+
+
 def test_export_codegen_outputs():
     class CodegenSourcesField(SingleSourceField):
         pass

src/python/pants/core/goals/export.py

@@ -201,9 +201,6 @@ async def export_goal(
 
     if not (export_subsys.resolve or export_subsys.options.bin):
         raise ExportError("Must specify at least one `--resolve` or `--bin` to export")
-    if targets:
-        raise ExportError("The `export` goal does not take target specs.")
-
     requests = tuple(request_type(targets) for request_type in request_types)
     all_results = await concurrently(
         export(**implicitly({request: ExportRequest})) for request in requests