gh-138122: Move local imports to module level in sampling profiler

Lib/profiling/sampling/binary_reader.py

@@ -1,5 +1,11 @@
 """Thin Python wrapper around C binary reader for profiling data."""
 
+import _remote_debugging
+
+from .gecko_collector import GeckoCollector
+from .stack_collector import FlamegraphCollector, CollapsedStackCollector
+from .pstats_collector import PStatsCollector
+
 
 class BinaryReader:
     """High-performance binary reader using C implementation.
@@ -23,7 +29,6 @@ def __init__(self, filename):
         self._reader = None
 
     def __enter__(self):
-        import _remote_debugging
         self._reader = _remote_debugging.BinaryReader(self.filename)
         return self
 
@@ -99,10 +104,6 @@ def convert_binary_to_format(input_file, output_file, output_format,
     Returns:
         int: Number of samples converted
     """
-    from .gecko_collector import GeckoCollector
-    from .stack_collector import FlamegraphCollector, CollapsedStackCollector
-    from .pstats_collector import PStatsCollector
-
     with BinaryReader(input_file) as reader:
         info = reader.get_info()
         interval = sample_interval_usec or info['sample_interval_us']

Lib/profiling/sampling/cli.py

@@ -35,6 +35,8 @@
     SORT_MODE_NSAMPLES_CUMUL,
 )
 
+from ._child_monitor import ChildProcessMonitor
+
 try:
     from .live_collector import LiveStatsCollector
 except ImportError:
@@ -93,8 +95,6 @@ class CustomFormatter(
 }
 
 def _setup_child_monitor(args, parent_pid):
-    from ._child_monitor import ChildProcessMonitor
-
     # Build CLI args for child profilers (excluding --subprocesses to avoid recursion)
     child_cli_args = _build_child_profiler_args(args)
 
@@ -1123,8 +1123,6 @@ def _handle_live_run(args):
 
 def _handle_replay(args):
     """Handle the 'replay' command - convert binary profile to another format."""
-    import os
-
     if not os.path.exists(args.input_file):
         sys.exit(f"Error: Input file not found: {args.input_file}")
 

Lib/profiling/sampling/heatmap_collector.py

@@ -18,6 +18,7 @@
 from ._css_utils import get_combined_css
 from ._format_utils import fmt
 from .collector import normalize_location, extract_lineno
+from .opcode_utils import get_opcode_info, format_opcode
 from .stack_collector import StackTraceCollector
 
 
@@ -634,8 +635,6 @@ def _get_bytecode_data_for_line(self, filename, lineno):
         Returns:
             List of dicts with instruction info, sorted by samples descending
         """
-        from .opcode_utils import get_opcode_info, format_opcode
-
         key = (filename, lineno)
         opcode_data = self.line_opcodes.get(key, {})
 
@@ -1038,8 +1037,6 @@ def _render_source_with_highlights(self, line_content: str, line_num: int,
         Simple: collect ranges with sample counts, assign each byte position to
         smallest covering range, then emit spans for contiguous runs with sample data.
         """
-        import html as html_module
-
         content = line_content.rstrip('\n')
         if not content:
             return ''
@@ -1062,7 +1059,7 @@ def _render_source_with_highlights(self, line_content: str, line_num: int,
                             range_data[key]['opcodes'].append(opname)
 
         if not range_data:
-            return html_module.escape(content)
+            return html.escape(content)
 
         # For each byte position, find the smallest covering range
         byte_to_range = {}
@@ -1090,7 +1087,7 @@ def _render_source_with_highlights(self, line_content: str, line_num: int,
         def flush_span():
             nonlocal span_chars, current_range
             if span_chars:
-                text = html_module.escape(''.join(span_chars))
+                text = html.escape(''.join(span_chars))
                 if current_range:
                     data = range_data.get(current_range, {'samples': 0, 'opcodes': []})
                     samples = data['samples']
@@ -1104,7 +1101,7 @@ def flush_span():
                                   f'data-samples="{samples}" '
                                   f'data-max-samples="{max_range_samples}" '
                                   f'data-pct="{pct}" '
-                                  f'data-opcodes="{html_module.escape(opcodes)}">{text}</span>')
+                                  f'data-opcodes="{html.escape(opcodes)}">{text}</span>')
                 else:
                     result.append(text)
                 span_chars = []

