Skip to content

model.base: Protect NamespaceSet attributes from being replaced #619

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.

Sign up for GitHub

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

Open
zrgt wants to merge 3 commits into
base: develop
Choose a base branch
from
Open

model.base: Protect NamespaceSet attributes from being replaced #619

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5610011
model.base: Protect NamespaceSet attributes from being replaced
zrgt Aug 28, 2026
b38424d
docs: Fix unresolvable reference in NamespaceSet.assign docstring
zrgt Aug 28, 2026
a2a6045
model.base: Replace NamespaceSet guard with a descriptor
zrgt Oct 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 98 additions & 2 deletions sdk/basyx/aas/model/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -573,6 +573,75 @@ def _get_key_value_for_referable(referable: "Referable") -> str:
)


class NamespaceSetAttribute(Generic[_NSO]):
"""
Descriptor for the attributes of a :class:`~basyx.aas.model.base.Namespace` that hold a
:class:`~basyx.aas.model.base.NamespaceSet`, e.g. ``Submodel.submodel_element``.

The corresponding ``__init__`` parameters accept any ``Iterable``, so assigning an iterable to the attribute
is the obvious thing to write. A plain assignment would rebind the attribute and thereby silently shadow the
NamespaceSet: the set would stay registered in ``Namespace.namespace_element_sets``, so its objects would
remain reachable via ``get_referable()`` and friends while being invisible in the attribute, uniqueness
constraints would no longer be checked and the ``parent`` of the assigned objects would never be set.

This descriptor therefore redirects such an assignment to
:meth:`~basyx.aas.model.base.NamespaceSet.assign`, which replaces the *content* of the existing set::

submodel.submodel_element = [my_property] # replaces the content of the set
aas.extension = [] # clears the set

The first assignment, done by the ``__init__`` of the owning class, binds the NamespaceSet itself. Classes
outside of this package can use this descriptor for their own NamespaceSet attributes in the same way.
"""

_name: str
_storage_name: str

def __set_name__(self, owner: type, name: str) -> None:
self._name = name
self._storage_name = f"_{name}_namespace_set"
self.__doc__ = (
f"A :class:`~basyx.aas.model.base.NamespaceSet` holding the {name} of this element. Assigning an "
"iterable replaces its content, see "
":class:`~basyx.aas.model.base.NamespaceSetAttribute`."
)

def __get__(
self, obj: Optional["Namespace"], objtype: Optional[type] = None
) -> "NamespaceSet[_NSO]":
if obj is None:
# Class attribute access, e.g. by Sphinx autodoc
return self # type: ignore[return-value]
return getattr(obj, self._storage_name)

def __set__(self, obj: "Namespace", value: Iterable[_NSO]) -> None:
current: Optional["NamespaceSet[_NSO]"] = getattr(obj, self._storage_name, None)
if current is None:
if not isinstance(value, NamespaceSet):
raise TypeError(
f"{type(obj).__name__}.{self._name} must be initialized with a NamespaceSet, "
f"got {type(value).__name__}"
)
setattr(obj, self._storage_name, value)
elif current is not value:
current.assign(value)


class OrderedNamespaceSetAttribute(NamespaceSetAttribute[_NSO]):
"""
Descriptor for the attributes of a :class:`~basyx.aas.model.base.Namespace` that hold an
:class:`~basyx.aas.model.base.OrderedNamespaceSet`, see
:class:`~basyx.aas.model.base.NamespaceSetAttribute`. It only narrows the type of the attribute, the behavior
is identical.
"""

if TYPE_CHECKING:

def __get__(
self, obj: Optional["Namespace"], objtype: Optional[type] = None
) -> "OrderedNamespaceSet[_NSO]": ...


class Namespace(metaclass=abc.ABCMeta):
"""
Abstract baseclass for all objects which form a Namespace to hold objects and resolve them by their
Expand Down Expand Up @@ -649,11 +718,12 @@ class HasExtension(Namespace, metaclass=abc.ABCMeta):
:ivar extension: A :class:`~.NamespaceSet` of :class:`Extensions <.Extension>` of the element.
"""

extension: "NamespaceSetAttribute[Extension]" = NamespaceSetAttribute()

@abc.abstractmethod
def __init__(self) -> None:
super().__init__()
self.namespace_element_sets: List[NamespaceSet] = []
self.extension: NamespaceSet[Extension]

def get_extension_by_name(self, name: str) -> "Extension":
"""
Expand Down Expand Up @@ -1814,11 +1884,12 @@ class Qualifiable(Namespace, metaclass=abc.ABCMeta):
qualifiable element.
"""

qualifier: "NamespaceSetAttribute[Qualifier]" = NamespaceSetAttribute()

@abc.abstractmethod
def __init__(self) -> None:
super().__init__()
self.namespace_element_sets: List[NamespaceSet] = []
self.qualifier: NamespaceSet[Qualifier]

