We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
`log_level_by_target` doesn't seem to work well for Python targets
**Describe the bug**
Using the following:
```
level = "warn"
log_levels_by_target = {"pants.goal.stats_aggregator" = "info"}
```
when running `./pants --stats-log ...` I don't see the stats being logged.

**Pants version**
`main`

**OS**
Ubuntu

**Additional info**
I suspect this is because `log_levels_by_target` isn't used to configure the Python logger, and therefore the python logger for any Python module is configured to use the level set by `level`.

This can be seen by inspecting the logger in `src/python/pants/goal/stats_aggregator.py` which is set to the level `WARN`. Therefore I assume the log call never gets forwarded to the Rust-implemented handler, and therefore `log_levels_by_target` isn't considered.


--- 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/init/logging.py`
Content:
```
1 # Copyright 2018 Pants project contributors (see CONTRIBUTORS.md).
2 # Licensed under the Apache License, Version 2.0 (see LICENSE).
3 
4 from __future__ import annotations
5 
6 import http.client
7 import locale
8 import logging
9 import sys
10 from contextlib import contextmanager
11 from io import BufferedReader, TextIOWrapper
12 from logging import Formatter, LogRecord, StreamHandler
13 from pathlib import PurePath
14 from typing import Iterator
15 
16 import pants.util.logging as pants_logging
17 from pants.engine.internals import native_engine
18 from pants.option.option_value_container import OptionValueContainer
19 from pants.util.dirutil import safe_mkdir_for
20 from pants.util.docutil import doc_url
21 from pants.util.logging import LogLevel
22 from pants.util.strutil import strip_prefix
23 
24 # Although logging supports the WARN level, its not documented and could conceivably be yanked.
25 # Since pants has supported 'warn' since inception, leave the 'warn' choice as-is but explicitly
26 # setup a 'WARN' logging level name that maps to 'WARNING'.
27 logging.addLevelName(logging.WARNING, "WARN")
28 logging.addLevelName(pants_logging.TRACE, "TRACE")
29 
30 
31 class _NativeHandler(StreamHandler):
32     """This class is installed as a Python logging module handler (using the logging.addHandler
33     method) and proxies logs to the Rust logging infrastructure."""
34 
35     def emit(self, record: LogRecord) -> None:
36         native_engine.write_log(self.format(record), record.levelno, record.name)
37 
38     def flush(self) -> None:
39         native_engine.flush_log()
40 
41 
42 class _ExceptionFormatter(Formatter):
43     """Possibly render the stacktrace and possibly give debug hints, based on global options."""
44 
45     def __init__(self, level: LogLevel, *, print_stacktrace: bool) -> None:
46         super().__init__(None)
47         self.level = level
48         self.print_stacktrace = print_stacktrace
49 
50     def formatException(self, exc_info):
51         stacktrace = super().formatException(exc_info) if self.print_stacktrace else ""
52 
53         debug_instructions = []
54         if not self.print_stacktrace:
55             debug_instructions.append("--print-stacktrace for more error details")
56         if self.level not in {LogLevel.DEBUG, LogLevel.TRACE}:
57             debug_instructions.append("-ldebug for more logs")
58         debug_instructions = (
59             f"Use {' and/or '.join(debug_instructions)}. " if debug_instructions else ""
60         )
61 
62         return (
63             f"{stacktrace}\n\n{debug_instructions}\nSee {doc_url('troubleshooting')} for common "
64             f"issues.\nConsider reaching out for help: {doc_url('getting-help')}\n"
65         )
66 
67 
68 @contextmanager
69 def stdio_destination(stdin_fileno: int, stdout_fileno: int, stderr_fileno: int) -> Iterator[None]:
70     """Sets a destination for both logging and stdio: must be called after `initialize_stdio`.
71 
72     After `initialize_stdio` and outside of this contextmanager, the default stdio destination is
73     the pants.log. But inside of this block, all engine "tasks"/@rules that are spawned will have
74     thread/task-local state that directs their IO to the given destination. When the contextmanager
75     exits all tasks will be restored to the default destination (regardless of whether they have
76     completed).
77     """
78     if not logging.getLogger(None).handlers:
79         raise AssertionError("stdio_destination should only be called after initialize_stdio.")
80 
81     native_engine.stdio_thread_console_set(stdin_fileno, stdout_fileno, stderr_fileno)
82     try:
83         yield
84     finally:
85         native_engine.stdio_thread_console_clear()
86 
87 
88 def stdio_destination_use_color(use_color: bool) -> None:
89     """Sets a color mode for the current thread's destination.
90 
91     True or false force color to be used or not used: None causes TTY detection to decide whether
92     color will be used.
93 
94     NB: This method is independent from either `stdio_destination` or `initialize_stdio` because
95     we cannot decide whether to use color for a particular destination until it is open AND we have
96     parsed options for the relevant connection.
97     """
98     native_engine.stdio_thread_console_color_mode_set(use_color)
99 
100 
101 @contextmanager
102 def _python_logging_setup(level: LogLevel, *, print_stacktrace: bool) -> Iterator[None]:
103     """Installs a root Python logger that routes all logging through a Rust logger."""
104 
105     def trace_fn(self, message, *args, **kwargs):
106         if self.isEnabledFor(LogLevel.TRACE.level):
107             self._log(LogLevel.TRACE.level, message, *args, **kwargs)
108 
109     logging.Logger.trace = trace_fn  # type: ignore[attr-defined]
110     logger = logging.getLogger(None)
111 
112     def clear_logging_handlers():
113         handlers = tuple(logger.handlers)
114         for handler in handlers:
115             logger.removeHandler(handler)
116         return handlers
117 
118     def set_logging_handlers(handlers):
119         for handler in handlers:
120             logger.addHandler(handler)
121 
122     # Remove existing handlers, and restore them afterward.
123     handlers = clear_logging_handlers()
124     try:
125         # This routes warnings through our loggers instead of straight to raw stderr.
126         logging.captureWarnings(True)
127         handler = _NativeHandler()
128         exc_formatter = _ExceptionFormatter(level, print_stacktrace=print_stacktrace)
129         handler.setFormatter(exc_formatter)
130         logger.addHandler(handler)
131         level.set_level_for(logger)
132 
133         if logger.isEnabledFor(LogLevel.TRACE.level):
134             http.client.HTTPConnection.debuglevel = 1  # type: ignore[attr-defined]
135             requests_logger = logging.getLogger("requests.packages.urllib3")
136             LogLevel.TRACE.set_level_for(requests_logger)
137             requests_logger.propagate = True
138 
139         yield
140     finally:
141         clear_logging_handlers()
142         set_logging_handlers(handlers)
143 
144 
145 @contextmanager
146 def initialize_stdio(global_bootstrap_options: OptionValueContainer) -> Iterator[None]:
147     """Mutates sys.std* and logging to route stdio for a Pants process to thread local destinations.
148 
149     In this context, `sys.std*` and logging handlers will route through Rust code that uses
150     thread-local information to decide whether to write to a file, or to stdio file handles.
151 
152     To control the stdio destination set by this method, use the `stdio_destination` context manager.
153 
154     This is called in two different processes:
155     * PantsRunner, after it has determined that LocalPantsRunner will be running in process, and
156       immediately before setting a `stdio_destination` for the remainder of the run.
157     * PantsDaemon, immediately on startup. The process will then default to sending stdio to the log
158       until client connections arrive, at which point `stdio_destination` is used per-connection.
159     """
160     with initialize_stdio_raw(
161         global_bootstrap_options.level,
162         global_bootstrap_options.log_show_rust_3rdparty,
163         global_bootstrap_options.show_log_target,
164         _get_log_levels_by_target(global_bootstrap_options),
165         global_bootstrap_options.print_stacktrace,
166         global_bootstrap_options.ignore_warnings,
167         global_bootstrap_options.pants_workdir,
168     ):
169         yield
170 
171 
172 @contextmanager
173 def initialize_stdio_raw(
174     global_level: LogLevel,
175     log_show_rust_3rdparty: bool,
176     show_target: bool,
177     log_levels_by_target: dict[str, LogLevel],
178     print_stacktrace: bool,
179     ignore_warnings: list[str],
180     pants_workdir: str,
181 ) -> Iterator[None]:
182     literal_filters = []
183     regex_filters = []
184     for filt in ignore_warnings:
185         if filt.startswith("$regex$"):
186             regex_filters.append(strip_prefix(filt, "$regex$"))
187         else:
188             literal_filters.append(filt)
189 
190     # Set the pants log destination.
191     log_path = str(pants_log_path(PurePath(pants_workdir)))
192     safe_mkdir_for(log_path)
193 
194     # Initialize thread-local stdio, and replace sys.std* with proxies.
195     original_stdin, original_stdout, original_stderr = sys.stdin, sys.stdout, sys.stderr
196     try:
197         raw_stdin, sys.stdout, sys.stderr = native_engine.stdio_initialize(
198             global_level.level,
199             log_show_rust_3rdparty,
200             show_target,
201             {k: v.level for k, v in log_levels_by_target.items()},
202             tuple(literal_filters),
203             tuple(regex_filters),
204             log_path,
205         )
206         sys.stdin = TextIOWrapper(
207             BufferedReader(raw_stdin),
208             # NB: We set the default encoding explicitly to bypass logic in the TextIOWrapper
209             # constructor that would poke the underlying file (which is not valid until a
210             # `stdio_destination` is set).
211             encoding=locale.getpreferredencoding(False),
212         )
213 
214         sys.__stdin__, sys.__stdout__, sys.__stderr__ = sys.stdin, sys.stdout, sys.stderr
215         # Install a Python logger that will route through the Rust logger.
216         with _python_logging_setup(global_level, print_stacktrace=print_stacktrace):
217             yield
218     finally:
219         sys.stdin, sys.stdout, sys.stderr = original_stdin, original_stdout, original_stderr
220         sys.__stdin__, sys.__stdout__, sys.__stderr__ = sys.stdin, sys.stdout, sys.stderr
221 
222 
223 def pants_log_path(workdir: PurePath) -> PurePath:
224     """Given the path of the workdir, returns the `pants.log` path."""
225     return workdir / "pants.log"
226 
227 
228 def _get_log_levels_by_target(
229     global_bootstrap_options: OptionValueContainer,
230 ) -> dict[str, LogLevel]:
231     raw_levels = global_bootstrap_options.log_levels_by_target
232     levels: dict[str, LogLevel] = {}
233     for key, value in raw_levels.items():
234         if not isinstance(key, str):
235             raise ValueError(
236                 "Keys for log_domain_levels must be strings, but was given the key: {key} with type {type(key)}."
237             )
238         if not isinstance(value, str):
239             raise ValueError(
240                 "Values for log_domain_levels must be strings, but was given the value: {value} with type {type(value)}."
241             )
242         log_level = LogLevel[value.upper()]
243         levels[key] = log_level
244     return levels
245
```
--- 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/init/logging.py b/src/python/pants/init/logging.py
--- a/src/python/pants/init/logging.py
+++ b/src/python/pants/init/logging.py
@@ -99,7 +99,9 @@
 
 
 @contextmanager
-def _python_logging_setup(level: LogLevel, *, print_stacktrace: bool) -> Iterator[None]:
+def _python_logging_setup(
+    level: LogLevel, log_levels_by_target: dict[str, LogLevel], *, print_stacktrace: bool
+) -> Iterator[None]:
     """Installs a root Python logger that routes all logging through a Rust logger."""
 
     def trace_fn(self, message, *args, **kwargs):
@@ -130,6 +132,9 @@
         logger.addHandler(handler)
         level.set_level_for(logger)
 
+        for key, level in log_levels_by_target.items():
+            level.set_level_for(logging.getLogger(key))
+
         if logger.isEnabledFor(LogLevel.TRACE.level):
             http.client.HTTPConnection.debuglevel = 1  # type: ignore[attr-defined]
             requests_logger = logging.getLogger("requests.packages.urllib3")
@@ -213,7 +218,9 @@
 
         sys.__stdin__, sys.__stdout__, sys.__stderr__ = sys.stdin, sys.stdout, sys.stderr
         # Install a Python logger that will route through the Rust logger.
-        with _python_logging_setup(global_level, print_stacktrace=print_stacktrace):
+        with _python_logging_setup(
+            global_level, log_levels_by_target, print_stacktrace=print_stacktrace
+        ):
             yield
     finally:
         sys.stdin, sys.stdout, sys.stderr = original_stdin, original_stdout, original_stderr

{"golden_diff": "diff --git a/src/python/pants/init/logging.py b/src/python/pants/init/logging.py\n--- a/src/python/pants/init/logging.py\n+++ b/src/python/pants/init/logging.py\n@@ -99,7 +99,9 @@\n \n \n @contextmanager\n-def _python_logging_setup(level: LogLevel, *, print_stacktrace: bool) -> Iterator[None]:\n+def _python_logging_setup(\n+    level: LogLevel, log_levels_by_target: dict[str, LogLevel], *, print_stacktrace: bool\n+) -> Iterator[None]:\n     \"\"\"Installs a root Python logger that routes all logging through a Rust logger.\"\"\"\n \n     def trace_fn(self, message, *args, **kwargs):\n@@ -130,6 +132,9 @@\n         logger.addHandler(handler)\n         level.set_level_for(logger)\n \n+        for key, level in log_levels_by_target.items():\n+            level.set_level_for(logging.getLogger(key))\n+\n         if logger.isEnabledFor(LogLevel.TRACE.level):\n             http.client.HTTPConnection.debuglevel = 1  # type: ignore[attr-defined]\n             requests_logger = logging.getLogger(\"requests.packages.urllib3\")\n@@ -213,7 +218,9 @@\n \n         sys.__stdin__, sys.__stdout__, sys.__stderr__ = sys.stdin, sys.stdout, sys.stderr\n         # Install a Python logger that will route through the Rust logger.\n-        with _python_logging_setup(global_level, print_stacktrace=print_stacktrace):\n+        with _python_logging_setup(\n+            global_level, log_levels_by_target, print_stacktrace=print_stacktrace\n+        ):\n             yield\n     finally:\n         sys.stdin, sys.stdout, sys.stderr = original_stdin, original_stdout, original_stderr\n", "issue": "`log_level_by_target` doesn't seem to work well for Python targets\n**Describe the bug**\r\nUsing the following:\r\n```\r\nlevel = \"warn\"\r\nlog_levels_by_target = {\"pants.goal.stats_aggregator\" = \"info\"}\r\n```\r\nwhen running `./pants --stats-log ...` I don't see the stats being logged.\r\n\r\n**Pants version**\r\n`main`\r\n\r\n**OS**\r\nUbuntu\r\n\r\n**Additional info**\r\nI suspect this is because `log_levels_by_target` isn't used to configure the Python logger, and therefore the python logger for any Python module is configured to use the level set by `level`.\r\n\r\nThis can be seen by inspecting the logger in `src/python/pants/goal/stats_aggregator.py` which is set to the level `WARN`. Therefore I assume the log call never gets forwarded to the Rust-implemented handler, and therefore `log_levels_by_target` isn't considered.\r\n\n", "before_files": [{"content": "# Copyright 2018 Pants project contributors (see CONTRIBUTORS.md).\n# Licensed under the Apache License, Version 2.0 (see LICENSE).\n\nfrom __future__ import annotations\n\nimport http.client\nimport locale\nimport logging\nimport sys\nfrom contextlib import contextmanager\nfrom io import BufferedReader, TextIOWrapper\nfrom logging import Formatter, LogRecord, StreamHandler\nfrom pathlib import PurePath\nfrom typing import Iterator\n\nimport pants.util.logging as pants_logging\nfrom pants.engine.internals import native_engine\nfrom pants.option.option_value_container import OptionValueContainer\nfrom pants.util.dirutil import safe_mkdir_for\nfrom pants.util.docutil import doc_url\nfrom pants.util.logging import LogLevel\nfrom pants.util.strutil import strip_prefix\n\n# Although logging supports the WARN level, its not documented and could conceivably be yanked.\n# Since pants has supported 'warn' since inception, leave the 'warn' choice as-is but explicitly\n# setup a 'WARN' logging level name that maps to 'WARNING'.\nlogging.addLevelName(logging.WARNING, \"WARN\")\nlogging.addLevelName(pants_logging.TRACE, \"TRACE\")\n\n\nclass _NativeHandler(StreamHandler):\n    \"\"\"This class is installed as a Python logging module handler (using the logging.addHandler\n    method) and proxies logs to the Rust logging infrastructure.\"\"\"\n\n    def emit(self, record: LogRecord) -> None:\n        native_engine.write_log(self.format(record), record.levelno, record.name)\n\n    def flush(self) -> None:\n        native_engine.flush_log()\n\n\nclass _ExceptionFormatter(Formatter):\n    \"\"\"Possibly render the stacktrace and possibly give debug hints, based on global options.\"\"\"\n\n    def __init__(self, level: LogLevel, *, print_stacktrace: bool) -> None:\n        super().__init__(None)\n        self.level = level\n        self.print_stacktrace = print_stacktrace\n\n    def formatException(self, exc_info):\n        stacktrace = super().formatException(exc_info) if self.print_stacktrace else \"\"\n\n        debug_instructions = []\n        if not self.print_stacktrace:\n            debug_instructions.append(\"--print-stacktrace for more error details\")\n        if self.level not in {LogLevel.DEBUG, LogLevel.TRACE}:\n            debug_instructions.append(\"-ldebug for more logs\")\n        debug_instructions = (\n            f\"Use {' and/or '.join(debug_instructions)}. \" if debug_instructions else \"\"\n        )\n\n        return (\n            f\"{stacktrace}\\n\\n{debug_instructions}\\nSee {doc_url('troubleshooting')} for common \"\n            f\"issues.\\nConsider reaching out for help: {doc_url('getting-help')}\\n\"\n        )\n\n\n@contextmanager\ndef stdio_destination(stdin_fileno: int, stdout_fileno: int, stderr_fileno: int) -> Iterator[None]:\n    \"\"\"Sets a destination for both logging and stdio: must be called after `initialize_stdio`.\n\n    After `initialize_stdio` and outside of this contextmanager, the default stdio destination is\n    the pants.log. But inside of this block, all engine \"tasks\"/@rules that are spawned will have\n    thread/task-local state that directs their IO to the given destination. When the contextmanager\n    exits all tasks will be restored to the default destination (regardless of whether they have\n    completed).\n    \"\"\"\n    if not logging.getLogger(None).handlers:\n        raise AssertionError(\"stdio_destination should only be called after initialize_stdio.\")\n\n    native_engine.stdio_thread_console_set(stdin_fileno, stdout_fileno, stderr_fileno)\n    try:\n        yield\n    finally:\n        native_engine.stdio_thread_console_clear()\n\n\ndef stdio_destination_use_color(use_color: bool) -> None:\n    \"\"\"Sets a color mode for the current thread's destination.\n\n    True or false force color to be used or not used: None causes TTY detection to decide whether\n    color will be used.\n\n    NB: This method is independent from either `stdio_destination` or `initialize_stdio` because\n    we cannot decide whether to use color for a particular destination until it is open AND we have\n    parsed options for the relevant connection.\n    \"\"\"\n    native_engine.stdio_thread_console_color_mode_set(use_color)\n\n\n@contextmanager\ndef _python_logging_setup(level: LogLevel, *, print_stacktrace: bool) -> Iterator[None]:\n    \"\"\"Installs a root Python logger that routes all logging through a Rust logger.\"\"\"\n\n    def trace_fn(self, message, *args, **kwargs):\n        if self.isEnabledFor(LogLevel.TRACE.level):\n            self._log(LogLevel.TRACE.level, message, *args, **kwargs)\n\n    logging.Logger.trace = trace_fn  # type: ignore[attr-defined]\n    logger = logging.getLogger(None)\n\n    def clear_logging_handlers():\n        handlers = tuple(logger.handlers)\n        for handler in handlers:\n            logger.removeHandler(handler)\n        return handlers\n\n    def set_logging_handlers(handlers):\n        for handler in handlers:\n            logger.addHandler(handler)\n\n    # Remove existing handlers, and restore them afterward.\n    handlers = clear_logging_handlers()\n    try:\n        # This routes warnings through our loggers instead of straight to raw stderr.\n        logging.captureWarnings(True)\n        handler = _NativeHandler()\n        exc_formatter = _ExceptionFormatter(level, print_stacktrace=print_stacktrace)\n        handler.setFormatter(exc_formatter)\n        logger.addHandler(handler)\n        level.set_level_for(logger)\n\n        if logger.isEnabledFor(LogLevel.TRACE.level):\n            http.client.HTTPConnection.debuglevel = 1  # type: ignore[attr-defined]\n            requests_logger = logging.getLogger(\"requests.packages.urllib3\")\n            LogLevel.TRACE.set_level_for(requests_logger)\n            requests_logger.propagate = True\n\n        yield\n    finally:\n        clear_logging_handlers()\n        set_logging_handlers(handlers)\n\n\n@contextmanager\ndef initialize_stdio(global_bootstrap_options: OptionValueContainer) -> Iterator[None]:\n    \"\"\"Mutates sys.std* and logging to route stdio for a Pants process to thread local destinations.\n\n    In this context, `sys.std*` and logging handlers will route through Rust code that uses\n    thread-local information to decide whether to write to a file, or to stdio file handles.\n\n    To control the stdio destination set by this method, use the `stdio_destination` context manager.\n\n    This is called in two different processes:\n    * PantsRunner, after it has determined that LocalPantsRunner will be running in process, and\n      immediately before setting a `stdio_destination` for the remainder of the run.\n    * PantsDaemon, immediately on startup. The process will then default to sending stdio to the log\n      until client connections arrive, at which point `stdio_destination` is used per-connection.\n    \"\"\"\n    with initialize_stdio_raw(\n        global_bootstrap_options.level,\n        global_bootstrap_options.log_show_rust_3rdparty,\n        global_bootstrap_options.show_log_target,\n        _get_log_levels_by_target(global_bootstrap_options),\n        global_bootstrap_options.print_stacktrace,\n        global_bootstrap_options.ignore_warnings,\n        global_bootstrap_options.pants_workdir,\n    ):\n        yield\n\n\n@contextmanager\ndef initialize_stdio_raw(\n    global_level: LogLevel,\n    log_show_rust_3rdparty: bool,\n    show_target: bool,\n    log_levels_by_target: dict[str, LogLevel],\n    print_stacktrace: bool,\n    ignore_warnings: list[str],\n    pants_workdir: str,\n) -> Iterator[None]:\n    literal_filters = []\n    regex_filters = []\n    for filt in ignore_warnings:\n        if filt.startswith(\"$regex$\"):\n            regex_filters.append(strip_prefix(filt, \"$regex$\"))\n        else:\n            literal_filters.append(filt)\n\n    # Set the pants log destination.\n    log_path = str(pants_log_path(PurePath(pants_workdir)))\n    safe_mkdir_for(log_path)\n\n    # Initialize thread-local stdio, and replace sys.std* with proxies.\n    original_stdin, original_stdout, original_stderr = sys.stdin, sys.stdout, sys.stderr\n    try:\n        raw_stdin, sys.stdout, sys.stderr = native_engine.stdio_initialize(\n            global_level.level,\n            log_show_rust_3rdparty,\n            show_target,\n            {k: v.level for k, v in log_levels_by_target.items()},\n            tuple(literal_filters),\n            tuple(regex_filters),\n            log_path,\n        )\n        sys.stdin = TextIOWrapper(\n            BufferedReader(raw_stdin),\n            # NB: We set the default encoding explicitly to bypass logic in the TextIOWrapper\n            # constructor that would poke the underlying file (which is not valid until a\n            # `stdio_destination` is set).\n            encoding=locale.getpreferredencoding(False),\n        )\n\n        sys.__stdin__, sys.__stdout__, sys.__stderr__ = sys.stdin, sys.stdout, sys.stderr\n        # Install a Python logger that will route through the Rust logger.\n        with _python_logging_setup(global_level, print_stacktrace=print_stacktrace):\n            yield\n    finally:\n        sys.stdin, sys.stdout, sys.stderr = original_stdin, original_stdout, original_stderr\n        sys.__stdin__, sys.__stdout__, sys.__stderr__ = sys.stdin, sys.stdout, sys.stderr\n\n\ndef pants_log_path(workdir: PurePath) -> PurePath:\n    \"\"\"Given the path of the workdir, returns the `pants.log` path.\"\"\"\n    return workdir / \"pants.log\"\n\n\ndef _get_log_levels_by_target(\n    global_bootstrap_options: OptionValueContainer,\n) -> dict[str, LogLevel]:\n    raw_levels = global_bootstrap_options.log_levels_by_target\n    levels: dict[str, LogLevel] = {}\n    for key, value in raw_levels.items():\n        if not isinstance(key, str):\n            raise ValueError(\n                \"Keys for log_domain_levels must be strings, but was given the key: {key} with type {type(key)}.\"\n            )\n        if not isinstance(value, str):\n            raise ValueError(\n                \"Values for log_domain_levels must be strings, but was given the value: {value} with type {type(value)}.\"\n            )\n        log_level = LogLevel[value.upper()]\n        levels[key] = log_level\n    return levels\n", "path": "src/python/pants/init/logging.py"}], "after_files": [{"content": "# Copyright 2018 Pants project contributors (see CONTRIBUTORS.md).\n# Licensed under the Apache License, Version 2.0 (see LICENSE).\n\nfrom __future__ import annotations\n\nimport http.client\nimport locale\nimport logging\nimport sys\nfrom contextlib import contextmanager\nfrom io import BufferedReader, TextIOWrapper\nfrom logging import Formatter, LogRecord, StreamHandler\nfrom pathlib import PurePath\nfrom typing import Iterator\n\nimport pants.util.logging as pants_logging\nfrom pants.engine.internals import native_engine\nfrom pants.option.option_value_container import OptionValueContainer\nfrom pants.util.dirutil import safe_mkdir_for\nfrom pants.util.docutil import doc_url\nfrom pants.util.logging import LogLevel\nfrom pants.util.strutil import strip_prefix\n\n# Although logging supports the WARN level, its not documented and could conceivably be yanked.\n# Since pants has supported 'warn' since inception, leave the 'warn' choice as-is but explicitly\n# setup a 'WARN' logging level name that maps to 'WARNING'.\nlogging.addLevelName(logging.WARNING, \"WARN\")\nlogging.addLevelName(pants_logging.TRACE, \"TRACE\")\n\n\nclass _NativeHandler(StreamHandler):\n    \"\"\"This class is installed as a Python logging module handler (using the logging.addHandler\n    method) and proxies logs to the Rust logging infrastructure.\"\"\"\n\n    def emit(self, record: LogRecord) -> None:\n        native_engine.write_log(self.format(record), record.levelno, record.name)\n\n    def flush(self) -> None:\n        native_engine.flush_log()\n\n\nclass _ExceptionFormatter(Formatter):\n    \"\"\"Possibly render the stacktrace and possibly give debug hints, based on global options.\"\"\"\n\n    def __init__(self, level: LogLevel, *, print_stacktrace: bool) -> None:\n        super().__init__(None)\n        self.level = level\n        self.print_stacktrace = print_stacktrace\n\n    def formatException(self, exc_info):\n        stacktrace = super().formatException(exc_info) if self.print_stacktrace else \"\"\n\n        debug_instructions = []\n        if not self.print_stacktrace:\n            debug_instructions.append(\"--print-stacktrace for more error details\")\n        if self.level not in {LogLevel.DEBUG, LogLevel.TRACE}:\n            debug_instructions.append(\"-ldebug for more logs\")\n        debug_instructions = (\n            f\"Use {' and/or '.join(debug_instructions)}. \" if debug_instructions else \"\"\n        )\n\n        return (\n            f\"{stacktrace}\\n\\n{debug_instructions}\\nSee {doc_url('troubleshooting')} for common \"\n            f\"issues.\\nConsider reaching out for help: {doc_url('getting-help')}\\n\"\n        )\n\n\n@contextmanager\ndef stdio_destination(stdin_fileno: int, stdout_fileno: int, stderr_fileno: int) -> Iterator[None]:\n    \"\"\"Sets a destination for both logging and stdio: must be called after `initialize_stdio`.\n\n    After `initialize_stdio` and outside of this contextmanager, the default stdio destination is\n    the pants.log. But inside of this block, all engine \"tasks\"/@rules that are spawned will have\n    thread/task-local state that directs their IO to the given destination. When the contextmanager\n    exits all tasks will be restored to the default destination (regardless of whether they have\n    completed).\n    \"\"\"\n    if not logging.getLogger(None).handlers:\n        raise AssertionError(\"stdio_destination should only be called after initialize_stdio.\")\n\n    native_engine.stdio_thread_console_set(stdin_fileno, stdout_fileno, stderr_fileno)\n    try:\n        yield\n    finally:\n        native_engine.stdio_thread_console_clear()\n\n\ndef stdio_destination_use_color(use_color: bool) -> None:\n    \"\"\"Sets a color mode for the current thread's destination.\n\n    True or false force color to be used or not used: None causes TTY detection to decide whether\n    color will be used.\n\n    NB: This method is independent from either `stdio_destination` or `initialize_stdio` because\n    we cannot decide whether to use color for a particular destination until it is open AND we have\n    parsed options for the relevant connection.\n    \"\"\"\n    native_engine.stdio_thread_console_color_mode_set(use_color)\n\n\n@contextmanager\ndef _python_logging_setup(\n    level: LogLevel, log_levels_by_target: dict[str, LogLevel], *, print_stacktrace: bool\n) -> Iterator[None]:\n    \"\"\"Installs a root Python logger that routes all logging through a Rust logger.\"\"\"\n\n    def trace_fn(self, message, *args, **kwargs):\n        if self.isEnabledFor(LogLevel.TRACE.level):\n            self._log(LogLevel.TRACE.level, message, *args, **kwargs)\n\n    logging.Logger.trace = trace_fn  # type: ignore[attr-defined]\n    logger = logging.getLogger(None)\n\n    def clear_logging_handlers():\n        handlers = tuple(logger.handlers)\n        for handler in handlers:\n            logger.removeHandler(handler)\n        return handlers\n\n    def set_logging_handlers(handlers):\n        for handler in handlers:\n            logger.addHandler(handler)\n\n    # Remove existing handlers, and restore them afterward.\n    handlers = clear_logging_handlers()\n    try:\n        # This routes warnings through our loggers instead of straight to raw stderr.\n        logging.captureWarnings(True)\n        handler = _NativeHandler()\n        exc_formatter = _ExceptionFormatter(level, print_stacktrace=print_stacktrace)\n        handler.setFormatter(exc_formatter)\n        logger.addHandler(handler)\n        level.set_level_for(logger)\n\n        for key, level in log_levels_by_target.items():\n            level.set_level_for(logging.getLogger(key))\n\n        if logger.isEnabledFor(LogLevel.TRACE.level):\n            http.client.HTTPConnection.debuglevel = 1  # type: ignore[attr-defined]\n            requests_logger = logging.getLogger(\"requests.packages.urllib3\")\n            LogLevel.TRACE.set_level_for(requests_logger)\n            requests_logger.propagate = True\n\n        yield\n    finally:\n        clear_logging_handlers()\n        set_logging_handlers(handlers)\n\n\n@contextmanager\ndef initialize_stdio(global_bootstrap_options: OptionValueContainer) -> Iterator[None]:\n    \"\"\"Mutates sys.std* and logging to route stdio for a Pants process to thread local destinations.\n\n    In this context, `sys.std*` and logging handlers will route through Rust code that uses\n    thread-local information to decide whether to write to a file, or to stdio file handles.\n\n    To control the stdio destination set by this method, use the `stdio_destination` context manager.\n\n    This is called in two different processes:\n    * PantsRunner, after it has determined that LocalPantsRunner will be running in process, and\n      immediately before setting a `stdio_destination` for the remainder of the run.\n    * PantsDaemon, immediately on startup. The process will then default to sending stdio to the log\n      until client connections arrive, at which point `stdio_destination` is used per-connection.\n    \"\"\"\n    with initialize_stdio_raw(\n        global_bootstrap_options.level,\n        global_bootstrap_options.log_show_rust_3rdparty,\n        global_bootstrap_options.show_log_target,\n        _get_log_levels_by_target(global_bootstrap_options),\n        global_bootstrap_options.print_stacktrace,\n        global_bootstrap_options.ignore_warnings,\n        global_bootstrap_options.pants_workdir,\n    ):\n        yield\n\n\n@contextmanager\ndef initialize_stdio_raw(\n    global_level: LogLevel,\n    log_show_rust_3rdparty: bool,\n    show_target: bool,\n    log_levels_by_target: dict[str, LogLevel],\n    print_stacktrace: bool,\n    ignore_warnings: list[str],\n    pants_workdir: str,\n) -> Iterator[None]:\n    literal_filters = []\n    regex_filters = []\n    for filt in ignore_warnings:\n        if filt.startswith(\"$regex$\"):\n            regex_filters.append(strip_prefix(filt, \"$regex$\"))\n        else:\n            literal_filters.append(filt)\n\n    # Set the pants log destination.\n    log_path = str(pants_log_path(PurePath(pants_workdir)))\n    safe_mkdir_for(log_path)\n\n    # Initialize thread-local stdio, and replace sys.std* with proxies.\n    original_stdin, original_stdout, original_stderr = sys.stdin, sys.stdout, sys.stderr\n    try:\n        raw_stdin, sys.stdout, sys.stderr = native_engine.stdio_initialize(\n            global_level.level,\n            log_show_rust_3rdparty,\n            show_target,\n            {k: v.level for k, v in log_levels_by_target.items()},\n            tuple(literal_filters),\n            tuple(regex_filters),\n            log_path,\n        )\n        sys.stdin = TextIOWrapper(\n            BufferedReader(raw_stdin),\n            # NB: We set the default encoding explicitly to bypass logic in the TextIOWrapper\n            # constructor that would poke the underlying file (which is not valid until a\n            # `stdio_destination` is set).\n            encoding=locale.getpreferredencoding(False),\n        )\n\n        sys.__stdin__, sys.__stdout__, sys.__stderr__ = sys.stdin, sys.stdout, sys.stderr\n        # Install a Python logger that will route through the Rust logger.\n        with _python_logging_setup(\n            global_level, log_levels_by_target, print_stacktrace=print_stacktrace\n        ):\n            yield\n    finally:\n        sys.stdin, sys.stdout, sys.stderr = original_stdin, original_stdout, original_stderr\n        sys.__stdin__, sys.__stdout__, sys.__stderr__ = sys.stdin, sys.stdout, sys.stderr\n\n\ndef pants_log_path(workdir: PurePath) -> PurePath:\n    \"\"\"Given the path of the workdir, returns the `pants.log` path.\"\"\"\n    return workdir / \"pants.log\"\n\n\ndef _get_log_levels_by_target(\n    global_bootstrap_options: OptionValueContainer,\n) -> dict[str, LogLevel]:\n    raw_levels = global_bootstrap_options.log_levels_by_target\n    levels: dict[str, LogLevel] = {}\n    for key, value in raw_levels.items():\n        if not isinstance(key, str):\n            raise ValueError(\n                \"Keys for log_domain_levels must be strings, but was given the key: {key} with type {type(key)}.\"\n            )\n        if not isinstance(value, str):\n            raise ValueError(\n                \"Values for log_domain_levels must be strings, but was given the value: {value} with type {type(value)}.\"\n            )\n        log_level = LogLevel[value.upper()]\n        levels[key] = log_level\n    return levels\n", "path": "src/python/pants/init/logging.py"}]}