fixes to parser for defer

This commit is contained in:
2026-05-13 07:01:45 -04:00
parent f52eff6499
commit 8687207fe5
+45 -42
View File
@@ -1,10 +1,10 @@
import sys import sys
import os
from ast import ( from ast import (
AsyncFunctionDef, AsyncFunctionDef,
FunctionDef, FunctionDef,
) )
from collections import deque from collections import deque
from os.path import isabs
from types import FrameType from types import FrameType
from typing import Any, Callable, Optional, cast from typing import Any, Callable, Optional, cast
@@ -16,21 +16,19 @@ def _is_dunder(name: str) -> bool:
return name.startswith("__") and name.endswith("__") return name.startswith("__") and name.endswith("__")
def _is_excluded_path(filename: str) -> bool: _SITE_PACKAGES = "site-packages"
_VENV = ".venv"
_THIRDPARTY = "thirdparty"
_SRC_WIN = "\\src\\"
_SRC_UNIX = "/src/"
def _is_our_code(filename: str) -> bool:
if not filename: if not filename:
return False
if _SRC_WIN in filename or _SRC_UNIX in filename:
return True return True
return ( return False
filename.startswith(sys.base_exec_prefix)
or filename.startswith(sys.base_prefix)
or filename.startswith(sys.exec_prefix)
or filename.startswith(sys.prefix)
or ".venv" in filename
or filename.startswith("<frozen ")
or filename.startswith("<string>")
or filename.startswith("<pytest ")
or "thirdparty" in filename
or "__future__" in filename
)
class _ParseDefer: class _ParseDefer:
@@ -41,65 +39,70 @@ class _ParseDefer:
self.pending: deque[Executing] = deque() self.pending: deque[Executing] = deque()
def __call__(self, frame: FrameType, event: str, arg: Any): def __call__(self, frame: FrameType, event: str, arg: Any):
try:
self._do_call(frame, event, arg)
except Exception:
pass
return self
def _do_call(self, frame: FrameType, event: str, arg: Any):
self.tracefn(frame, event, arg) self.tracefn(frame, event, arg)
try: try:
filename = frame.f_code.co_filename filename = frame.f_code.co_filename
if not filename:
return self
if _SITE_PACKAGES in filename or _VENV in filename or _THIRDPARTY in filename:
return self
if not _is_our_code(filename):
return self
except Exception: except Exception:
return pass
if _is_excluded_path(filename):
return
if not isabs(filename):
return
if event != "line": if event != "line":
return return self
try: try:
from executing.executing import Executing, Source from executing.executing import Executing, Source
except ImportError: except ImportError:
return return self
try: try:
exc = Source.executing(frame) exc = Source.executing(frame)
except Exception: except Exception:
return return self
if not (stmt := next(iter(exc.statements), None)): if not (stmt := next(iter(exc.statements), None)):
return return self
if isinstance(stmt, (AsyncFunctionDef, FunctionDef)): if isinstance(stmt, (AsyncFunctionDef, FunctionDef)):
if _is_dunder(stmt.name): if _is_dunder(stmt.name):
return return self
self.pending.append(exc) self.pending.append(exc)
return return self
if not self.pending or frame.f_back is not self.pending[-1].frame.f_back: if not self.pending or frame.f_back is not self.pending[-1].frame.f_back:
return return self
stmts = self.pending.pop().statements stmts = self.pending.pop().statements
node = cast(FunctionDef | AsyncFunctionDef, next(iter(stmts))) node = cast(FunctionDef | AsyncFunctionDef, next(iter(stmts)))
fn_name = node.name fn_name = node.name
if fn_name not in frame.f_locals: if fn_name not in frame.f_locals:
return return self
fn = frame.f_locals[fn_name] fn = frame.f_locals[fn_name]
if fn.__module__ in sys.stdlib_module_names: if not callable(fn):
return return self
if fn.__code__.co_freevars: try:
raise FreeVarsError(fn) code = fn.__code__
except AttributeError:
return self
try:
mod = fn.__module__
except AttributeError:
return self
if mod in sys.stdlib_module_names:
return self
if code.co_freevars:
return self
if not (ast := RewriteDefer.transform(node)): if not (ast := RewriteDefer.transform(node)):
return return self
locals = frame.f_locals.copy() locals = frame.f_locals.copy()
del locals[fn_name] del locals[fn_name]
exec(compile(ast, frame.f_code.co_filename, "exec"), frame.f_globals, locals) exec(compile(ast, frame.f_code.co_filename, "exec"), frame.f_globals, locals)
frame.f_locals[fn_name].__code__ = locals[fn_name].__code__ frame.f_locals[fn_name].__code__ = locals[fn_name].__code__
return self