Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 46 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -351,6 +351,52 @@ def classify(status):
$ python -m pyflowchart example_match.py -f classify
```

### try/except/else/finally (Beta)

> ⚠️ **Beta feature:** `try`/`except` support is still in beta and may not work correctly in all cases.

PyFlowchart translates `try`/`except`/`else`/`finally` blocks into a structured flowchart that shows all exception-handling paths.

```python
# example_try.py
def fetch(url):
try:
data = requests.get(url)
except Timeout:
data = cached()
except Exception as e:
log(e)
else:
process(data)
finally:
close()
```

```sh
$ python -m pyflowchart example_try.py -f fetch
```

The generated flowchart represents the following structure:

```
[try body]
exception raised? ──no──▶ [else body]
│ yes │
▼ │
except Timeout? ──yes──▶ [handler body]
│ no │
except Exception as e? ──yes──▶ [handler body]
│ no (unhandled) │
└──────────────────────────►┤
[finally body]
```

Each `except` clause is rendered as a condition diamond. The `else` branch is taken when no exception is raised. All paths — handled exceptions, unhandled exceptions, and the no-exception path — converge into the `finally` block. When the `try` body contains multiple statements they are folded into a single operation node so that the `exception raised?` diamond covers the whole block; for clarity it is recommended to keep `try` bodies minimal (ideally a single statement).

Python 3.11+ `except*` (ExceptionGroup) blocks are dispatched through the same mechanism.

### output html and images

Pass `-o output.html` to write the flowchart directly to an HTML file:
Expand Down
46 changes: 46 additions & 0 deletions README_zh-CN.md
Original file line number Diff line number Diff line change
Expand Up @@ -349,6 +349,52 @@ def classify(status):
$ python -m pyflowchart example_match.py -f classify
```

### try/except/else/finally(Beta)

> ⚠️ **Beta 功能:** `try`/`except` 支持仍处于测试阶段,在某些情况下可能无法正常工作。

PyFlowchart 将 `try`/`except`/`else`/`finally` 语句块翻译为结构化流程图,呈现所有异常处理路径。

```python
# example_try.py
def fetch(url):
try:
data = requests.get(url)
except Timeout:
data = cached()
except Exception as e:
log(e)
else:
process(data)
finally:
close()
```

```sh
$ python -m pyflowchart example_try.py -f fetch
```

生成的流程图结构如下:

```
[try 语句体]
是否发生异常?──否──▶ [else 语句体]
│ 是 │
▼ │
except Timeout?──是──▶ [处理代码]
│ 否 │
except Exception as e?──是──▶ [处理代码]
│ 否(未处理) │
└──────────────────────►┤
[finally 语句体]
```

每个 `except` 子句被渲染为一个条件菱形节点。当没有异常被触发时,走 `else` 分支。无论是已处理异常、未处理异常还是无异常路径,最终都会汇聚到 `finally` 块。当 `try` 语句体包含多条语句时,它们会被折叠为单一操作节点,以确保整个语句块都被 `exception raised?` 菱形覆盖;为保持清晰,建议尽量保持 `try` 语句体简短(最理想是只有一条语句)。

Python 3.11+ 的 `except*`(ExceptionGroup)块也通过相同机制处理。

### 输出 HTML 与图片

传入 `-o output.html` 可将流程图直接写入 HTML 文件:
Expand Down
3 changes: 3 additions & 0 deletions pyflowchart/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,9 @@
"Return",
"Match",
"MatchCase",
"Try",
"TryExceptCondition",
"ExceptHandlerCondition",
# Parsing
"ParseProcessGraph",
"parse",
Expand Down
168 changes: 166 additions & 2 deletions pyflowchart/ast_node.py
Original file line number Diff line number Diff line change
Expand Up @@ -885,14 +885,172 @@ def simplify(self) -> None:
if sys.version_info < (3, 10):
Match = CommonOperation


###########
# Try #
###########

class TryExceptCondition(ConditionNode):
"""ConditionNode representing 'exception raised?' in a try/except block."""

def __init__(self):
ConditionNode.__init__(self, cond="exception raised?")

def fc_connection(self) -> str:
return ""


class ExceptHandlerCondition(ConditionNode):
"""ConditionNode for each except clause: 'except {ExcType}?'"""

def __init__(self, ast_handler: _ast.ExceptHandler):
ConditionNode.__init__(self, cond=self._cond_text(ast_handler))

@staticmethod
def _cond_text(handler: _ast.ExceptHandler) -> str:
if handler.type is None:
return "except"
type_name = astunparse.unparse(handler.type).strip()
if handler.name:
return f"except {type_name} as {handler.name}"
return f"except {type_name}"

def fc_connection(self) -> str:
return ""


class Try(NodesGroup, AstNode):
"""
Try is an AstNode for _ast.Try (try/except statements in Python source code).

Flowchart structure::

[try body]
exception raised? → no → [else body (if any)]
↓ yes
except {Type1}? → yes → [handler1 body]
↓ no
except {Type2}? → yes → [handler2 body]
↓ no
(unhandled)

[finally body] ← all paths converge here

Addresses: https://github.com/cdfmlr/pyflowchart/issues/18
"""

def __init__(self, ast_try: _ast.Try, **kwargs):
AstNode.__init__(self, ast_try, **kwargs)

# "exception raised?" condition node
self.exc_cond = TryExceptCondition()

