Fix EdgePropertyList construction via _from_object_info (#562)

Fixes #561

The `EdgePropertyList` retains the _same_ list instance during its lifetime 
for containing the edge property wrappers. This is done so that references
user code might hold to its items to correctly update its contents.

In contrast to other types, it means that the state is not re-fetched from
the `_pb_object` if it were to change outside of the `EdgePropertyList`'s
control.
This led to a bug when constructing objects via `_from_object_info`, which
first default-constructs the object (in the process creating the `EdgePropertyList`)
and then modifies the `_pb_object`.

To fix this, the `EdgePropertyList` now stores whether its parent object was
stored when this list was last accessed. If it changes from unstored to stored,
the following logic is applied:
- If the current list is already populated (the "regular" case), only a sanity check
  for matching length is performed
- If the current list is empty, it is re-fetched from the parent object. This is the
  case which occurs during construction with `_from_object_info`

Since the edge property list can be in an inconsistent state (empty when it shouldn't 
be) while the parent is unstored, we disallow accessing it in this state. It is still
allowed however to fully replace the contents.

This PR also fixes a bug in `tree_object_from_resource_path`, where the `channel`
argument was still used instead of `server_wrapper`.

src/ansys/acp/core/_tree_objects/_grpc_helpers/edge_property_list.py

@@ -85,6 +85,16 @@ class EdgePropertyList(ObjectCacheMixin, MutableSequence[ValueT]):
     For instance, element sets of an oriented element set.
     """
 
+    __slots__ = (
+        "_object_list_store",
+        "_parent_object",
+        "_parent_was_stored",
+        "_object_type",
+        "_attribute_name",
+        "_name",
+        "_object_constructor",
+    )
+
     @classmethod
     @constructor_with_cache(
         # NOTE greschd Feb'23:
@@ -137,49 +147,72 @@ def __init__(
         _attribute_name: str,
         _from_pb_constructor: Callable[[CreatableTreeObject, Message, Callable[[], None]], ValueT],
     ) -> None:
-        getter = grpc_data_getter(_attribute_name, from_protobuf=list)
-        setter = grpc_data_setter(_attribute_name, to_protobuf=lambda x: x)
-
         self._parent_object = _parent_object
+        self._parent_was_stored = self._parent_object._is_stored
         self._object_type = _object_type
+        self._attribute_name = _attribute_name
         self._name = _attribute_name.split(".")[-1]
 
         self._object_constructor: Callable[[Message], ValueT] = (
             lambda pb_object: _from_pb_constructor(
                 self._parent_object, pb_object, self._apply_changes
             )
         )
-
-        # get initial object list
-        def get_object_list_from_parent_object() -> list[ValueT]:
-            obj_list = []
-            for item in getter(_parent_object):
-                obj_list.append(self._object_constructor(item))
-            return obj_list
-
-        self._object_list = get_object_list_from_parent_object()
-
-        def set_object_list(items: list[ValueT]) -> None:
-            """Set the object list on the parent AND updates the internal object list."""
-            pb_obj_list = []
-
-            for item in items:
-                if not isinstance(item, _object_type):
-                    raise TypeError(
-                        f"Expected items of type {_object_type}, got type {type(item)} instead. "
-                        f"Item: {item}."
-                    )
-                if not item._check():
-                    raise RuntimeError("Cannot initialize incomplete object.")
-                pb_obj_list.append(item._to_pb_object())
-                # update callback in case item was copied from another tree object
-                # or if it is a new object
-                item._set_callback_apply_changes(self._apply_changes)
-            setter(_parent_object, pb_obj_list)
-            # keep object list in sync with the backend. This is needed for the in-place editing
-            self._object_list = items
-
-        self._set_object_list = set_object_list
+        if self._parent_object._is_stored:
+            self._object_list_store = self._get_object_list_from_parent()
+        else:
+            # Cannot instantiate the objects if the server is not available
+            self._object_list_store = []
+
+    @property
+    def _object_list(self) -> list[ValueT]:
+        if self._parent_object._is_stored and not self._parent_was_stored:
+            # There are two scenarios when the parent object becomes
+            # stored:
+            # - The _object_list already contained some values. In this case, we
+            #   simply keep it, and make a (inexaustive) check that the size
+            #   matches.
+            # - The parent object was default-constructed and then its _pb_object
+            #   was then replaced (e.g. by a call to _from_object_info). In this
+            #   case, the empty '_object_list' is no longer reflecting the actual
+            #   state, and needs to be replaced.
+            #   In general, we don't replace the _object_list to ensure any references
+            #   to its elements are correctly updated (and retain the ability to
+            #   update the object itself), but here it's not a concern since this
+            #   only happens within the constructor.
+            if self._object_list_store:
+                assert len(self._object_list_store) == len(self._get_object_list_from_parent())
+            else:
+                self._object_list_store = self._get_object_list_from_parent()
+            self._parent_was_stored = True
+        return self._object_list_store
+
+    def _get_object_list_from_parent(self) -> list[ValueT]:
+        obj_list = []
+        for item in grpc_data_getter(self._attribute_name, from_protobuf=list)(self._parent_object):
+            obj_list.append(self._object_constructor(item))
+        return obj_list
+
+    def _set_object_list(self, items: list[ValueT]) -> None:
+        """Set the object list on the parent AND updates the internal object list."""
+        pb_obj_list = []
+        for item in items:
+            if not isinstance(item, self._object_type):
+                raise TypeError(
+                    f"Expected items of type {self._object_type}, got type {type(item)} instead. "
+                    f"Item: {item}."
+                )
+            if not item._check():
+                raise RuntimeError("Cannot initialize incomplete object.")
+            pb_obj_list.append(item._to_pb_object())
+            # update callback in case item was copied from another tree object
+            # or if it is a new object
+            item._set_callback_apply_changes(self._apply_changes)
+        grpc_data_setter(self._attribute_name, to_protobuf=lambda x: x)(
+            self._parent_object, pb_obj_list
+        )
+        # keep object list in sync with the backend. This is needed for the in-place editing
+        self._object_list_store = items
 
     def __len__(self) -> int:
         return len(self._object_list)
@@ -364,7 +397,13 @@ def define_edge_property_list(
 ) -> Any:
     """Define a list of linked tree objects with link properties."""
 
-    def getter(self: CreatableTreeObject) -> EdgePropertyList[GenericEdgePropertyType]:
+    def getter(
+        self: CreatableTreeObject, *, check_stored: bool = True
+    ) -> EdgePropertyList[GenericEdgePropertyType]:
+        if check_stored and not self._is_stored:
+            raise RuntimeError(
+                f"Cannot access property {attribute_name.split('.')[-1]} on an object that is not stored."
+            )
         return EdgePropertyList._initialize_with_cache(
             parent_object=self,
             object_type=value_type,
@@ -373,7 +412,8 @@ def getter(self: CreatableTreeObject) -> EdgePropertyList[GenericEdgePropertyTyp
         )
 
     def setter(self: CreatableTreeObject, value: list[GenericEdgePropertyType]) -> None:
-        getter(self)[:] = value
+        # allow wholesale replacement on unstored objects
+        getter(self, check_stored=False)[:] = value
 
     return _wrap_doc(_exposed_grpc_property(getter).setter(setter), doc=doc)
 

src/ansys/acp/core/_tree_objects/_grpc_helpers/polymorphic_from_pb.py

@@ -22,26 +22,31 @@
 
 from __future__ import annotations
 
+import typing
 from typing import Protocol
 
-import grpc
 from typing_extensions import Self
 
 from ansys.api.acp.v0.base_pb2 import ResourcePath
 
+if typing.TYPE_CHECKING:  # pragma: no cover
+    from ..base import ServerWrapper
+
 __all__ = ["CreatableFromResourcePath", "tree_object_from_resource_path"]
 
 
 class CreatableFromResourcePath(Protocol):
     """Interface for objects that can be created from a resource path."""
 
     @classmethod
-    def _from_resource_path(cls, resource_path: ResourcePath, channel: grpc.Channel) -> Self: ...
+    def _from_resource_path(
+        cls, resource_path: ResourcePath, server_wrapper: ServerWrapper
+    ) -> Self: ...
 
 
 def tree_object_from_resource_path(
     resource_path: ResourcePath,
-    channel: grpc.Channel,
+    server_wrapper: ServerWrapper,
     allowed_types: tuple[type[CreatableFromResourcePath], ...] | None = None,
 ) -> CreatableFromResourcePath | None:
     """Instantiate a tree object from its resource path.
@@ -50,8 +55,8 @@ def tree_object_from_resource_path(
     ----------
     resource_path :
         Resource path of the object.
-    channel :
-        gRPC channel to the server.
+    server_wrapper :
+        Representation of the ACP server.
     allowed_types :
         Allowed types of the object. If None, all registered types are allowed.
     """
@@ -72,4 +77,4 @@ def tree_object_from_resource_path(
                 f"Resource path {resource_path.value} does not point to a valid "
                 f"object type. Allowed types: {allowed_types}"
             )
-    return resource_class._from_resource_path(resource_path, channel)
+    return resource_class._from_resource_path(resource_path, server_wrapper=server_wrapper)

src/ansys/acp/core/_tree_objects/_grpc_helpers/property_helper.py

@@ -28,7 +28,6 @@
 from __future__ import annotations
 
 from functools import reduce
-import typing
 from typing import Any, Callable, TypeVar
 
 from google.protobuf.message import Message
@@ -97,7 +96,9 @@ def inner(self: Readable) -> CreatableFromResourcePath | None:
         self._get()
         object_resource_path = _get_data_attribute(self._pb_object, name)
 
-        return tree_object_from_resource_path(object_resource_path, self._channel)
+        return tree_object_from_resource_path(
+            object_resource_path, server_wrapper=self._server_wrapper
+        )
 
     return inner
 
@@ -128,6 +129,20 @@ def inner(self: Readable) -> Any:
     return inner
 
 
+def grpc_linked_object_setter(
+    name: str, to_protobuf: _TO_PROTOBUF_T[Readable | None]
+) -> Callable[[Editable, Readable | None], None]:
+    """Create a setter method which updates the linked object via the gRPC Put endpoint."""
+    func = grpc_data_setter(name, to_protobuf)
+
+    def inner(self: Editable, value: Readable | None) -> None:
+        if value is not None and not value._is_stored:
+            raise Exception("Cannot link to an unstored object.")
+        func(self, value)
+
+    return inner
+
+
 def grpc_data_setter(
     name: str, to_protobuf: _TO_PROTOBUF_T[_SET_T]
 ) -> Callable[[Editable, _SET_T], None]:
@@ -284,10 +299,8 @@ def grpc_link_property(
         Types which are allowed to be set on the property. An
         error will be raised if an object of a different type is set.
     """
-    if typing.TYPE_CHECKING:
-        from ..base import TreeObjectBase
 
-    def to_protobuf(obj: TreeObjectBase | None) -> ResourcePath:
+    def to_protobuf(obj: Readable | None) -> ResourcePath:
         if obj is None:
             return ResourcePath(value="")
         if not isinstance(obj, allowed_types):
@@ -301,7 +314,7 @@ def to_protobuf(obj: TreeObjectBase | None) -> ResourcePath:
     return _wrap_doc(
         _exposed_grpc_property(grpc_linked_object_getter(name)).setter(
             # Resource path represents an object that is not set as an empty string
-            grpc_data_setter(name=name, to_protobuf=to_protobuf)
+            grpc_linked_object_setter(name=name, to_protobuf=to_protobuf)
         ),
         doc=doc,
     )

src/ansys/acp/core/_tree_objects/_grpc_helpers/protocols.py

@@ -24,6 +24,7 @@
 
 from collections.abc import Iterable
 import textwrap
+import typing
 from typing import Any, Protocol
 
 from google.protobuf.message import Message
@@ -36,8 +37,12 @@
     Empty,
     GetRequest,
     ListRequest,
+    ResourcePath,
 )
 
+if typing.TYPE_CHECKING:  # pragma: no cover
+    from ..base import ServerWrapper
+
 
 class CreateRequest(Protocol):
     """Interface definition for CreateRequest messages.
@@ -176,7 +181,10 @@ def _get_if_stored(self) -> None: ...
     def _is_stored(self) -> bool: ...
 
     @property
-    def _channel(self) -> grpc.Channel: ...
+    def _server_wrapper(self) -> ServerWrapper: ...
+
+    @property
+    def _resource_path(self) -> ResourcePath: ...
 
     _pb_object: Any
 

src/ansys/acp/core/_tree_objects/_mesh_data.py

@@ -44,7 +44,7 @@
     nodal_data_type_to_pb,
 )
 
-if typing.TYPE_CHECKING:
+if typing.TYPE_CHECKING:  # pragma: no cover
     from .model import MeshData  # avoid circular import
 
 __all__ = [

src/ansys/acp/core/_tree_objects/base.py

@@ -43,6 +43,7 @@
 from .._utils.resource_paths import to_parts
 from ._grpc_helpers.exceptions import wrap_grpc_errors
 from ._grpc_helpers.linked_object_helpers import linked_path_fields, unlink_objects
+from ._grpc_helpers.polymorphic_from_pb import CreatableFromResourcePath
 from ._grpc_helpers.property_helper import (
     _get_data_attribute,
     grpc_data_property,
@@ -61,7 +62,7 @@
 )
 from ._object_cache import ObjectCacheMixin, constructor_with_cache
 
-if typing.TYPE_CHECKING:
+if typing.TYPE_CHECKING:  # pragma: no cover
     from .._server import ACP
 
 
@@ -476,8 +477,12 @@ def inner(self: T, /, *args: P.args, **kwargs: P.kwargs) -> R:
     return decorator
 
 
-if typing.TYPE_CHECKING:
+if typing.TYPE_CHECKING:  # pragma: no cover
     # Ensure that the ReadOnlyTreeObject satisfies the Gettable interface
     _x: Readable = typing.cast(ReadOnlyTreeObject, None)
-    # Ensure that the TreeObject satisfies the Editable interface
+    # Ensure that the TreeObject satisfies the Editable and Readable interfaces
     _y: Editable = typing.cast(TreeObject, None)
+    _z: Readable = typing.cast(TreeObject, None)
+
+    # Ensure the TreeObjectBase satisfies the CreatableFromResourcePath interface
+    _a: CreatableFromResourcePath = typing.cast(TreeObjectBase, None)

src/ansys/acp/core/_tree_objects/linked_selection_rule.py

@@ -45,7 +45,7 @@
 from .tube_selection_rule import TubeSelectionRule
 from .variable_offset_selection_rule import VariableOffsetSelectionRule
 
-if typing.TYPE_CHECKING:
+if typing.TYPE_CHECKING:  # pragma: no cover
     # Since the 'LinkedSelectionRule' class is used by the boolean selection rule,
     # this would cause a circular import at run-time.
     from .boolean_selection_rule import BooleanSelectionRule
@@ -212,7 +212,7 @@ def _from_pb_object(
         allowed_types = tuple(allowed_types_list)
 
         selection_rule = tree_object_from_resource_path(
-            resource_path=message.resource_path, channel=parent_object._channel
+            resource_path=message.resource_path, server_wrapper=parent_object._server_wrapper
         )
         if not isinstance(selection_rule, allowed_types):
             raise TypeError(

src/ansys/acp/core/_tree_objects/oriented_selection_set.py

@@ -79,7 +79,7 @@
     "OrientedSelectionSetNodalData",
 ]
 
-if typing.TYPE_CHECKING:
+if typing.TYPE_CHECKING:  # pragma: no cover
     # Since the 'LinkedSelectionRule' class is used by the boolean selection rule,
     # this would cause a circular import at run-time.
     from .. import BooleanSelectionRule, GeometricalSelectionRule

src/ansys/acp/core/_tree_objects/sublaminate.py

@@ -107,7 +107,7 @@ def _from_pb_object(
         apply_changes: Callable[[], None],
     ) -> Lamina:
         material = tree_object_from_resource_path(
-            resource_path=message.material, channel=parent_object._channel
+            resource_path=message.material, server_wrapper=parent_object._server_wrapper
         )
 
         if not isinstance(material, get_args(_LINKABLE_MATERIAL_TYPES)):

src/ansys/acp/core/_tree_objects/virtual_geometry.py

@@ -39,7 +39,7 @@
 from .enums import status_type_from_pb, virtual_geometry_dimension_from_pb
 from .object_registry import register
 
-if typing.TYPE_CHECKING:
+if typing.TYPE_CHECKING:  # pragma: no cover
     from .cad_component import CADComponent
 
 

src/ansys/acp/core/_workflow.py

@@ -33,7 +33,7 @@
 from ._typing_helper import PATH
 
 # Avoid dependencies on pydpf-composites and dpf-core if it is not used
-if typing.TYPE_CHECKING:
+if typing.TYPE_CHECKING:  # pragma: no cover
     from ansys.dpf.composites.data_sources import ContinuousFiberCompositesFiles
     from ansys.dpf.core import UnitSystem