Per mpage's suggestion: use traceback style formatring for frames

Author: 1st1

Doc/library/asyncio-stack.rst

@@ -18,7 +18,7 @@ a suspended *future*.
 .. versionadded:: 3.14
 
 
-.. function:: print_call_graph(*, future=None, file=None)
+.. function:: print_call_graph(*, future=None, file=None, depth=1)
 
    Print the async call graph for the current task or the provided
    :class:`Task` or :class:`Future`.
@@ -27,6 +27,10 @@ a suspended *future*.
    If not passed, the current running task will be used. If there's no
    current task, the function returns ``None``.
 
+   If the function is called on *the current task*, the optional
+   keyword-only ``depth`` argument can be used to skip the specified
+   number of frames from top of the stack.
+
    If *file* is not specified the function will print to :data:`sys.stdout`.
 
    **Example:**
@@ -48,19 +52,14 @@ a suspended *future*.
 
    will print::
 
-      * Task(name='Task-2', id=0x105038fe0)
-        + Call stack:
-        | * print_call_graph()
-        |   asyncio/stack.py:231
-        | * async test()
-        |   test.py:4
-        + Awaited by:
-           * Task(name='Task-1', id=0x1050a6060)
-              + Call stack:
-              | * async TaskGroup.__aexit__()
-              |   asyncio/taskgroups.py:107
-              | * async main()
-              |   test.py:7
+      * Task(name='Task-2', id=0x1039f0fe0)
+      + Call stack:
+      |   File 't2.py', line 4, in async test()
+      + Awaited by:
+         * Task(name='Task-1', id=0x103a5e060)
+            + Call stack:
+            |   File 'taskgroups.py', line 107, in async TaskGroup.__aexit__()
+            |   File 't2.py', line 7, in async main()
 
    For rendering the call stack to a string the following pattern
    should be used:
@@ -85,6 +84,10 @@ a suspended *future*.
    If not passed, the current running task will be used. If there's no
    current task, the function returns ``None``.
 
+   If the function is called on *the current task*, the optional
+   keyword-only ``depth`` argument can be used to skip the specified
+   number of frames from top of the stack.
+
    Returns a ``FutureCallGraph`` named tuple:
 
    * ``FutureCallGraph(future, call_graph, awaited_by)``

Lib/asyncio/stack.py

@@ -75,7 +75,11 @@ def _build_stack_for_future(future: any) -> FutureCallGraph:
     return FutureCallGraph(future, st, awaited_by)
 
 
-def capture_call_graph(*, future: any = None) -> FutureCallGraph | None:
+def capture_call_graph(
+    *,
+    future: any = None,
+    depth: int = 1,
+) -> FutureCallGraph | None:
     """Capture async call stack for the current task or the provided Future.
 
     The stack is represented with three data structures:
@@ -103,6 +107,10 @@ def capture_call_graph(*, future: any = None) -> FutureCallGraph | None:
     Receives an optional keyword-only "future" argument. If not passed,
     the current task will be used. If there's no current task, the function
     returns None.
+
+    If "capture_call_graph()" is introspecting *the current task*, the
+    optional keyword-only "depth" argument can be used to skip the specified
+    number of frames from top of the stack.
     """
 
     loop = events._get_running_loop()
@@ -135,7 +143,7 @@ def capture_call_graph(*, future: any = None) -> FutureCallGraph | None:
 
     call_graph: list[FrameCallGraphEntry | CoroutineCallGraphEntry] = []
 
-    f = sys._getframe(1)
+    f = sys._getframe(depth)
     try:
         while f is not None:
             is_async = f.f_generator is not None
@@ -161,7 +169,7 @@ def capture_call_graph(*, future: any = None) -> FutureCallGraph | None:
     return FutureCallGraph(future, call_graph, awaited_by)
 
 
-def print_call_graph(*, future: any = None, file=None) -> None:
+def print_call_graph(*, future: any = None, file=None, depth: int = 1) -> None:
     """Print async call stack for the current task or the provided Future."""
 
     def render_level(st: FutureCallGraph, buf: list[str], level: int):
@@ -185,10 +193,9 @@ def add_line(line: str):
                 if isinstance(ste, FrameCallGraphEntry):
                     f = ste.frame
                     add_line(
-                        f'  | * {f.f_code.co_qualname}()'
-                    )
-                    add_line(
-                        f'  |   {f.f_code.co_filename}:{f.f_lineno}'
+                        f'  |   File {f.f_code.co_filename!r},'
+                        f' line {f.f_lineno}, in'
+                        f' {f.f_code.co_qualname}()'
                     )
                 else:
                     assert isinstance(ste, CoroutineCallGraphEntry)
@@ -209,10 +216,9 @@ def add_line(line: str):
                             tag = 'generator'
 
                     add_line(
-                        f'  | * {tag} {code.co_qualname}()'
-                    )
-                    add_line(
-                        f'  |   {f.f_code.co_filename}:{f.f_lineno}'
+                        f'  |   File {f.f_code.co_filename!r},'
+                        f' line {f.f_lineno}, in'
+                        f' {tag} {code.co_qualname}()'
                     )
 
         if st.awaited_by:
@@ -222,7 +228,7 @@ def add_line(line: str):
             for fut in st.awaited_by:
                 render_level(fut, buf, level + 1)
 
-    stack = capture_call_graph(future=future)
+    stack = capture_call_graph(future=future, depth=depth + 1)
     if stack is None:
         return
 

Lib/test/test_asyncio/test_stack.py

@@ -107,7 +107,7 @@ async def main():
         ])
 
         self.assertIn(
-            '* async TestCallStack.test_stack_tgroup()',
+            ' async TestCallStack.test_stack_tgroup()',
             stack_for_c5[1])