Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Feat/watchpoints #13248

Open
wants to merge 12 commits into
base: main
Choose a base branch
from
1 change: 1 addition & 0 deletions pyrightconfig.stricter.json
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,7 @@
"stubs/tqdm",
"stubs/ttkthemes",
"stubs/vobject",
"stubs/watchpoints",
"stubs/workalendar",
"stubs/wurlitzer",
],
Expand Down
1 change: 1 addition & 0 deletions stubs/watchpoints/@tests/stubtest_allowlist.txt
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
watchpoints.all
2 changes: 2 additions & 0 deletions stubs/watchpoints/METADATA.toml
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
version = "0.2.5"
upstream_repository = "https://github.com/gaogaotiantian/watchpoints"
10 changes: 10 additions & 0 deletions stubs/watchpoints/watchpoints/__init__.pyi
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
from collections.abc import Callable
from typing import Final
from typing_extensions import LiteralString, Unpack

from .watch import Watch

__version__: Final[LiteralString]

watch: Watch
unwatch: Final[Callable[[Unpack[tuple[object, ...]]], None]]
3 changes: 3 additions & 0 deletions stubs/watchpoints/watchpoints/ast_monkey.pyi
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
import ast

def ast_parse_node(node: ast.expr) -> ast.Module: ...
6 changes: 6 additions & 0 deletions stubs/watchpoints/watchpoints/util.pyi
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
import ast
from collections.abc import Iterable
from types import FrameType

def getline(frame: FrameType) -> str: ...
def getargnodes(frame: FrameType) -> Iterable[tuple[ast.expr, str]]: ...
68 changes: 68 additions & 0 deletions stubs/watchpoints/watchpoints/watch.pyi
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
import threading
from _typeshed import SupportsWrite, TraceFunction
from collections.abc import Callable
from pdb import Pdb
from types import FrameType
from typing import Any, Literal, Protocol, TypeVar
from typing_extensions import TypeAlias

from .watch_element import WatchElement

_T = TypeVar("_T")

# Alias used for fields that must always be valid identifiers
# A string `x` counts as a valid identifier if both the following are True
# (1) `x.isidentifier()` evaluates to `True`
# (2) `keyword.iskeyword(x)` evaluates to `False`
_Identifier: TypeAlias = str

class Watch:
# User-defined callbacks passed to `__call__()` or `config()` set as instance variables have arguments of type `Any` to be
# compatible with more precisely-annotated signatures.

custom_printer: Callable[[Any], None] | None
Copy link
Contributor Author

Choose a reason for hiding this comment

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

Several callback types this package have Any as argument type(s), which are assigned with user-defined callbacks. I avoided using object here to be compatible with user-defined callbacks, which may annotate the parameters with more precise types.

  • Printer callbacks: accepts any object and writes to a stream
  • Compare (cmp) callbacks: Accepts any 2 objects and compare their equality
  • copy: Makes a copy of any object
  • when callbacks: Inspect/trace/watch an object when a condition is fulfilled; takes any object and outputs a bool.

Copy link
Member

Choose a reason for hiding this comment

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

That makes sense, but please add a comment to the code explaining this.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Fixed in fcb9691

enable: bool
file: str | SupportsWrite[str] | None
pdb: Pdb | None
pdb_enable: bool
set_lock: threading.Lock
stack_limit: int | None
tracefunc_lock: threading.Lock
tracefunc_stack: list[TraceFunction | None]
watch_list: list[WatchElement]