Lib/profiling/sampling/live_collector/collector.py

@@ -932,8 +932,6 @@ def _show_terminal_size_warning_and_wait(self, height, width):
 
     def _handle_input(self):
         """Handle keyboard input (non-blocking)."""
-        from . import constants
-
         self.display.set_nodelay(True)
         ch = self.display.get_input()
 

Lib/profiling/sampling/live_collector/widgets.py

@@ -31,6 +31,7 @@
     PROFILING_MODE_GIL,
     PROFILING_MODE_WALL,
 )
+from ..opcode_utils import get_opcode_info, format_opcode
 
 
 class Widget(ABC):
@@ -1007,8 +1008,6 @@ def render(self, line, width, **kwargs):
         Returns:
             Next available line number
         """
-        from ..opcode_utils import get_opcode_info, format_opcode
-
         stats_list = kwargs.get("stats_list", [])
         height = kwargs.get("height", 24)
         selected_row = self.collector.selected_row

Lib/profiling/sampling/pstats_collector.py

@@ -1,9 +1,10 @@
 import collections
 import marshal
+import pstats
 
 from _colorize import ANSIColors
 from .collector import Collector, extract_lineno
-from .constants import MICROSECONDS_PER_SECOND
+from .constants import MICROSECONDS_PER_SECOND, PROFILING_MODE_CPU
 
 
 class PstatsCollector(Collector):
@@ -86,9 +87,6 @@ def create_stats(self):
 
     def print_stats(self, sort=-1, limit=None, show_summary=True, mode=None):
         """Print formatted statistics to stdout."""
-        import pstats
-        from .constants import PROFILING_MODE_CPU
-
         # Create stats object
         stats = pstats.SampledStats(self).strip_dirs()
         if not stats.stats:

Lib/profiling/sampling/sample.py

@@ -1,5 +1,6 @@
 import _remote_debugging
 import contextlib
+import curses
 import os
 import statistics
 import sys
@@ -459,8 +460,6 @@ def sample_live(
     Returns:
         The collector with collected samples
     """
-    import curses
-
     # Check if process is alive before doing any heavy initialization
     if not _is_process_running(pid):
         print(f"No samples collected - process {pid} exited before profiling could begin.", file=sys.stderr)

Lib/profiling/sampling/stack_collector.py

@@ -6,6 +6,7 @@
 import linecache
 import os
 import sys
+import sysconfig
 
 from ._css_utils import get_combined_css
 from .collector import Collector, extract_lineno
@@ -244,7 +245,6 @@ def convert_children(children, min_samples):
             }
 
         # Calculate thread status percentages for display
