Return the async call graph as a string for `future`. If `future` is not provided, format the call graph for the current task.
(
future: futures.Future | None = None,
/,
*,
depth: int = 1,
limit: int | None = None,
)
| 183 | |
| 184 | |
| 185 | def format_call_graph( |
| 186 | future: futures.Future | None = None, |
| 187 | /, |
| 188 | *, |
| 189 | depth: int = 1, |
| 190 | limit: int | None = None, |
| 191 | ) -> str: |
| 192 | """Return the async call graph as a string for `future`. |
| 193 | |
| 194 | If `future` is not provided, format the call graph for the current task. |
| 195 | """ |
| 196 | |
| 197 | def render_level(st: FutureCallGraph, buf: list[str], level: int) -> None: |
| 198 | def add_line(line: str) -> None: |
| 199 | buf.append(level * ' ' + line) |
| 200 | |
| 201 | if isinstance(st.future, tasks.Task): |
| 202 | add_line( |
| 203 | f'* Task(name={st.future.get_name()!r}, id={id(st.future):#x})' |
| 204 | ) |
| 205 | else: |
| 206 | add_line( |
| 207 | f'* Future(id={id(st.future):#x})' |
| 208 | ) |
| 209 | |
| 210 | if st.call_stack: |
| 211 | add_line( |
| 212 | f' + Call stack:' |
| 213 | ) |
| 214 | for ste in st.call_stack: |
| 215 | f = ste.frame |
| 216 | |
| 217 | if f.f_generator is None: |
| 218 | f = ste.frame |
| 219 | add_line( |
| 220 | f' | File {f.f_code.co_filename!r},' |
| 221 | f' line {f.f_lineno}, in' |
| 222 | f' {f.f_code.co_qualname}()' |
| 223 | ) |
| 224 | else: |
| 225 | c = f.f_generator |
| 226 | |
| 227 | try: |
| 228 | f = c.cr_frame |
| 229 | code = c.cr_code |
| 230 | tag = 'async' |
| 231 | except AttributeError: |
| 232 | try: |
| 233 | f = c.ag_frame |
| 234 | code = c.ag_code |
| 235 | tag = 'async generator' |
| 236 | except AttributeError: |
| 237 | f = c.gi_frame |
| 238 | code = c.gi_code |
| 239 | tag = 'generator' |
| 240 | |
| 241 | add_line( |
| 242 | f' | File {f.f_code.co_filename!r},' |
no test coverage detected
searching dependent graphs…