25405422172

Committed 05 May 2026 10:18PM UTC coverage: 92.879% (-0.07%) from 92.944%

Build # 25405422172

Build Type

Pull #23319

github

Committed by

web-flow

Commit Message

Merge c82d0f333 into e8b784f89

Pull Request Pull Request #23319: [pants_ng] Scaffolding for a pants_ng mode.

Coverage Stats

25 of 76 new or added lines in 9 files covered. (32.89%)

209 existing lines in 15 files now uncovered.

92234 of 99306 relevant lines covered (92.88%)

4.05 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/python/pants/bin/pants_loader.py

# Copyright 2017 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).


import importlib
import locale
import logging
import os
import sys
import time
import warnings
from datetime import datetime

from pants.base.exiter import PANTS_FAILED_EXIT_CODE
from pants.bin.pants_env_vars import (
    DAEMON_ENTRYPOINT,
    IGNORE_UNRECOGNIZED_ENCODING,
    RECURSION_LIMIT,
)
from pants.bin.pants_runner import PantsRunner
from pants.engine.internals import native_engine
from pants.util.strutil import softwrap

# pants: infer-dep(/src/python/pants/backend/**/register.py)
# pants: infer-dep(/src/python/pants/core/**/*)


class PantsLoader:
    """Initial entrypoint for pants.

    Executes a pants_runner by default, or executes a pantsd-specific entrypoint.
    """

    @staticmethod
    def setup_warnings() -> None:
        # We want to present warnings to the user, set this up before importing any of our own code,
        # to ensure all deprecation warnings are seen, including module deprecations.
        # The "default" action displays a warning for a particular file and line number exactly once.
        # See https://docs.python.org/3/library/warnings.html#the-warnings-filter for the complete list.
        #
        # However, we do turn off deprecation warnings for libraries that Pants uses for which we do
        # not have a fixed upstream version, typically because the library is no longer maintained.
        warnings.simplefilter("default", category=DeprecationWarning)
        # TODO: Eric-Arellano has emailed the author to see if he is willing to accept a PR fixing the
        # deprecation warnings and to release the fix. If he says yes, remove this once fixed.
        warnings.filterwarnings("ignore", category=DeprecationWarning, module="ansicolors")
        # Silence ctypes _pack_/_layout_ warning from HdrHistogram; due by Python 3.19
        # See: https://github.com/HdrHistogram/HdrHistogram/issues/216
        warnings.filterwarnings(
            "ignore",
            category=DeprecationWarning,
            message=r"Due to '_pack_', the '.+' Structure will use memory layout compatible with MSVC",
        )
        # Silence this ubiquitous warning. Several of our 3rd party deps incur this.
        warnings.filterwarnings(
            "ignore",
            category=DeprecationWarning,
            message="Using or importing the ABCs from 'collections' instead of from 'collections.abc' is deprecated",
        )

    @classmethod
    def ensure_locale(cls) -> None:
        """Ensure the locale uses UTF-8 encoding, or prompt for an explicit bypass."""

        encoding = locale.getpreferredencoding()
        if (
            encoding.lower() != "utf-8"
            and os.environ.get(IGNORE_UNRECOGNIZED_ENCODING, None) is None
        ):
            raise RuntimeError(
                softwrap(
                    f"""
                    Your system's preferred encoding is `{encoding}`, but Pants requires `UTF-8`.
                    Specifically, Python's `locale.getpreferredencoding()` must resolve to `UTF-8`.

                    You can fix this by setting the LC_* and LANG environment variables, e.g.:
                      LC_ALL=en_US.UTF-8
                      LANG=en_US.UTF-8
                    Or, bypass it by setting {IGNORE_UNRECOGNIZED_ENCODING}=1. Note that
                    pants may exhibit inconsistent behavior if this check is bypassed.
                    """
                )
            )

    @staticmethod
    def sandboxer_bin() -> str | None:
        # In theory as_file could return a temporary file and clean it up, so we'd be returning
        # an invalid path. But in practice we know that we're running either in a venv with all
        # resources expanded on disk, or from sources, and either way we will get a persistent
        # valid path that will not be cleaned up.
        with importlib.resources.as_file(
            importlib.resources.files("pants.bin").joinpath("sandboxer")
        ) as sandboxer_bin:
            if os.path.isfile(sandboxer_bin):
                os.chmod(sandboxer_bin, 0o755)
                return str(sandboxer_bin)
        return None

    @staticmethod
    def run_alternate_entrypoint(entrypoint: str) -> None:
        try:
            module_path, func_name = entrypoint.split(":", 1)
        except ValueError:
            print(
                f"{DAEMON_ENTRYPOINT} must be of the form `module.path:callable`", file=sys.stderr
            )
            sys.exit(PANTS_FAILED_EXIT_CODE)

        module = importlib.import_module(module_path)
        entrypoint_fn = getattr(module, func_name)
        entrypoint_fn()

    @staticmethod
    def run_default_entrypoint() -> None:
        start_time = time.time()
        try:
            runner = PantsRunner(args=sys.argv, env=os.environ)
            exit_code = runner.run(start_time)
        except KeyboardInterrupt as e:
            print(f"Interrupted by user:\n{e}", file=sys.stderr)
            exit_code = PANTS_FAILED_EXIT_CODE
        sys.exit(exit_code)

    @classmethod
    def main(cls) -> None:
        # Set up logging. This is just temporary, as the PantsRunner will re-initialize logging to
        # be emitted via Rust, but until that happens we want basic logging to do something useful.
        # We set it up to be visually compatible with the Rust logging.
        class HundredthsFormatter(logging.Formatter):
            def formatTime(self, record, datefmt=None):
                dt = datetime.fromtimestamp(record.created)
                formatted_time = dt.strftime(datefmt or "%H:%M:%S")
                hundredths = str(dt.microsecond // 10000).zfill(2)
                return f"{formatted_time}.{hundredths}"

        handler = logging.StreamHandler()
        handler.setFormatter(HundredthsFormatter(fmt="%(asctime)s [%(levelname)s] %(message)s"))
        logging.basicConfig(level=logging.INFO, handlers=[handler])

        native_engine.initialize()
        cls.setup_warnings()
        cls.ensure_locale()

        sys.setrecursionlimit(int(os.environ.get(RECURSION_LIMIT, "10000")))

        os.environ["PANTS_SANDBOXER_BINARY_PATH"] = cls.sandboxer_bin() or ""
        entrypoint = os.environ.pop(DAEMON_ENTRYPOINT, None)
        if entrypoint:
            cls.run_alternate_entrypoint(entrypoint)
        else:
            cls.run_default_entrypoint()


def main() -> None:
    PantsLoader.main()


if __name__ == "__main__":
    main()

1	# Copyright 2017 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4
5	import importlib	×
6	import locale	×
NEW 7	import logging	×
8	import os	×
9	import sys	×
10	import time	×
11	import warnings	×
NEW 12	from datetime import datetime	×
13
14	from pants.base.exiter import PANTS_FAILED_EXIT_CODE	×
15	from pants.bin.pants_env_vars import (	×
16	DAEMON_ENTRYPOINT,
17	IGNORE_UNRECOGNIZED_ENCODING,
18	RECURSION_LIMIT,
19	)
20	from pants.bin.pants_runner import PantsRunner	×
21	from pants.engine.internals import native_engine	×
22	from pants.util.strutil import softwrap	×
23
24	# pants: infer-dep(/src/python/pants/backend/**/register.py)
25	# pants: infer-dep(/src/python/pants/core/*/)
26
27
28	class PantsLoader:	×
29	"""Initial entrypoint for pants.
30
31	Executes a pants_runner by default, or executes a pantsd-specific entrypoint.
32	"""
33
34	@staticmethod	×
35	def setup_warnings() -> None:	×
36	# We want to present warnings to the user, set this up before importing any of our own code,
37	# to ensure all deprecation warnings are seen, including module deprecations.
38	# The "default" action displays a warning for a particular file and line number exactly once.
39	# See https://docs.python.org/3/library/warnings.html#the-warnings-filter for the complete list.
40	#
41	# However, we do turn off deprecation warnings for libraries that Pants uses for which we do
42	# not have a fixed upstream version, typically because the library is no longer maintained.
43	warnings.simplefilter("default", category=DeprecationWarning)	×
44	# TODO: Eric-Arellano has emailed the author to see if he is willing to accept a PR fixing the
45	# deprecation warnings and to release the fix. If he says yes, remove this once fixed.
46	warnings.filterwarnings("ignore", category=DeprecationWarning, module="ansicolors")	×
47	# Silence ctypes _pack_/_layout_ warning from HdrHistogram; due by Python 3.19
48	# See: https://github.com/HdrHistogram/HdrHistogram/issues/216
49	warnings.filterwarnings(	×
50	"ignore",
51	category=DeprecationWarning,
52	message=r"Due to '_pack_', the '.+' Structure will use memory layout compatible with MSVC",
53	)
54	# Silence this ubiquitous warning. Several of our 3rd party deps incur this.
55	warnings.filterwarnings(	×
56	"ignore",
57	category=DeprecationWarning,
58	message="Using or importing the ABCs from 'collections' instead of from 'collections.abc' is deprecated",
59	)
60
61	@classmethod	×
62	def ensure_locale(cls) -> None:	×
63	"""Ensure the locale uses UTF-8 encoding, or prompt for an explicit bypass."""
64
65	encoding = locale.getpreferredencoding()	×
66	if (	×
67	encoding.lower() != "utf-8"
68	and os.environ.get(IGNORE_UNRECOGNIZED_ENCODING, None) is None
69	):
70	raise RuntimeError(	×
71	softwrap(
72	f"""
73	Your system's preferred encoding is `{encoding}`, but Pants requires `UTF-8`.
74	Specifically, Python's `locale.getpreferredencoding()` must resolve to `UTF-8`.
75
76	You can fix this by setting the LC_* and LANG environment variables, e.g.:
77	LC_ALL=en_US.UTF-8
78	LANG=en_US.UTF-8
79	Or, bypass it by setting {IGNORE_UNRECOGNIZED_ENCODING}=1. Note that
80	pants may exhibit inconsistent behavior if this check is bypassed.
81	"""
82	)
83	)
84
85	@staticmethod	×
86	def sandboxer_bin() -> str \| None:	×
87	# In theory as_file could return a temporary file and clean it up, so we'd be returning
88	# an invalid path. But in practice we know that we're running either in a venv with all
89	# resources expanded on disk, or from sources, and either way we will get a persistent
90	# valid path that will not be cleaned up.
91	with importlib.resources.as_file(	×
92	importlib.resources.files("pants.bin").joinpath("sandboxer")
93	) as sandboxer_bin:
94	if os.path.isfile(sandboxer_bin):	×
95	os.chmod(sandboxer_bin, 0o755)	×
96	return str(sandboxer_bin)	×
97	return None	×
98
99	@staticmethod	×
100	def run_alternate_entrypoint(entrypoint: str) -> None:	×
101	try:	×
102	module_path, func_name = entrypoint.split(":", 1)	×
103	except ValueError:	×
104	print(	×
105	f"{DAEMON_ENTRYPOINT} must be of the form `module.path:callable`", file=sys.stderr
106	)
107	sys.exit(PANTS_FAILED_EXIT_CODE)	×
108
109	module = importlib.import_module(module_path)	×
110	entrypoint_fn = getattr(module, func_name)	×
111	entrypoint_fn()	×
112
113	@staticmethod	×
114	def run_default_entrypoint() -> None:	×
115	start_time = time.time()	×
116	try:	×
117	runner = PantsRunner(args=sys.argv, env=os.environ)	×
118	exit_code = runner.run(start_time)	×
119	except KeyboardInterrupt as e:	×
120	print(f"Interrupted by user:\n{e}", file=sys.stderr)	×
121	exit_code = PANTS_FAILED_EXIT_CODE	×
122	sys.exit(exit_code)	×
123
124	@classmethod	×
125	def main(cls) -> None:	×
126	# Set up logging. This is just temporary, as the PantsRunner will re-initialize logging to
127	# be emitted via Rust, but until that happens we want basic logging to do something useful.
128	# We set it up to be visually compatible with the Rust logging.
NEW 129	class HundredthsFormatter(logging.Formatter):	×
NEW 130	def formatTime(self, record, datefmt=None):	×
NEW 131	dt = datetime.fromtimestamp(record.created)	×
NEW 132	formatted_time = dt.strftime(datefmt or "%H:%M:%S")	×
NEW 133	hundredths = str(dt.microsecond // 10000).zfill(2)	×
NEW 134	return f"{formatted_time}.{hundredths}"	×
135
NEW 136	handler = logging.StreamHandler()	×
NEW 137	handler.setFormatter(HundredthsFormatter(fmt="%(asctime)s [%(levelname)s] %(message)s"))	×
NEW 138	logging.basicConfig(level=logging.INFO, handlers=[handler])	×
139
140	native_engine.initialize()	×
141	cls.setup_warnings()	×
142	cls.ensure_locale()	×
143
144	sys.setrecursionlimit(int(os.environ.get(RECURSION_LIMIT, "10000")))	×
145
146	os.environ["PANTS_SANDBOXER_BINARY_PATH"] = cls.sandboxer_bin() or ""	×
147	entrypoint = os.environ.pop(DAEMON_ENTRYPOINT, None)	×
148	if entrypoint:	×
149	cls.run_alternate_entrypoint(entrypoint)	×
150	else:
151	cls.run_default_entrypoint()	×
152
153
154	def main() -> None:	×
155	PantsLoader.main()	×
156
157
158	if __name__ == "__main__":	×
159	main()	×

pantsbuild / pants / 25405422172

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous