ITQC/pydase
24
0
Fork 0
mirror of https://github.com/tiqi-group/pydase.git synced 2025-06-05 21:20:40 +02:00
Code Issues Packages Projects Releases Wiki Activity

fixes PYI063 errors (ruff)

Browse Source
This commit is contained in:
Mose Müller 2025-05-20 14:08:34 +02:00
parent 98e9791d09
commit 051e616280
3 changed files with 28 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
src/pydase/client/proxy_loader.py
View File

@ -123,35 +123,35 @@ class ProxyList(list[Any]):
update_value(self._sio, self._loop, full_access_path, value)
def append(self, __object: Any) -> None:
def append(self, object_: Any, /) -> None:
full_access_path = f"{self._parent_path}.append"
trigger_method(self._sio, self._loop, full_access_path, [__object], {})
trigger_method(self._sio, self._loop, full_access_path, [object_], {})
def clear(self) -> None:
full_access_path = f"{self._parent_path}.clear"
trigger_method(self._sio, self._loop, full_access_path, [], {})
def extend(self, __iterable: Iterable[Any]) -> None:
def extend(self, iterable: Iterable[Any], /) -> None:
full_access_path = f"{self._parent_path}.extend"
trigger_method(self._sio, self._loop, full_access_path, [__iterable], {})
trigger_method(self._sio, self._loop, full_access_path, [iterable], {})
def insert(self, __index: SupportsIndex, __object: Any) -> None:
def insert(self, index: SupportsIndex, object_: Any, /) -> None:
full_access_path = f"{self._parent_path}.insert"
trigger_method(self._sio, self._loop, full_access_path, [__index, __object], {})
trigger_method(self._sio, self._loop, full_access_path, [index, object_], {})
def pop(self, __index: SupportsIndex = -1) -> Any:
def pop(self, index: SupportsIndex = -1, /) -> Any:
full_access_path = f"{self._parent_path}.pop"
return trigger_method(self._sio, self._loop, full_access_path, [__index], {})
return trigger_method(self._sio, self._loop, full_access_path, [index], {})
def remove(self, __value: Any) -> None:
def remove(self, value: Any, /) -> None:
full_access_path = f"{self._parent_path}.remove"
trigger_method(self._sio, self._loop, full_access_path, [__value], {})
trigger_method(self._sio, self._loop, full_access_path, [value], {})
class ProxyClassMixin:

16
src/pydase/data_service/data_service.py
View File

@ -27,17 +27,17 @@ class DataService(AbstractDataService):
super().__init__()
self.__check_instance_classes()
def __setattr__(self, __name: str, __value: Any) -> None:
def __setattr__(self, name: str, value: Any, /) -> None:
# Check and warn for unexpected type changes in attributes
self._warn_on_type_change(__name, __value)
self._warn_on_type_change(name, value)
# every class defined by the user should inherit from DataService if it is
# assigned to a public attribute
if not __name.startswith("_") and not inspect.isfunction(__value):
self.__warn_if_not_observable(__value)
if not name.startswith("_") and not inspect.isfunction(value):
self.__warn_if_not_observable(value)
# Set the attribute
super().__setattr__(__name, __value)
super().__setattr__(name, value)
def _warn_on_type_change(self, attr_name: str, new_value: Any) -> None:
if is_property_attribute(self, attr_name):
@ -64,8 +64,8 @@ class DataService(AbstractDataService):
)
)
def __warn_if_not_observable(self, __value: Any) -> None:
value_class = __value if inspect.isclass(__value) else __value.__class__
def __warn_if_not_observable(self, value: Any, /) -> None:
value_class = value if inspect.isclass(value) else value.__class__
if not issubclass(
value_class,
@ -81,7 +81,7 @@ class DataService(AbstractDataService):
| Observable
| Callable
),
) and not is_descriptor(__value):
) and not is_descriptor(value):
logger.warning(
"Class '%s' does not inherit from DataService. This may lead to"
" unexpected behaviour!",

20
src/pydase/observer_pattern/observable/observable_object.py
View File

@ -165,9 +165,9 @@ class _ObservableList(ObservableObject, list[Any]):
self._notify_changed(f"[{key}]", value)
def append(self, __object: Any) -> None:
def append(self, object_: Any, /) -> None:
self._notify_change_start("")
super().append(self._initialise_new_objects(f"[{len(self)}]", __object))
super().append(self._initialise_new_objects(f"[{len(self)}]", object_))
self._notify_changed("", self)
def clear(self) -> None:
@ -177,33 +177,33 @@ class _ObservableList(ObservableObject, list[Any]):
self._notify_changed("", self)
def extend(self, __iterable: Iterable[Any]) -> None:
def extend(self, iterable: Iterable[Any], /) -> None:
self._remove_self_from_observables()
try:
super().extend(__iterable)
super().extend(iterable)
finally:
for i, item in enumerate(self):
super().__setitem__(i, self._initialise_new_objects(f"[{i}]", item))
self._notify_changed("", self)
def insert(self, __index: SupportsIndex, __object: Any) -> None:
def insert(self, index: SupportsIndex, object_: Any, /) -> None:
self._remove_self_from_observables()
try:
super().insert(__index, __object)
super().insert(index, object_)
finally:
for i, item in enumerate(self):
super().__setitem__(i, self._initialise_new_objects(f"[{i}]", item))
self._notify_changed("", self)
def pop(self, __index: SupportsIndex = -1) -> Any:
def pop(self, index: SupportsIndex = -1, /) -> Any:
self._remove_self_from_observables()
try:
popped_item = super().pop(__index)
popped_item = super().pop(index)
finally:
for i, item in enumerate(self):
super().__setitem__(i, self._initialise_new_objects(f"[{i}]", item))
@ -211,11 +211,11 @@ class _ObservableList(ObservableObject, list[Any]):
self._notify_changed("", self)
return popped_item
def remove(self, __value: Any) -> None:
def remove(self, value: Any, /) -> None:
self._remove_self_from_observables()
try:
super().remove(__value)
super().remove(value)
finally:
for i, item in enumerate(self):
super().__setitem__(i, self._initialise_new_objects(f"[{i}]", item))