def get_qualifier_by_type(self, qualifier_type: QualifierType) -> "Qualifier":
"""
Expand Down Expand Up @@ -2339,6 +2410,31 @@ def clear(self) -> None:
for attr_name, (backend, case_sensitive) in self._backend.items():
backend.clear()

def assign(self, items: Iterable[_NSO]) -> None:
"""
Replace all objects of this set with the given ones, atomically.

This is the implementation of assignments to NamespaceSet attributes, see ``Namespace.__setattr__()``.
If adding one of the new items fails, the previous content of the set is restored.

:param items: The objects this set should contain afterwards
:raises AASConstraintViolation: If the given items violate a uniqueness constraint of this namespace
:raises ValueError: If one of the given items already belongs to another namespace
"""
# items may be a lazy iterable over this very set, e.g. filter(..., submodel.submodel_element), so it has
# to be consumed before the set is cleared
new_items: List[_NSO] = list(items)
previous: List[_NSO] = list(self)
self.clear()

@s-heppner s-heppner Oct 20, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

assign() clears the set before it reads items. If items is a lazy iterable over the same set, iterating it fails:

sm = model.Submodel("urn:sm", submodel_element=[
    model.Property("P1", model.datatypes.Int),
    model.Property("P2", model.datatypes.Int),
])
sm.submodel_element = filter(lambda x: x.id_short == "P1", sm.submodel_element)
# RuntimeError: dictionary changed size during iteration

The self.clear() leaves the set unchanged, which is good, but we'd lose functionality that currently works on develop.
One could argue that the actual set should not be accessed, but filtering a set into itself (like above) is a pretty common thing to write imo. Reading items into a list first fixes it:

Suggested change
self.clear()
new_items: List[_NSO] = list(items)
previous: List[_NSO] = list(self)
self.clear()
try:
for item in new_items:
self.add(item)

A test with a generator or filter() over the set itself would cover this case.

@zrgt zrgt Oct 22, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Fixed in a2a6045: assign() now does list(items) before self.clear().

Added test_assignment_of_lazy_iterable_over_the_same_set, which covers both your filter() example and a generator expression over the set.

Generated by Claude Code

try:
for item in new_items:
self.add(item)
except Exception:
self.clear()
for item in previous:
self.add(item)
raise

def get_object_by_attribute(
self, attribute_name: str, attribute_value: ATTRIBUTE_TYPES
) -> _NSO:
Expand Down
18 changes: 15 additions & 3 deletions sdk/basyx/aas/model/submodel.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,6 +144,10 @@ class Submodel(
:ivar embedded_data_specifications: List of Embedded data specification.
"""

submodel_element: "base.NamespaceSetAttribute[SubmodelElement]" = (
base.NamespaceSetAttribute()
)

def __init__(
self,
id_: base.Identifier,
Expand Down Expand Up @@ -730,6 +734,8 @@ class SubmodelElementCollection(SubmodelElement, base.UniqueIdShortNamespace):
:ivar embedded_data_specifications: List of Embedded data specification.
"""

value: "base.NamespaceSetAttribute[SubmodelElement]" = base.NamespaceSetAttribute()

def __init__(
self,
id_short: Optional[base.NameType],
Expand Down Expand Up @@ -757,9 +763,7 @@ def __init__(
supplemental_semantic_id,
embedded_data_specifications,
)
self.value: base.NamespaceSet[SubmodelElement] = base.NamespaceSet(
self, [("id_short", True)], value
)
self.value = base.NamespaceSet(self, [("id_short", True)], value)


_SE = TypeVar("_SE", bound=SubmodelElement)
Expand Down Expand Up @@ -1100,6 +1104,8 @@ class AnnotatedRelationshipElement(RelationshipElement, base.UniqueIdShortNamesp
:ivar embedded_data_specifications: List of Embedded data specification.
"""

annotation: "base.NamespaceSetAttribute[DataElement]" = base.NamespaceSetAttribute()

def __init__(
self,
id_short: Optional[base.NameType],
Expand Down Expand Up @@ -1177,6 +1183,10 @@ class Operation(SubmodelElement, base.UniqueIdShortNamespace):
:ivar embedded_data_specifications: List of Embedded data specification.
"""

input_variable: "base.NamespaceSetAttribute[SubmodelElement]" = base.NamespaceSetAttribute()
output_variable: "base.NamespaceSetAttribute[SubmodelElement]" = base.NamespaceSetAttribute()
in_output_variable: "base.NamespaceSetAttribute[SubmodelElement]" = base.NamespaceSetAttribute()

def __init__(
self,
id_short: Optional[base.NameType],
Expand Down Expand Up @@ -1316,6 +1326,8 @@ class Entity(SubmodelElement, base.UniqueIdShortNamespace):
:ivar embedded_data_specifications: List of Embedded data specification.
"""

statement: "base.NamespaceSetAttribute[SubmodelElement]" = base.NamespaceSetAttribute()

def __init__(
self,
id_short: Optional[base.NameType],
Expand Down
Loading
Loading