|
| 1 | +"""4XX error classes, which handle exception groups. |
| 2 | +
|
| 3 | +ASYNC400 except-star-invalid-attribute checks for invalid attribute access on except* |
| 4 | +""" |
| 5 | + |
| 6 | +from __future__ import annotations |
| 7 | + |
| 8 | +import ast |
| 9 | +from typing import TYPE_CHECKING, Any |
| 10 | + |
| 11 | +from .flake8asyncvisitor import Flake8AsyncVisitor |
| 12 | +from .helpers import error_class |
| 13 | + |
| 14 | +if TYPE_CHECKING: |
| 15 | + from collections.abc import Mapping |
| 16 | + |
| 17 | +EXCGROUP_ATTRS = ( |
| 18 | + # from ExceptionGroup |
| 19 | + "message", |
| 20 | + "exceptions", |
| 21 | + "subgroup", |
| 22 | + "split", |
| 23 | + "derive", |
| 24 | + # from BaseException |
| 25 | + "args", |
| 26 | + "with_traceback", |
| 27 | + "add_notes", |
| 28 | +) |
| 29 | + |
| 30 | + |
| 31 | +@error_class |
| 32 | +class Visitor4xx(Flake8AsyncVisitor): |
| 33 | + |
| 34 | + error_codes: Mapping[str, str] = { |
| 35 | + "ASYNC400": ( |
| 36 | + "Accessing attribute {} on ExceptionGroup as if it was a bare Exception" |
| 37 | + ) |
| 38 | + } |
| 39 | + |
| 40 | + def __init__(self, *args: Any, **kwargs: Any): |
| 41 | + super().__init__(*args, **kwargs) |
| 42 | + self.exception_groups: list[str] = [] |
| 43 | + self.trystar = False |
| 44 | + |
| 45 | + def visit_TryStar(self, node: ast.TryStar): # type: ignore[name-defined] |
| 46 | + self.save_state(node, "trystar") |
| 47 | + self.trystar = True |
| 48 | + |
| 49 | + def visit_Try(self, node: ast.Try): |
| 50 | + self.save_state(node, "trystar") |
| 51 | + self.trystar = False |
| 52 | + |
| 53 | + def visit_ExceptHandler(self, node: ast.ExceptHandler): |
| 54 | + if not self.trystar or node.name is None: |
| 55 | + return |
| 56 | + self.save_state(node, "exception_groups", copy=True) |
| 57 | + self.exception_groups.append(node.name) |
| 58 | + self.visit_nodes(node.body) |
| 59 | + |
| 60 | + def visit_Attribute(self, node: ast.Attribute): |
| 61 | + if ( |
| 62 | + isinstance(node.value, ast.Name) |
| 63 | + and node.value.id in self.exception_groups |
| 64 | + and node.attr not in EXCGROUP_ATTRS |
| 65 | + and not (node.attr.startswith("__") and node.attr.endswith("__")) |
| 66 | + ): |
| 67 | + self.error(node, node.attr) |
| 68 | + |
| 69 | + def _clear_if_name(self, node: ast.AST | None): |
| 70 | + if isinstance(node, ast.Name) and node.id in self.exception_groups: |
| 71 | + self.exception_groups.remove(node.id) |
| 72 | + |
| 73 | + def _walk_and_clear(self, node: ast.AST | None): |
| 74 | + if node is None: |
| 75 | + return |
| 76 | + for n in ast.walk(node): |
| 77 | + self._clear_if_name(n) |
| 78 | + |
| 79 | + def visit_Assign(self, node: ast.Assign): |
| 80 | + for t in node.targets: |
| 81 | + self._walk_and_clear(t) |
| 82 | + |
| 83 | + def visit_AnnAssign(self, node: ast.AnnAssign): |
| 84 | + self._clear_if_name(node.target) |
| 85 | + |
| 86 | + def visit_withitem(self, node: ast.withitem): |
| 87 | + self._walk_and_clear(node.optional_vars) |
| 88 | + |
| 89 | + def visit_FunctionDef( |
| 90 | + self, node: ast.FunctionDef | ast.AsyncFunctionDef | ast.Lambda |
| 91 | + ): |
| 92 | + self.save_state(node, "exception_groups", "trystar", copy=False) |
| 93 | + self.exception_groups = [] |
| 94 | + |
| 95 | + visit_AsyncFunctionDef = visit_FunctionDef |
| 96 | + visit_Lambda = visit_FunctionDef |
0 commit comments