fix issue with optional fields in dependency validator

docs/notes/2.32.x.md

@@ -85,6 +85,8 @@ The `runtime` field of [`aws_python_lambda_layer`](https://www.pantsbuild.org/2.
 
 The `grpc-python-plugin` tool now uses an updated `v1.73.1` plugin built from  <https://github.com/nhurden/protoc-gen-grpc-python-prebuilt]. This also brings `macos_arm64` support.
 
+Fixed a bug in how dependencies are validated which prevented resolve-specific interpreter constraints from taking effect when `[python].default_to_resolve_interpreter_constraints` was enabled. Instead, code would fall back to the global interpreter constraints which is not correct when `default_to_resolve_interpreter_constraints` is in use.
+
 The default version of the [Ruff](https://docs.astral.sh/ruff/) tool has been updated to [0.14.14](https://github.com/astral-sh/ruff/releases/tag/0.14.14).
 
 The version of [Pex](https://github.com/pex-tool/pex) used by the Python backend has been upgraded to [`v2.90.2`](https://github.com/pex-tool/pex/releases/tag/v2.90.2). Of particular note for Pants users:

src/python/pants/backend/python/target_types_test.py

@@ -5,11 +5,13 @@
 
 import logging
 from collections.abc import Iterable
+from dataclasses import dataclass
 from textwrap import dedent
 
 import pytest
 
 from pants.backend.python import target_types_rules
+from pants.backend.python.dependency_inference import rules as python_dependency_inference_rules
 from pants.backend.python.dependency_inference.rules import import_rules
 from pants.backend.python.macros.python_artifact import PythonArtifact
 from pants.backend.python.target_types import (
@@ -21,16 +23,20 @@
     PexEntryPointField,
     PexExecutableField,
     PexScriptField,
+    PythonDependenciesField,
     PythonDistribution,
     PythonRequirementsField,
     PythonRequirementTarget,
+    PythonResolveField,
     PythonSourcesGeneratorTarget,
+    PythonSourceTarget,
     ResolvedPexEntryPoint,
     ResolvePexEntryPointRequest,
     ResolvePythonDistributionEntryPointsRequest,
     normalize_module_mapping,
 )
 from pants.backend.python.target_types_rules import (
+    DependencyValidationFieldSet,
     InferPexBinaryEntryPointDependency,
     InferPythonDistributionDependencies,
     PexBinaryEntryPointDependencyInferenceFieldSet,
@@ -40,15 +46,20 @@
 from pants.backend.python.util_rules import python_sources
 from pants.core.goals.generate_lockfiles import UnrecognizedResolveNamesError
 from pants.core.util_rules.unowned_dependency_behavior import UnownedDependencyError
-from pants.engine.addresses import Address
+from pants.engine.addresses import Address, Addresses
 from pants.engine.internals.graph import _TargetParametrizations, _TargetParametrizationsRequest
+from pants.engine.internals.parametrize import Parametrize
 from pants.engine.internals.scheduler import ExecutionError
 from pants.engine.target import (
+    DependenciesRequest,
+    FieldSet,
     InferredDependencies,
     InvalidFieldException,
     InvalidFieldTypeException,
     InvalidTargetException,
+    StringField,
     Tags,
+    Target,
 )
 from pants.testutil.rule_runner import QueryRule, RuleRunner
 from pants.util.frozendict import FrozenDict
@@ -610,3 +621,134 @@ def gen_pex_binary_tgt(entry_point: str, tags: list[str] | None = None) -> PexBi
         gen_pex_binary_tgt("subdir.f.py", tags=["overridden"]),
         gen_pex_binary_tgt("subdir.f:main"),
     }
+
+
+def test_python_dependency_validation_with_parametrized_resolve_repro() -> None:
+    rule_runner = RuleRunner(
+        rules=[
+            *target_types_rules.rules(),
+            *python_sources.rules(),
+            *python_dependency_inference_rules.rules(),
+            QueryRule(_TargetParametrizations, [_TargetParametrizationsRequest]),
+            QueryRule(Addresses, [DependenciesRequest]),
+        ],
+        target_types=[PythonSourceTarget, PythonSourcesGeneratorTarget],
+        objects={"parametrize": Parametrize},
+    )
+    rule_runner.write_files(
+        {
+            "pants.toml": dedent(
+                """\
+                [python]
+                enable_resolves = true
+                interpreter_constraints = ["==3.11.*", "==3.14.*"]
+                interpreter_versions_universe = ["3.11", "3.14"]
+                default_to_resolve_interpreter_constraints = true
+
+                [python.resolves]
+                "a" = "a.lock"
+                "b" = "b.lock"
+
+                [python.resolves_to_interpreter_constraints]
+                "a" = ["==3.11.*"]
+                "b" = ["==3.14.*"]
+
+                [python-infer]
+                imports = true
+
+                [source]
+                root_patterns = ["/src/python"]
+                """
+            ),
+            "src/python/pkg/base.py": "class Base:\n    pass\n",
+            "src/python/pkg/derived.py": (
+                "from pkg.base import Base\n\nclass Derived(Base):\n    pass\n"
+            ),
+            "src/python/pkg/BUILD": dedent(
+                """\
+                python_sources(
+                    sources=["base.py", "derived.py"],
+                    **parametrize("b", resolve="b"),
+                    **parametrize("a", resolve="a"),
+                    overrides={
+                        "derived.py": {
+                            "dependencies": ["./base.py"],
+                        },
+                    },
+                )
+                """
+            ),
+        }
+    )
+    rule_runner.set_options(["--pants-config-files=['pants.toml']"])
+    generated_targets = [
+        tgt
+        for tgt in rule_runner.request(
+            _TargetParametrizations,
+            [
+                _TargetParametrizationsRequest(
+                    Address("src/python/pkg"), description_of_origin="tests"
+                )
+            ],
+        ).parametrizations.values()
+        if isinstance(tgt, PythonSourceTarget)
+    ]
+    assert len(generated_targets) == 4
+    base_a = next(
+        tgt
+        for tgt in generated_targets
+        if tgt.address.relative_file_path == "base.py"
+        and tgt.address.parameters.get("parametrize") == "a"
+    )
+    derived_a = next(
+        tgt
+        for tgt in generated_targets
+        if tgt.address.relative_file_path == "derived.py"
+        and tgt.address.parameters.get("parametrize") == "a"
+    )
+
+    # The generated targets have the correct resolve.
+    assert base_a[PythonResolveField].value == "a"
+    assert derived_a[PythonResolveField].value == "a"
+    validation_field_set = DependencyValidationFieldSet.create(derived_a)
+    assert validation_field_set.resolve is not None
+    assert validation_field_set.resolve.value == "a"
+
+    resolved = rule_runner.request(
+        Addresses,
+        [DependenciesRequest(derived_a[PythonDependenciesField])],
+    )
+    assert tuple(resolved) == (base_a.address,)
+
+
+class _ReproRequiredField(StringField):
+    alias = "required"
+    required = True
+
+
+class _ReproOptionalField(StringField):
+    alias = "optional"
+
+
+class _ReproTarget(Target):
+    alias = "repro_target"
+    core_fields = (_ReproRequiredField,)
+
+
+@dataclass(frozen=True)
+class _ReproFieldSet(FieldSet):
+    required_fields = (_ReproRequiredField,)
+
+    required: _ReproRequiredField
+    optional: _ReproOptionalField | None = None
+
+
+def test_field_set_optional_union_field_is_none_when_target_lacks_field_repro() -> None:
+    target = _ReproTarget(
+        {_ReproRequiredField.alias: "value"},
+        Address("src/python/pkg", target_name="repro"),
+    )
+    field_set = _ReproFieldSet.create(target)
+
+    assert field_set.required.value == "value"
+    assert field_set.optional is None

src/python/pants/engine/target.py

@@ -19,6 +19,7 @@
 from enum import Enum
 from operator import attrgetter
 from pathlib import PurePath
+from types import UnionType
 from typing import (
     AbstractSet,
     Any,
@@ -27,8 +28,11 @@
     Protocol,
     Self,
     TypeVar,
+    Union,
     cast,
     final,
+    get_args,
+    get_origin,
     get_type_hints,
 )
 
@@ -1381,13 +1385,24 @@ def gen_tgt(address: Address, full_fp: str, generated_target_fields: dict[str, A
 # -----------------------------------------------------------------------------------------------
 def _get_field_set_fields_from_target(
     field_set: type[FieldSet], target: Target
-) -> dict[str, Field]:
-    return {
-        dataclass_field_name: (
-            target[field_cls] if field_cls in field_set.required_fields else target.get(field_cls)
-        )
-        for dataclass_field_name, field_cls in field_set.fields.items()
-    }
+) -> dict[str, Field | None]:
+    result: dict[str, Field | None] = {}
+    for dataclass_field_name, field_cls in field_set.fields.items():
+        if field_cls in field_set.required_fields:
+            result[dataclass_field_name] = target[field_cls]
+            continue
+
+        if dataclass_field_name in field_set.none_on_absence_fields and not target.has_field(
+            field_cls
+        ):
+            # Preserve true optionality for `Field | None` annotations when the target type
+            # doesn't define that field.
+            result[dataclass_field_name] = None
+            continue
+
+        result[dataclass_field_name] = target.get(field_cls)
+
+    return result
 
 
 _FS = TypeVar("_FS", bound="FieldSet")
@@ -1399,8 +1414,9 @@ class FieldSet(EngineAwareParameter, metaclass=ABCMeta):
 
     Subclasses should declare all the fields they consume as dataclass attributes. They should also
     indicate which of these are required, rather than optional, through the class property
-    `required_fields`. When a field is optional, the default constructor for the field will be used
-    for any targets that do not have that field registered.
+    `required_fields`. For fields annotated as `Field | None`, Pants will preserve `None` when the
+    target type does not have that field registered. For non-union `Field` annotations, Pants will
+    construct the default field value when the target type does not have that field registered.
 
     Subclasses must set `@dataclass(frozen=True)` for their declared fields to be recognized.
 
@@ -1472,15 +1488,77 @@ def applicable_target_types(
     def create(cls: type[_FS], tgt: Target) -> _FS:
         return cls(address=tgt.address, **_get_field_set_fields_from_target(cls, tgt))
 
+    @final
+    @memoized_classproperty
+    def _field_info(cls) -> FrozenDict[str, tuple[type[Field], bool]]:
+        def field_type_from_annotation(annotation: Any) -> tuple[type[Field], bool] | None:
+            if isinstance(annotation, type) and issubclass(annotation, Field):
+                return annotation, False
+
+            origin = get_origin(annotation)
+            if origin not in (Union, UnionType):
+                return None
+
+            union_args = get_args(annotation)
+            field_types = [
+                arg for arg in union_args if isinstance(arg, type) and issubclass(arg, Field)
+            ]
+            if len(field_types) != 1:
+                return None
+
+            preserve_none_on_absence = type(None) in union_args
+            # Only allow optional Field annotations (`Field | None`).
+            if not preserve_none_on_absence or len(union_args) != 2:
+                return None
+
+            return field_types[0], True
+
+        type_hints = get_type_hints(cls)
+        base_dataclass_field_names = {f.name for f in dataclasses.fields(FieldSet)}
+        parsed: dict[str, tuple[type[Field], bool]] = {}
+        invalid_dataclass_fields: dict[str, Any] = {}
+
+        for dataclass_field in dataclasses.fields(cls):
+            if dataclass_field.name in base_dataclass_field_names:
+                continue
+
+            annotation = type_hints[dataclass_field.name]
+            parsed_annotation = field_type_from_annotation(annotation)
+            if parsed_annotation is None:
+                invalid_dataclass_fields[dataclass_field.name] = annotation
+                continue
+
+            parsed[dataclass_field.name] = parsed_annotation
+
+        if invalid_dataclass_fields:
+            field_set_name = getattr(cls, "__name__", type(cls).__name__)
+            invalid_field_descriptions = ", ".join(
+                f"{name}: {annotation!r}"
+                for name, annotation in sorted(invalid_dataclass_fields.items())
+            )
+            raise TypeError(
+                f"The FieldSet `{field_set_name}` has invalid dataclass field annotations. "
+                "Every declared dataclass field on a FieldSet must be annotated with a "
+                "`Field` subclass or `Field | None`. Invalid fields: "
+                f"{invalid_field_descriptions}"
+            )
+
+        return FrozenDict(parsed)
+
     @final
     @memoized_classproperty
     def fields(cls) -> FrozenDict[str, type[Field]]:
         return FrozenDict(
-            (
-                (name, field_type)
-                for name, field_type in get_type_hints(cls).items()
-                if isinstance(field_type, type) and issubclass(field_type, Field)
-            )
+            (field_name, field_type) for field_name, (field_type, _) in cls._field_info.items()
+        )
+
+    @final
+    @memoized_classproperty
+    def none_on_absence_fields(cls) -> FrozenOrderedSet[str]:
+        return FrozenOrderedSet(
+            field_name
+            for field_name, (_, preserve_none_on_absence) in cls._field_info.items()
+            if preserve_none_on_absence
         )
 
     def debug_hint(self) -> str: