Refactor evaluation logic to have a separate DAP-unaware object inspection layer.

This commit is contained in:
Pavel Minaev 2023-12-13 11:02:24 -08:00 committed by Pavel Minaev
parent 6b5806c5f4
commit a15a1ce400
5 changed files with 231 additions and 131 deletions

View file

@ -14,12 +14,13 @@ exclude = '''
'''
[tool.pyright]
pythonVersion = "3.8"
include = ["src/**", "tests/**" ]
pythonVersion = "3.12"
include = ["src/**", "tests/**"]
extraPaths = ["src/debugpy/_vendored/pydevd"]
ignore = ["src/debugpy/_vendored/pydevd", "src/debugpy/_version.py"]
executionEnvironments = [
{ root = "src" }, { root = "." }
{ root = "src" },
{ root = "." },
]
[tool.ruff]
@ -28,10 +29,19 @@ executionEnvironments = [
# McCabe complexity (`C901`) by default.
select = ["E", "F"]
ignore = [
"E203", "E221", "E222", "E226", "E261", "E262", "E265", "E266",
"E401", "E402",
"E501",
"E722", "E731"
"E203",
"E221",
"E222",
"E226",
"E261",
"E262",
"E265",
"E266",
"E401",
"E402",
"E501",
"E722",
"E731",
]
# Allow autofix for all enabled rules (when `--fix`) is provided.
@ -40,29 +50,29 @@ unfixable = []
# Exclude a variety of commonly ignored directories.
exclude = [
".bzr",
".direnv",
".eggs",
".git",
".git-rewrite",
".hg",
".mypy_cache",
".nox",
".pants.d",
".pytype",
".ruff_cache",
".svn",
".tox",
".venv",
"__pypackages__",
"_build",
"buck-out",
"build",
"dist",
"node_modules",
"venv",
"versioneer.py",
"src/debugpy/_vendored/pydevd"
".bzr",
".direnv",
".eggs",
".git",
".git-rewrite",
".hg",
".mypy_cache",
".nox",
".pants.d",
".pytype",
".ruff_cache",
".svn",
".tox",
".venv",
"__pypackages__",
"_build",
"buck-out",
"build",
"dist",
"node_modules",
"venv",
"versioneer.py",
"src/debugpy/_vendored/pydevd",
]
per-file-ignores = {}
@ -73,4 +83,4 @@ line-length = 88
dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
# Assume Python 3.8
target-version = "py38"
target-version = "py38"

View file

@ -14,7 +14,7 @@ from debugpy.server import tracing, eval
from debugpy.server.tracing import Breakpoint, StackFrame
class Adapter(object):
class Adapter:
"""Represents the debug adapter connected to this debug server."""
class Capabilities(components.Capabilities):

View file

@ -4,14 +4,12 @@
import threading
from collections.abc import Iterable, Mapping
from itertools import count
from collections.abc import Iterable
from types import FrameType
from typing import ClassVar, Dict, Literal, Self
from debugpy.server import tracing
from debugpy.common import log
from debugpy.server.safe_repr import SafeRepr
from debugpy.server.inspect import inspect
ScopeKind = Literal["global", "nonlocal", "local"]
@ -99,28 +97,6 @@ class Variable(VariableContainer):
self.name = name
self.value = value
if isinstance(value, Mapping):
self._items = self._items_dict
else:
try:
it = iter(value)
except:
it = None
# Use (iter(value) is value) to distinguish iterables from iterators.
if it is not None and it is not value:
self._items = self._items_iterable
@property
def typename(self) -> str:
try:
return type(self.value).__name__
except:
return ""
@property
def repr(self) -> str:
return SafeRepr()(self.value)
def __getstate__(self):
state = super().__getstate__()
state.update(
@ -132,82 +108,23 @@ class Variable(VariableContainer):
)
return state
@property
def typename(self) -> str:
try:
return type(self.value).__name__
except:
return ""
@property
def repr(self) -> str:
return "".join(inspect(self.value).repr())
def variables(self) -> Iterable["Variable"]:
get_name = lambda var: var.name
return [
*sorted(self._attributes(), key=get_name),
*sorted(self._synthetic(), key=get_name),
*self._items(),
]
def _attributes(self) -> Iterable["Variable"]:
# TODO: group class/instance/function/special
try:
names = dir(self.value)
except:
names = []
for name in names:
if name.startswith("__"):
continue
try:
value = getattr(self.value, name)
except BaseException as exc:
value = exc
try:
if hasattr(type(value), "__call__"):
continue
except:
pass
yield Variable(self.frame, name, value)
def _synthetic(self) -> Iterable["Variable"]:
try:
length = len(self.value)
except:
pass
else:
yield Variable(self.frame, "len()", length)
def _items(self) -> Iterable["Variable"]:
return ()
def _items_iterable(self) -> Iterable["Variable"]:
try:
it = iter(self.value)
except:
return
for i in count():
try:
item = next(it)
except StopIteration:
break
except:
log.exception("Error retrieving next item.")
break
yield Variable(self.frame, f"[{i}]", item)
def _items_dict(self) -> Iterable["Variable"]:
try:
keys = self.value.keys()
except:
return
it = iter(keys)
safe_repr = SafeRepr()
while True:
try:
key = next(it)
except StopIteration:
break
except:
break
try:
value = self.value[key]
except BaseException as exc:
value = exc
yield Variable(self.frame, f"[{safe_repr(key)}]", value)
for child in inspect(self.value).children():
yield Variable(self.frame, child.name, child.value)
def evaluate(expr: str, frame_id: int):
def evaluate(expr: str, frame_id: int) -> Variable:
from debugpy.server.tracing import StackFrame
frame = StackFrame.get(frame_id)

View file

@ -0,0 +1,87 @@
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See LICENSE in the project root
# for license information.
"""
Object inspection: rendering values, enumerating children etc.
"""
from typing import Iterable
class ChildObject:
name: str
value: object
def __init__(self, value: object):
self.value = value
@property
def name(self) -> str:
raise NotImplementedError
def expr(self, obj: object) -> str:
raise NotImplementedError
class ChildAttribute(ChildObject):
name: str
def __init__(self, name: str, value: object):
super().__init__(value)
self.name = name
def expr(self, obj_expr: str) -> str:
return f"({obj_expr}).{self.name}"
class ObjectInspector:
"""
Inspects a generic object. Uses builtins.repr() to render values and dir() to enumerate children.
"""
obj: object
def __init__(self, obj: object):
self.obj = obj
def repr(self) -> Iterable[str]:
yield repr(self.obj)
def children(self) -> Iterable[ChildObject]:
return sorted(self._attributes(), key=lambda var: var.name)
def _attributes(self) -> Iterable[ChildObject]:
# TODO: group class/instance/function/special
try:
names = dir(self.obj)
except:
names = []
for name in names:
if name.startswith("__"):
continue
try:
value = getattr(self.obj, name)
except BaseException as exc:
value = exc
try:
if hasattr(type(value), "__call__"):
continue
except:
pass
yield ChildAttribute(name, value)
def inspect(obj: object) -> ObjectInspector:
from debugpy.server.inspect import stdlib
# TODO: proper extensible registry
match obj:
case list():
return stdlib.ListInspector(obj)
case {}:
return stdlib.MappingInspector(obj)
case []:
return stdlib.SequenceInspector(obj)
case _:
return ObjectInspector(obj)

View file

@ -0,0 +1,86 @@
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See LICENSE in the project root
# for license information.
"""Object inspection for builtin Python types."""
from itertools import count
from typing import Iterable
from debugpy.common import log
from debugpy.server.inspect import ChildObject, ObjectInspector, inspect
from debugpy.server.safe_repr import SafeRepr
class ChildLen(ChildObject):
name = "len()"
def __init__(self, parent: object):
super().__init__(len(parent))
def expr(self, obj_expr: str) -> str:
return f"len({obj_expr})"
class ChildItem(ChildObject):
key: object
def __init__(self, key: object, value: object):
super().__init__(value)
self.key = key
@property
def name(self) -> str:
key_repr = "".join(inspect(self.key).repr())
return f"[{key_repr}]"
def expr(self, obj_expr: str) -> str:
return f"({obj_expr}){self.name}"
class SequenceInspector(ObjectInspector):
def children(self) -> Iterable[ChildObject]:
yield from super().children()
yield ChildLen(self.obj)
try:
it = iter(self.obj)
except:
return
for i in count():
try:
item = next(it)
except StopIteration:
break
except:
log.exception("Error retrieving next item.")
break
yield ChildItem(i, item)
class MappingInspector(ObjectInspector):
def children(self) -> Iterable["ChildObject"]:
yield from super().children()
yield ChildLen(self.obj)
try:
keys = self.obj.keys()
except:
return
it = iter(keys)
while True:
try:
key = next(it)
except StopIteration:
break
except:
break
try:
value = self.obj[key]
except BaseException as exc:
value = exc
yield ChildItem(key, value)
class ListInspector(SequenceInspector):
def repr(self) -> Iterable[str]:
# TODO: move logic from SafeRepr here
yield SafeRepr()(self.obj)