-        import sysconfig
         is_free_threaded = bool(sysconfig.get_config_var("Py_GIL_DISABLED"))
         total_threads = max(1, self.thread_status_counts["total"])
         thread_stats = {

Lib/test/test_profiling/test_sampling_profiler/test_advanced.py

@@ -11,6 +11,8 @@
     import _remote_debugging  # noqa: F401
     import profiling.sampling
     import profiling.sampling.sample
+    from profiling.sampling.pstats_collector import PstatsCollector
+    from profiling.sampling.stack_collector import CollapsedStackCollector
 except ImportError:
     raise unittest.SkipTest(
         "Test only runs when _remote_debugging is available"
@@ -61,7 +63,6 @@ def test_gc_frames_enabled(self):
             io.StringIO() as captured_output,
             mock.patch("sys.stdout", captured_output),
         ):
-            from profiling.sampling.pstats_collector import PstatsCollector
             collector = PstatsCollector(sample_interval_usec=5000, skip_idle=False)
             profiling.sampling.sample.sample(
                 subproc.process.pid,
@@ -88,7 +89,6 @@ def test_gc_frames_disabled(self):
             io.StringIO() as captured_output,
             mock.patch("sys.stdout", captured_output),
         ):
-            from profiling.sampling.pstats_collector import PstatsCollector
             collector = PstatsCollector(sample_interval_usec=5000, skip_idle=False)
             profiling.sampling.sample.sample(
                 subproc.process.pid,
@@ -140,7 +140,6 @@ def test_native_frames_enabled(self):
                 io.StringIO() as captured_output,
                 mock.patch("sys.stdout", captured_output),
             ):
-                from profiling.sampling.stack_collector import CollapsedStackCollector
                 collector = CollapsedStackCollector(1000, skip_idle=False)
                 profiling.sampling.sample.sample(
                     subproc.process.pid,
@@ -176,7 +175,6 @@ def test_native_frames_disabled(self):
             io.StringIO() as captured_output,
             mock.patch("sys.stdout", captured_output),
         ):
-            from profiling.sampling.pstats_collector import PstatsCollector
             collector = PstatsCollector(sample_interval_usec=5000, skip_idle=False)
             profiling.sampling.sample.sample(
                 subproc.process.pid,

Lib/test/test_profiling/test_sampling_profiler/test_async.py

@@ -6,11 +6,14 @@
 3. Stack traversal: _build_linear_stacks() with BFS
 """
 
+import inspect
 import unittest
 
 try:
     import _remote_debugging  # noqa: F401
     from profiling.sampling.pstats_collector import PstatsCollector
+    from profiling.sampling.stack_collector import FlamegraphCollector
+    from profiling.sampling.sample import sample, sample_live, SampleProfiler
 except ImportError:
     raise unittest.SkipTest(
         "Test only runs when _remote_debugging is available"
@@ -561,8 +564,6 @@ class TestFlamegraphCollectorAsync(unittest.TestCase):
 
     def test_flamegraph_with_async_frames(self):
         """Test FlamegraphCollector correctly processes async task frames."""
-        from profiling.sampling.stack_collector import FlamegraphCollector
-
         collector = FlamegraphCollector(sample_interval_usec=1000)
 
         # Build async task tree: Root -> Child
@@ -607,8 +608,6 @@ def test_flamegraph_with_async_frames(self):
 
     def test_flamegraph_with_task_markers(self):
         """Test FlamegraphCollector includes <task> boundary markers."""
-        from profiling.sampling.stack_collector import FlamegraphCollector
-
         collector = FlamegraphCollector(sample_interval_usec=1000)
 
         task = MockTaskInfo(
@@ -643,8 +642,6 @@ def find_task_marker(node, depth=0):
 
     def test_flamegraph_multiple_async_samples(self):
         """Test FlamegraphCollector aggregates multiple async samples correctly."""
-        from profiling.sampling.stack_collector import FlamegraphCollector
-
         collector = FlamegraphCollector(sample_interval_usec=1000)
 
         task = MockTaskInfo(
@@ -675,25 +672,16 @@ class TestAsyncAwareParameterFlow(unittest.TestCase):
 
     def test_sample_function_accepts_async_aware(self):
         """Test that sample() function accepts async_aware parameter."""
-        from profiling.sampling.sample import sample
-        import inspect
-
         sig = inspect.signature(sample)
         self.assertIn("async_aware", sig.parameters)
 
     def test_sample_live_function_accepts_async_aware(self):
         """Test that sample_live() function accepts async_aware parameter."""
-        from profiling.sampling.sample import sample_live
-        import inspect
-
         sig = inspect.signature(sample_live)
         self.assertIn("async_aware", sig.parameters)
 
     def test_sample_profiler_sample_accepts_async_aware(self):
         """Test that SampleProfiler.sample() accepts async_aware parameter."""
-        from profiling.sampling.sample import SampleProfiler
-        import inspect
-
         sig = inspect.signature(SampleProfiler.sample)
         self.assertIn("async_aware", sig.parameters)