# parse try body
# When there are multiple statements in the try body we fold them into a
# single OperationNode. Expanding them as separate nodes would only
# connect the *last* statement to the "exception raised?" diamond,
# leaving all preceding statements outside the exception-handling scope.
if len(ast_try.body) > 1:
body_text = '\n'.join(
Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@copilot join the lines manually? would it work properly with complex cases? for example, nested condition or loop statements. is this an existing pattern used in the existing code? it's ok to use it if it's an existing pattern or else it should be introduced more carefully.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, astunparse.unparse(stmt) handles complex statements correctly — it's a complete AST→source serialiser and produces valid multi-line output for nested if, for, while, etc. (tested above with both a nested if and a for loop inside the try body).

This is already the established pattern in the codebase: CommonOperation.__init__ does exactly OperationNode.__init__(self, operation=self.ast_to_source()) which is astunparse.unparse(self.ast_object).strip(), and it's the fallback used for any unrecognised statement including complex ones. The new code is just doing the same thing per-statement in the try body and joining the results — no new pattern is introduced.

astunparse.unparse(stmt).strip() for stmt in ast_try.body
)
try_body_node = OperationNode(body_text)
NodesGroup.__init__(self, try_body_node)
try_body_node.connect(self.exc_cond)
else:
try_body = parse(ast_try.body, **kwargs)
if try_body.head is not None:
NodesGroup.__init__(self, try_body.head)
for tail in try_body.tails:
if isinstance(tail, Node):
tail.connect(self.exc_cond)
else:
NodesGroup.__init__(self, self.exc_cond)

# yes-path: except handlers (chained)
self._parse_handlers(ast_try.handlers, **kwargs)

# no-path: else body (runs only when no exception was raised)
self._parse_else(ast_try.orelse, **kwargs)

# finally body — all current tails connect into it
self._parse_finally(ast_try.finalbody, **kwargs)

def _parse_handlers(self, handlers, **kwargs) -> None:
"""Chain except handlers as nested condition nodes on the yes-path of exc_cond."""
if not handlers:
self.exc_cond.connect_yes(None)
self.append_tails(self.exc_cond.connection_yes.next_node)
return

connect_fn = self.exc_cond.connect_yes
last_handler_cond = None
for handler in handlers:
handler_cond = ExceptHandlerCondition(handler)
connect_fn(handler_cond)

body = parse(handler.body, **kwargs)
if body.head is not None:
handler_cond.connect_yes(body.head)
self.extend_tails(body.tails)
else:
handler_cond.connect_yes(None)
self.append_tails(handler_cond.connection_yes.next_node)

connect_fn = handler_cond.connect_no
last_handler_cond = handler_cond

# last handler's no-path: unhandled exception / implicit re-raise
if last_handler_cond is not None:
last_handler_cond.connect_no(None)
self.append_tails(last_handler_cond.connection_no.next_node)

def _parse_else(self, orelse, **kwargs) -> None:
"""Parse the else clause (no-path of exc_cond: no exception raised)."""
if orelse:
else_proc = parse(orelse, **kwargs)
if else_proc.head is not None:
self.exc_cond.connect_no(else_proc.head)
self.extend_tails(else_proc.tails)
return

# no else body: virtual no-connection
self.exc_cond.connect_no(None)
self.append_tails(self.exc_cond.connection_no.next_node)

def _parse_finally(self, finalbody, **kwargs) -> None:
"""Parse the finally clause — all current tails connect into it."""
if not finalbody:
return

finally_proc = parse(finalbody, **kwargs)
if finally_proc.head is None:
return

current_tails = list(self.tails)
self.tails = []
for tail in current_tails:
if isinstance(tail, Node):
tail.connect(finally_proc.head)
self.extend_tails(finally_proc.tails)


# Python 3.11+ introduces TryStar for `except*` (exception groups).
# Its AST structure mirrors _ast.Try, so we reuse the same handler.
_ast_TryStar_t = _ast.AST # placeholder for Python < 3.11
if sys.version_info >= (3, 11):
_ast_TryStar_t = _ast.TryStar


# Sentence: common | func | cond | loop | ctrl
# - func: def
# - cond: if
# - cond: if, try
# - loop: for, while
# - ctrl: break, continue, return, yield, call
# - common: others
# Special sentence: cond | loop | ctrl
# TODO: Try, With
# TODO: With

__func_stmts = {
_ast.FunctionDef: FunctionDef,
Expand All @@ -901,7 +1059,9 @@ def simplify(self) -> None:

__cond_stmts = {
_ast.If: If,
_ast.Try: Try,
# _ast_Match_t: Match, # need to check Python version, handle it later manually.
# _ast_TryStar_t: Try, # need to check Python version, handle it later manually.
}

__loop_stmts = {
Expand Down Expand Up @@ -959,6 +1119,10 @@ def parse(ast_list: List[_ast.AST], **kwargs) -> ParseProcessGraph:
if sys.version_info >= (3, 10) and isinstance(ast_object, _ast_Match_t):
ast_node_class = Match

# special case: TryStar (`except*`) for Python 3.11+
if sys.version_info >= (3, 11) and isinstance(ast_object, _ast_TryStar_t):
ast_node_class = Try

# special case: special stmt as a expr value. e.g. function call
if isinstance(ast_object, _ast.Expr):
if hasattr(ast_object, "value"):
Expand Down
Loading
Loading