Build a list of strings for pretty-print an async call tree. The call tree is produced by `get_all_async_stacks()`, prefixing tasks with `task_emoji` and coroutine frames with `cor_emoji`.
(result, task_emoji="(T)", cor_emoji="")
| 146 | |
| 147 | |
| 148 | def build_async_tree(result, task_emoji="(T)", cor_emoji=""): |
| 149 | """ |
| 150 | Build a list of strings for pretty-print an async call tree. |
| 151 | |
| 152 | The call tree is produced by `get_all_async_stacks()`, prefixing tasks |
| 153 | with `task_emoji` and coroutine frames with `cor_emoji`. |
| 154 | """ |
| 155 | id2name, awaits, task_stacks = _index(result) |
| 156 | g = _task_graph(awaits) |
| 157 | cycles = _find_cycles(g) |
| 158 | if cycles: |
| 159 | raise CycleFoundException(cycles, id2name) |
| 160 | labels, children = _build_tree(id2name, awaits, task_stacks) |
| 161 | |
| 162 | def pretty(node): |
| 163 | flag = task_emoji if node[0] == NodeType.TASK else cor_emoji |
| 164 | return f"{flag} {labels[node]}" |
| 165 | |
| 166 | def render(node, prefix="", last=True, buf=None): |
| 167 | if buf is None: |
| 168 | buf = [] |
| 169 | buf.append(f"{prefix}{'└── ' if last else '├── '}{pretty(node)}") |
| 170 | new_pref = prefix + (" " if last else "│ ") |
| 171 | kids = children.get(node, []) |
| 172 | for i, kid in enumerate(kids): |
| 173 | render(kid, new_pref, i == len(kids) - 1, buf) |
| 174 | return buf |
| 175 | |
| 176 | return [render(root) for root in _roots(labels, children)] |
| 177 | |
| 178 | |
| 179 | def build_task_table(result): |
no test coverage detected
searching dependent graphs…