Hook the Rust options parser into Python (not for review)

src/python/pants/engine/internals/native_engine.pyi

@@ -12,6 +12,7 @@ from typing import (
     Generic,
     Iterable,
     Mapping,
+    Optional,
     Protocol,
     Sequence,
     TextIO,
@@ -533,6 +534,49 @@ class PantsdConnectionException(Exception):
 class PantsdClientException(Exception):
     pass
 
+# ------------------------------------------------------------------------------
+# Options
+# ------------------------------------------------------------------------------
+
+class PyOptionId:
+    def __init__(
+        self, *components: str, scope: str | None = None, switch: str | None = None
+    ) -> None: ...
+
+class PyConfigSource:
+    def __init__(self, path: str, content: bytes) -> None: ...
+
+# A pair of (option value, rank). See src/python/pants/option/ranked_value.py.
+T = TypeVar("T")
+OptionValue = Tuple[Optional[T], int]
+OptionListValue = Tuple[list[T], int]
+OptionDictValue = Tuple[dict[str, Any], int]
+
+class PyOptionParser:
+    def __init__(
+        self,
+        args: Optional[Sequence[str]],
+        env: dict[str, str],
+        configs: Optional[Sequence[PyConfigSource]],
+        allow_pantsrc: bool,
+    ) -> None: ...
+    def get_bool(self, option_id: PyOptionId, default: Optional[bool]) -> OptionValue[bool]: ...
+    def get_int(self, option_id: PyOptionId, default: Optional[int]) -> OptionValue[int]: ...
+    def get_float(self, option_id: PyOptionId, default: Optional[float]) -> OptionValue[float]: ...
+    def get_string(self, option_id: PyOptionId, default: Optional[str]) -> OptionValue[str]: ...
+    def get_bool_list(
+        self, option_id: PyOptionId, default: list[bool]
+    ) -> OptionListValue[bool]: ...
+    def get_int_list(self, option_id: PyOptionId, default: list[int]) -> OptionListValue[int]: ...
+    def get_float_list(
+        self, option_id: PyOptionId, default: list[float]
+    ) -> OptionListValue[float]: ...
+    def get_string_list(
+        self, option_id: PyOptionId, default: list[str]
+    ) -> OptionListValue[str]: ...
+    def get_dict(self, option_id: PyOptionId, default: dict[str, Any]) -> OptionDictValue: ...
+    def get_passthrough_args(self) -> Optional[list[str]]: ...
+
 # ------------------------------------------------------------------------------
 # Testutil
 # ------------------------------------------------------------------------------

src/python/pants/option/options.py

@@ -15,6 +15,7 @@
 from pants.option.option_util import is_list_option
 from pants.option.option_value_container import OptionValueContainer, OptionValueContainerBuilder
 from pants.option.parser import Parser
+from pants.option.rust_options import NativeOptionParser
 from pants.option.scope import GLOBAL_SCOPE, GLOBAL_SCOPE_CONFIG_SECTION, ScopeInfo
 from pants.util.memo import memoized_method
 from pants.util.ordered_set import FrozenOrderedSet, OrderedSet
@@ -149,6 +150,9 @@ def create(
 
         parser_by_scope = {si.scope: Parser(env, config, si) for si in complete_known_scope_infos}
         known_scope_to_info = {s.scope: s for s in complete_known_scope_infos}
+
+        native_parser = NativeOptionParser(args, env, config.sources(), allow_pantsrc=True)
+
         return cls(
             builtin_goal=split_args.builtin_goal,
             goals=split_args.goals,
@@ -157,6 +161,7 @@ def create(
             specs=split_args.specs,
             passthru=split_args.passthru,
             parser_by_scope=parser_by_scope,
+            native_parser=native_parser,
             bootstrap_option_values=bootstrap_option_values,
             known_scope_to_info=known_scope_to_info,
             allow_unknown_options=allow_unknown_options,
@@ -171,6 +176,7 @@ def __init__(
         specs: list[str],
         passthru: list[str],
         parser_by_scope: dict[str, Parser],
+        native_parser: NativeOptionParser,
         bootstrap_option_values: OptionValueContainer | None,
         known_scope_to_info: dict[str, ScopeInfo],
         allow_unknown_options: bool = False,
@@ -186,6 +192,7 @@ def __init__(
         self._specs = specs
         self._passthru = passthru
         self._parser_by_scope = parser_by_scope
+        self._native_parser = native_parser
         self._bootstrap_option_values = bootstrap_option_values
         self._known_scope_to_info = known_scope_to_info
         self._allow_unknown_options = allow_unknown_options
@@ -368,12 +375,35 @@ def for_scope(
             parse_args_request, log_warnings=log_parser_warnings
         )
 
+        native_values = self.get_parser(scope).parse_args_native(self._native_parser)
+
         # Check for any deprecation conditions, which are evaluated using `self._flag_matchers`.
         if check_deprecations:
             values_builder = values.to_builder()
             self._check_and_apply_deprecations(scope, values_builder)
             values = values_builder.build()
 
+            native_values_builder = native_values.to_builder()
+            self._check_and_apply_deprecations(scope, native_values_builder)
+            native_values = native_values_builder.build()
+
+        def listify_tuples(x):
+            if isinstance(x, (tuple, list)):
+                return [listify_tuples(y) for y in x]
+            elif isinstance(x, dict):
+                return {k: listify_tuples(v) for k, v in x.items()}
+            else:
+                return x
+
+        for key, rv in values.as_dict().items():
+            rv = listify_tuples(rv)
+            if native_values[key] != rv:
+                logger.warning(
+                    f"Value mismatch for option `{key}` in scope [{scope}]:\n"
+                    f"Rust   value: {native_values[key]} of type {type(native_values[key])} provided by {native_values.get_rank(key)}\n"
+                    f"Python value: {rv} of type {type(rv)} provided by {values.get_rank(key)}"
+                )
+                raise Exception("Option value mismatch")
         return values
 
     def get_fingerprintable_for_scope(

src/python/pants/option/parser.py

@@ -52,6 +52,7 @@
 from pants.option.option_util import is_dict_option, is_list_option
 from pants.option.option_value_container import OptionValueContainer, OptionValueContainerBuilder
 from pants.option.ranked_value import Rank, RankedValue
+from pants.option.rust_options import NativeOptionParser
 from pants.option.scope import GLOBAL_SCOPE, GLOBAL_SCOPE_CONFIG_SECTION, ScopeInfo
 from pants.util.strutil import softwrap
 
@@ -193,6 +194,22 @@ def _create_flag_value_map(flags: Iterable[str]) -> DefaultDict[str, list[str |
                 flag_value_map[key].append(flag_val)
             return flag_value_map
 
+    def parse_args_native(self, native_parser: NativeOptionParser) -> OptionValueContainer:
+        namespace = OptionValueContainerBuilder()
+        for args, kwargs in self._option_registrations:
+            self._validate(args, kwargs)
+            dest = self.parse_dest(*args, **kwargs)
+            val, rank = native_parser.get(
+                scope=self.scope,
+                flags=args,
+                default=kwargs.get("default"),
+                option_type=kwargs.get("type"),
+                member_type=kwargs.get("member_type"),
+                passthrough=kwargs.get("passthrough"),
+            )
+            setattr(namespace, dest, RankedValue(rank, val))
+        return namespace.build()
+
     def parse_args(
         self, parse_args_request: ParseArgsRequest, log_warnings: bool = False
     ) -> OptionValueContainer:

src/python/pants/option/ranked_value.py

@@ -9,6 +9,7 @@
 from typing import Any, Dict, Iterator, List, Optional, Tuple, Union
 
 
+# NB: Must mirror the Rank enum in src/rust/engine/options/src/lib.rs.
 @total_ordering
 class Rank(Enum):
     # The ranked value sources. Higher ranks override lower ones.

src/python/pants/option/rust_options.py

@@ -0,0 +1,138 @@
+# Copyright 2024 Pants project contributors (see CONTRIBUTORS.md).
+# Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+from __future__ import annotations
+
+import inspect
+import logging
+import shlex
+from enum import Enum
+from typing import Any, Mapping, Optional, Sequence, Tuple
+
+from pants.engine.internals import native_engine
+from pants.engine.internals.native_engine import PyConfigSource
+from pants.option.config import ConfigSource
+from pants.option.custom_types import _flatten_shlexed_list, shell_str
+from pants.option.errors import OptionsError
+from pants.option.ranked_value import Rank
+from pants.util.strutil import get_strict_env
+
+logger = logging.getLogger()
+
+
+class NativeOptionParser:
+    int_to_rank = [
+        Rank.NONE,
+        Rank.HARDCODED,
+        Rank.CONFIG_DEFAULT,
+        Rank.CONFIG,
+        Rank.ENVIRONMENT,
+        Rank.FLAG,
+    ]
+
+    def __init__(
+        self,
+        args: Optional[Sequence[str]],
+        env: Mapping[str, str],
+        config_sources: Optional[Sequence[ConfigSource]],
+        allow_pantsrc: bool,
+    ):
+        py_config_sources = (
+            None
+            if config_sources is None
+            else [PyConfigSource(cs.path, cs.content) for cs in config_sources]
+        )
+        self._native_parser = native_engine.PyOptionParser(
+            args,
+            dict(get_strict_env(env, logger)),
+            py_config_sources,
+            allow_pantsrc,
+        )
+
+        self._getter_by_type = {
+            (bool, None): self._native_parser.get_bool,
+            (int, None): self._native_parser.get_int,
+            (float, None): self._native_parser.get_float,
+            (str, None): self._native_parser.get_string,
+            (list, bool): self._native_parser.get_bool_list,
+            (list, int): self._native_parser.get_int_list,
+            (list, float): self._native_parser.get_float_list,
+            (list, str): self._native_parser.get_string_list,
+            (dict, None): self._native_parser.get_dict,
+        }
+
+    def get(
+        self, *, scope, flags, default, option_type, member_type=None, passthrough=False
+    ) -> Tuple[Any, Rank]:
+        def is_enum(typ):
+            # TODO: When we switch to Python 3.11, use: return isinstance(typ, EnumType)
+            return inspect.isclass(typ) and issubclass(typ, Enum)
+
+        name_parts = (
+            flags[-1][2:].replace(".", "-").split("-")
+        )  # '--foo.bar-baz' -> ['foo', 'bar', 'baz']
+        switch = flags[0][1:] if len(flags) > 1 else None  # '-d' -> 'd'
+        option_id = native_engine.PyOptionId(*name_parts, scope=scope or "GLOBAL", switch=switch)
+
+        rust_option_type = option_type
+        rust_member_type = member_type
+
+        if option_type is dict:
+            # The Python code allows registering default=None for dicts/lists, and forces it to
+            # an empty dict/list at registration. Since here we only have access to what the user
+            # provided, we do the same.
+            if default is None:
+                default = {}
+            elif isinstance(default, str):
+                default = eval(default)
+        elif option_type is list:
+            if default is None:
+                default = []
+            if member_type is None:
+                member_type = rust_member_type = str
+
+            if member_type == shell_str:
+                rust_member_type = str
+                if isinstance(default, str):
+                    default = shlex.split(default)
+            elif is_enum(member_type):
+                rust_member_type = str
+                default = [x.value for x in default]
+            elif inspect.isfunction(rust_member_type):
+                rust_member_type = str
+            elif rust_member_type != str and isinstance(default, str):
+                default = eval(default)
+        elif is_enum(option_type):
+            if default is not None:
+                default = default.value
+                rust_option_type = type(default)
+            else:
+                rust_option_type = str
+        elif option_type not in {bool, int, float, str}:
+            # For enum and other specialized types.
+            rust_option_type = str
+            if default is not None:
+                default = str(default)
+
+        getter = self._getter_by_type.get((rust_option_type, rust_member_type))
+        if getter is None:
+            suffix = f" with member type {rust_member_type}" if rust_option_type is list else ""
+            raise OptionsError(f"Unsupported type: {rust_option_type}{suffix}")
+
+        val, rank_int = getter(option_id, default)  # type:ignore
+        rank = self.int_to_rank[rank_int]
+
+        if val is not None:
+            if option_type is list:
+                if member_type == shell_str:
+                    val = _flatten_shlexed_list(val)
+                elif callable(member_type):
+                    val = [member_type(x) for x in val]
+                if passthrough:
+                    val += self._native_parser.get_passthrough_args() or []
+            elif is_enum(option_type):
+                val = option_type(val)
+            elif callable(option_type):
+                val = option_type(val)
+
+        return (val, rank)