================ @@ -0,0 +1,2025 @@ +# NOTE: this module must not include `from __future__ import annotations` +# as the `annotations` import changes some of the types hints to strings. +# especially when you have a forward declared type reference. +# see https://peps.python.org/pep-0649/#motivation-for-this-pep +# https://peps.python.org/pep-0749/#rejected-alternatives +# +# This module may not depend on any other module. + +from contextlib import suppress +import copy +import dataclasses +import enum +import json +import os +import sys +import typing +from dataclasses import asdict, dataclass, field, is_dataclass +from enum import Enum +from functools import lru_cache +from typing import ( + Any, + ClassVar, + Dict, + List, + Literal, + Optional, + Protocol, + Set, + Tuple, + Type, + TypeVar, + Union, + cast, + runtime_checkable, +) + +if sys.version_info >= (3, 11): + from enum import StrEnum +else: + + class StrEnum(str, Enum): + """Backport of StrEnum for Python < 3.11.""" + + def __str__(self) -> str: + return self.value + + def __repr__(self) -> str: + return self.value + + @staticmethod + def _generate_next_value_(name: str, start, count, last_values) -> str: + return name.lower() + + +T = TypeVar("T") + + +class DAPError(AssertionError): + """The base error for all DAP related errors + + Inherits from assertion error because of unittests treats assertions outside of a test as a failure + instead of a test error. see https://docs.python.org/3/library/unittest.html#unittest.TestCase.setUp + """ + + @classmethod + def history_closed(cls, reason=None, last_event: Optional["Event"] = None): + suffix = f" (Reason: {reason})" if reason else "" + return cls( + f"EventHistory is closed{suffix}. Last recorded event: {last_event}." + ) + + +RawMessage = Dict[str, Any] +"""Representation of a json protocol message """ + + +class MessageType(StrEnum): + REQUEST = "request" + RESPONSE = "response" + EVENT = "event" + + +class EventName(StrEnum): + BREAKPOINT = "breakpoint" + CAPABILITIES = "capabilities" + CONTINUED = "continued" + EXITED = "exited" + INITIALIZED = "initialized" + INVALIDATED = "invalidated" + MEMORY = "memory" + MODULE = "module" + OUTPUT = "output" + PROCESS = "process" + PROGRESS_END = "progressEnd" + PROGRESS_START = "progressStart" + PROGRESS_UPDATE = "progressUpdate" + STOPPED = "stopped" + TERMINATED = "terminated" + THREAD = "thread" + + +@dataclass(frozen=True) +class ProtocolMessage: + type: MessageType + seq: int + + def to_dict(self): + return message_to_dict(self) + + @classmethod + def from_json(cls: Type[T], json: RawMessage) -> T: + if not dataclasses.is_dataclass(cls): + raise ValueError(f"{cls.__name__} must be a dataclass") + + return dict_to_message(cls, json) + + +@dataclass(frozen=True) +class Request(ProtocolMessage): + command: str + arguments: Any + + def __post_init__(self): + assert ( + self.type == MessageType.REQUEST + ), f"expected request type to be 'request' got '{self.type}' in : {self}" + + +@dataclass(frozen=True) +class Response(ProtocolMessage): + command: str + request_seq: int + success: bool + + def __post_init__(self): + assert ( + self.type == MessageType.RESPONSE + ), f"expected '{type(self).__name__}' to be of type response: {self}" + + def __init_subclass__(cls, **kwargs): + super().__init_subclass__(**kwargs) + if not dataclasses.is_dataclass(cls): + raise TypeError(f"{cls.__name__} must be a dataclass") + + +AnyResponse = TypeVar("AnyResponse", bound=Response) + + +class RequestError(DAPError): + """Raised if a DAP request fails.""" + + def __init__( + self, + request: Union[Request, dict], + response: Optional[Union[Response, dict]] = None, + ): + super().__init__() + self.request = request + self.response = response + + def __str__(self) -> str: + desc = f"request failed request={self.request!r}" + if self.response: + desc += f" response={self.response!r}" + return desc + + +@dataclass(frozen=True) +class Event(ProtocolMessage): + event: Union[EventName, str] + type: MessageType + __registry: ClassVar[Dict[str, Type]] = {} + + def __init_subclass__(cls, *, event: str, **kwargs): + super().__init_subclass__(**kwargs) + + assert event is not None + # Attach metadata (not a field of an event). + cls.__message_type__ = event + + # Prevent duplicate types. + existing_event_class = Event.__registry.get(event) + if existing_event_class is not None: + raise Exception( + f"cannot register '{event}' event to class '{cls}' because it is already registered to '{existing_event_class}'" + ) + + Event.__registry[event] = cls + + def __post_init__(self): + if self.type != MessageType.EVENT: + raise ValueError(f"event must have type of 'event': {self}.") + + @classmethod + def from_json(cls: Type[T], json: dict) -> T: + event_type = json["event"] + + if event_type not in Event.__registry: + raise ValueError(f"event type '{event_type}' is not registered.") + + event_class = Event.__registry[event_type] + + if cls not in (Event, event_class): + raise ValueError( + f"class: {cls.__name__!r} is not of the expected type {event_class.__name__!r}." + ) + + if not dataclasses.is_dataclass(event_class): + raise ValueError(f"{cls.__name__!r} must be a dataclass.") + + return cast(T, dict_to_message(event_class, json)) + + +AnyEvent = TypeVar("AnyEvent", bound=Event) + + +@dataclass(frozen=True) +class EmptyBodyResponse(Response): + body: None = field(init=False, default=None) + + +def args_protocol(cls): + """Decorator to check a class conforms to the ArgsProtocol""" + + required_fields = filter(lambda x: not x.startswith("_"), dir(ArgsProtocol)) + for r_field in required_fields: + if not hasattr(cls, r_field): + raise AttributeError( + f"{cls.__name__} must define '{r_field}' to implement ArgProtocol" + ) + + command_name = getattr(cls, "command_") + if not issubclass(type(command_name), str): + raise TypeError( + f"the command_ type '{type(command_name)}' for class '{cls.__name__}' must be string like" + ) + + return cls + + +def _message_to_dict_impl(obj: typing.Any, skip_none: bool = True) -> typing.Any: + if dataclasses.is_dataclass(obj): + fields = _get_dataclass_fields(type(obj)) + visited: Dict[str, Any] = {} + for f in fields: + dict_name = f.metadata.get("alias", f.name) + if dict_name in visited or not hasattr(obj, f.name): + continue + name_attr = getattr(obj, f.name) + if skip_none and name_attr is None: + continue + visited[dict_name] = _message_to_dict_impl(name_attr, skip_none) + return visited + + if isinstance(obj, (list, tuple)): + return type(obj)(_message_to_dict_impl(item, skip_none) for item in obj) + + if isinstance(obj, dict): + result: Dict[str, Any] = {} + for key, value in obj.items(): + result[str(key)] = _message_to_dict_impl(value, skip_none) + return result + + # Test enum first as it can also be a subclass of other primitives. + if isinstance(obj, Enum): + return obj.value + + if isinstance(obj, (bool, int, float, str, bytes, type, type(None))): + return obj + + return copy.deepcopy(obj) + + +def message_to_dict(args: Any) -> RawMessage: + """ + Converts DAP types to dictionaries. + We always skip optional types in dataclasses during conversion. + """ + if not is_dataclass(args): + raise TypeError( + f"expected a dataclass instance, got {type(args).__name__}: {args!r}" + ) + result = _message_to_dict_impl(args) + assert isinstance(result, dict) + return result + + +@lru_cache +def _get_dataclass_fields(cls: Type) -> Tuple[dataclasses.Field, ...]: + data_class_hints = typing.get_type_hints(cls) + result: List[dataclasses.Field] = [] + for f in dataclasses.fields(cls): # noqa + # Ignores 'command_' and 'response_class_'. + if f.name.endswith("_"): + continue + f_copy = copy.copy(f) + f_copy.type = data_class_hints[f_copy.name] + result.append(f_copy) + return tuple(result) + + +@lru_cache +def _prepare_dataclass_fields(cls: Type) -> Dict[str, dataclasses.Field]: + # Excludes init=False fields since they cannot be passed to __init__. + return {f.name: f for f in _get_dataclass_fields(cls) if f.init} + + +def _get_compatible_union_types(data: Any, possible_types: List[Type]) -> List[Type]: + """Filters a list of candidate types to find those compatible with the given raw data. + + This helps fail early on type mismatches during parsing. For example, it + ensures that a raw payload like `[10, 20]` is correctly matched to `List[int]` + rather than a primitive `int`. + """ + + def _is_type_compatible(candidate_type: Type, data: Any) -> bool: + """Helper function to check if a single type is compatible with the data.""" + origin = typing.get_origin(candidate_type) + + if dataclasses.is_dataclass(candidate_type) or origin is dict: + return isinstance(data, dict) + + if origin in (list, tuple): + return isinstance(data, (list, tuple)) + + if candidate_type is bool: + return isinstance(data, bool) + + if candidate_type in (int, float): + # Prevent implicit bool conversion as bool is a subclass of int. + return isinstance(data, (int, float)) and not isinstance(data, bool) + + if candidate_type is str: + return isinstance(data, str) + + if isinstance(candidate_type, type) and issubclass(candidate_type, enum.Enum): + if issubclass(candidate_type, enum.IntEnum): + return isinstance(data, int) and not isinstance(data, bool) + return isinstance(data, str) + + # Fallback for unknown or other generic types + return True + + result = [a_type for a_type in possible_types if _is_type_compatible(a_type, data)] + return result + + +def _generic_to_message(cls: Optional[Type], data: Any, scope: List[str]) -> Any: + origin = typing.get_origin(cls) + args = typing.get_args(cls) + full_path = ".".join(scope) + + if origin is Literal: + unique_literals: Set = set() + + def flatten_literal(args: Any): + """'Literal' types can be nested e.g. + Literal[Literal["book"], Literal["pen"]] is the same as Literal["book", "pen"] + """ + for val in args: + if typing.get_origin(val) is None: + unique_literals.add(val) + else: + flatten_literal(typing.get_args(val)) + + flatten_literal(args) + + if data not in unique_literals: + raise ValueError( + f"expected one of {unique_literals!r} at {full_path}, got {data!r}" + ) + return data + + elif origin is Union: + none_type = type(None) + is_optional = none_type in args + candidate_args = [a for a in args if a is not none_type] + + # Handle Optional. + # Optional is represented as Union[T, None]. + if is_optional and len(candidate_args) == 1: + return _dict_to_message_impl(candidate_args[0], data, scope) + + matches = _get_compatible_union_types(data, candidate_args) + if not matches: + raise TypeError( + f"no variant of {cls} is compatible with " + f"{type(data).__name__} at {full_path}: {data!r}" + ) + + # Only one match, try conversion. + if len(matches) == 1: + return _dict_to_message_impl(matches[0], data, scope) + + for arg in matches: + with suppress(TypeError, ValueError, AttributeError, AssertionError): + return _dict_to_message_impl(arg, data, scope) + + raise TypeError( + f"no variant of {cls} matched {type(data).__name__} at {full_path}: {data!r}" + ) + + elif origin is list: + if not isinstance(data, list): + raise TypeError( + f"expected list at {full_path}, got {type(data).__name__}: {data!r}" + ) + list_type = args[0] + return [ + _dict_to_message_impl(list_type, val, scope + [str(idx)]) + for idx, val in enumerate(data) + ] + + elif origin is tuple: + if not isinstance(data, list): + raise TypeError( + f"expected list for tuple at {full_path}, got {type(data).__name__}: {data!r}" + ) + item_type = args[0] + return tuple( + _dict_to_message_impl(item_type, val, scope + [str(idx)]) + for idx, val in enumerate(data) + ) + + elif origin is dict: + if not isinstance(data, dict): + raise TypeError( + f"expected dict at {full_path}, got {type(data).__name__}: {data!r}" + ) + if len(args) == 0: + args = (str, Any) + key_type = args[0] + value_type = args[1] + return { + _dict_to_message_impl(key_type, key, scope): _dict_to_message_impl( + value_type, value, scope + [str(key)] + ) + for key, value in data.items() + } + + raise TypeError(f"unhandled generic type {cls} at {full_path}: {data!r}") + + +def _dict_to_message_impl( + cls: Optional[Type], data: typing.Any, scope: List[str] +) -> typing.Any: + """ + Recursively deserializes a dictionary into a specified Python type or dataclass. + + Args: + cls: The target Python type or dataclass to deserialize the data into. + data: The raw data (usually from a dictionary or JSON payload) to be converted. + scope: A list of keys representing the current depth in the nested data structure. + Used to provide error messages when validation fails. + + Returns: + The deserialized data cast to the requested type or instantiated dataclass. + + Raises: + TypeError: If the data type does not match the expected type and cannot be coerced. + ValueError: If a dataclass field has a required value that does not match the data. + """ + if not cls: + return data + + full_path = ".".join(scope) + if data is None: + # Only pass None through if the declared type permits it. + if cls is type(None): + return None + is_union = typing.get_origin(cls) is Union + if is_union and type(None) in typing.get_args(cls): # Is Optional + return None + + raise TypeError(f"got None for non-optional type '{cls}' at {full_path}") + + if cls in (bytes, bytearray, Any): + return data + + if dataclasses.is_dataclass(cls): + if not isinstance(data, dict): + raise TypeError( + f"expected dict for '{cls.__name__}' at {full_path}, " + f"got {type(data).__name__}: {data!r}" + ) + fields = _prepare_dataclass_fields(cls) + deserialized = {} + for key, f in fields.items(): ---------------- da-viper wrote:
Updated. https://github.com/llvm/llvm-project/pull/203978 _______________________________________________ lldb-commits mailing list [email protected] https://lists.llvm.org/cgi-bin/mailman/listinfo/lldb-commits
