Move local imports to module level in sampling profiler

The sampling profiler code had numerous imports placed inside functions
rather than at module level. While deferred imports can reduce startup
time for rarely-used code paths, these imports were in functions called
during normal profiler operation, adding repeated import overhead.

Moving imports to module level follows Python best practices for code
that runs frequently. This makes import dependencies explicit at file
scope and eliminates per-call import lookup costs. The test files also
had redundant local imports that duplicated module-level imports.

Author: pablogsal

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)