def __init__(self) -> None: ...
def __call__(
self,
*args: object,
Copy link
Member

Choose a reason for hiding this comment

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

Seems like these are unused, consider using _typeshed.Unused.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

My understanding of the library is that it inspects the expressions of positional-only arguments passed to __call__ (using frames and AST), so while the runtime values don't seem to be used, the expressions passed as *args to __call__ actually do matter.

Passing nothing will cause nothing to be watched, so I'm also tempted to enforce at least 1 argument to be passed here, but let me know if it's still desirable to use _typeshed.Unused.

alias: str = ...,
Copy link
Member

Choose a reason for hiding this comment

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

This one and most of the others support None, since the runtime does kwargs.get("alias", None). Add None to the allowed type of these parameters.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Unlike in watch_element.WatchElement.__init__ and watch_print.WatchPrint.__init__, which receives arguments from watch.Watch which must handle None, I avoided doing this in watch.Watch.__call__ because

  1. watch.Watch is the only public-facing object/method given in the documentation;
  2. The README documentation doesn't describe None as valid arguments to most of Watch.__call__ - the implementation uses it as a sentinel for unpassed arguments;
  3. As per the README, Watch.__call__(stack_limit=None) and Watch.config(stack_limit=None) is where a None actually means something if it is passed.

Let me know if we'd still prefer None to be valid in watch.Watch.__call__ as I understand that it does match the runtime better.

callback: Callable[[FrameType, WatchElement, tuple[str, str, int | None]], None] = ...,
cmp: Callable[[Any, Any], bool] = ..., # User-defined comparison callback; compares 2 arguments of any type
copy: Callable[[_T], _T] = ...,
# User-defined printing callback; writes a string representation of any object to a stream
custom_printer: Callable[[Any], None] = ...,
deepcopy: bool = False,
file: str | SupportsWrite[str] = ...,
stack_limit: int | None = 5,
track: Literal["object", "variable"] = ...,
Copy link
Member

Choose a reason for hiding this comment

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

Suggested change
track: Literal["object", "variable"] = ...,
track: list[Literal["object", "variable"]] = ...,

Or does it also support other kinds of Sequences?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

The documentation describes track to expect only "object" or "variable", and to not pass anything to track if the user wants both "object" and "variable". It's handled internally as a list[Literal["object", "variable"]] if the user doesn't pass anything (see watchpoints.watch_element.WatchElement.track) but I interpret this as an implementation detail.

when: Callable[[Any], bool] = ..., # User-defined callback for conditional watchpoints
) -> None: ...
def config(
self,
*,
callback: Callable[[FrameType, WatchElement, tuple[str, str, int | None]], None] = ...,
pdb: Literal[True] = ...,
file: str | SupportsWrite[str] = ...,
stack_limit: int | None = 5,
custom_printer: Callable[[Any], None] = ..., # User-defined printing callback
) -> None: ...
def install(self, func: _Identifier = "watch") -> None: ...
def restore(self) -> None: ...
def start_trace(self, frame: FrameType) -> None: ...
def stop_trace(self, frame: FrameType) -> None: ...
def tracefunc(self, frame: FrameType, event: str, arg: object) -> _TraceFunc: ...
def uninstall(self, func: _Identifier = "watch") -> None: ...
def unwatch(self, *args: object) -> None: ...

class _TraceFunc(Protocol):
def __call__(self, frame: FrameType, event: str, arg: object) -> _TraceFunc: ...
Copy link
Member

Choose a reason for hiding this comment

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

Should these parameters be positional-only?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

The watchpoints.watch.Watch.tracefunc method returns itself, so I kept the callback's signature the same as the method.

56 changes: 56 additions & 0 deletions stubs/watchpoints/watchpoints/watch_element.pyi
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
import ast
from collections.abc import Callable, Iterable
from types import FrameType
from typing import Any, Literal, TypeVar
from typing_extensions import TypeAlias

from .watch_print import WatchPrint

_T = TypeVar("_T")
_TrackKind: TypeAlias = Literal["object", "variable"] | list[Literal["object", "variable"]]

class WatchElement:
# User-defined callbacks passed to `__init__` set as instance variables have arguments of type `Any` to be
# compatible with more precisely-annotated signatures. These callbacks are passed from `watchpoints.watch.Watch`.

alias: str | None
attr: str | None
cmp: Callable[[Any, Any], bool] | None
copy: Callable[[Any], object] | None # User-defined copy callback
default_alias: str | None
deepcopy: bool
exist: bool
frame: FrameType
localvar: str | None
obj: Any
parent: Any
prev_obj: Any
prev_obj_repr: str
subscr: Any
watch_print: WatchPrint
when: Callable[[Any], bool] | None

def __init__(
self,
frame: FrameType,
node: ast.expr,
*,
alias: str | None = ...,
callback: Callable[[FrameType, WatchElement, tuple[str, str, int | None]], None] | None = ...,
cmp: Callable[[Any, Any], bool] | None = ..., # User-defined comparison callback
copy: Callable[[_T], _T] | None = ...,
deepcopy: bool = False,
default_alias: str | None = ...,
track: _TrackKind = ...,
watch_print: WatchPrint = ...,
when: Callable[[Any], bool] | None = ..., # User-defined callback for conditional watchpoints
) -> None: ...
def belong_to(self, lst: Iterable[object]) -> bool: ...
def changed(self, frame: FrameType) -> tuple[bool, bool]: ...
def obj_changed(self, other: object) -> bool: ...
def same(self, other: object) -> bool: ...
@property
def track(self) -> _TrackKind: ...
@track.setter
def track(self, val: _TrackKind) -> None: ...
def update(self) -> None: ...
24 changes: 24 additions & 0 deletions stubs/watchpoints/watchpoints/watch_print.pyi
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
from _typeshed import SupportsWrite
from collections.abc import Callable
from types import FrameType
from typing import Any

from .watch_element import WatchElement

class WatchPrint:
# User-defined callbacks passed to `__init__` set as instance variables have arguments of type `Any` to be
# compatible with more precisely-annotated signatures. These callbacks are passed from `watchpoints.watch.Watch`.

custom_printer: Callable[[Any], None] | None
file: str | SupportsWrite[str] | None
stack_limit: int | None

def __init__(
self,
file: str | SupportsWrite[str] | None = ...,
stack_limit: int | None = ...,
custom_printer: Callable[[Any], None] | None = ..., # User-defined printing callback
) -> None: ...
def __call__(self, frame: FrameType, elem: WatchElement, exec_info: tuple[str, str, int | None]) -> None: ...
def getsourceline(self, exec_info: tuple[str, str, int | None]) -> str: ...
def printer(self, obj: object) -> None: ...
Loading