class PropertiesRegistry(DescriptorRegistry):
"""
A `DescriptorRegistry` for properties of a `Thing` class or `Thing` instance.
[UML Diagram](https://docs.hololinked.dev/UML/PDF/DescriptorRegistry.pdf)
"""
def __init__(self, owner_cls: ThingMeta, owner_class_members: dict, owner_inst=None):
super().__init__(owner_cls, owner_inst)
if self.owner_inst is None and owner_class_members is not None:
# instantiated by class
self.event_resolver = ParamEventResolver(owner_cls=owner_cls)
self.event_dispatcher = ParamEventDispatcher(owner_cls, self.event_resolver)
self.event_resolver.create_unresolved_watcher_info(owner_class_members)
else:
# instantiated by instance
self._instance_params = {}
self.event_resolver = self.owner_cls.properties.event_resolver
self.event_dispatcher = ParamEventDispatcher(owner_inst, self.event_resolver)
self.event_dispatcher.prepare_instance_dependencies()
@property
def descriptor_object(self) -> type[Parameter]:
return Parameter
@property
def descriptors(self) -> typing.Dict[str, Parameter]:
if self.owner_inst is None:
return super().get_descriptors()
return dict(super().get_descriptors(), **self._instance_params)
values = property(DescriptorRegistry.get_values,
doc=DescriptorRegistry.get_values.__doc__) # type: typing.Dict[str, Parameter | Property | typing.Any]
def __getitem__(self, key: str) -> Property | Parameter:
return self.descriptors[key]
def __contains__(self, value: str | Property | Parameter) -> bool:
return value in self.descriptors.values() or value in self.descriptors
@property
def defaults(self) -> typing.Dict[str, typing.Any]:
"""default values of all properties as a dictionary with property names as keys"""
defaults = {}
for key, val in self.descriptors.items():
defaults[key] = val.default
return defaults
@property
def remote_objects(self) -> typing.Dict[str, Property]:
"""
dictionary of properties that are remotely accessible (`remote=True`),
which is also a default setting for all properties
"""
try:
return getattr(self, f'_{self._qualified_prefix}_{self.__class__.__name__.lower()}_remote')
except AttributeError:
props = self.descriptors
remote_props = {}
for name, desc in props.items():
if not isinstance(desc, Property):
continue
if desc.is_remote:
remote_props[name] = desc
setattr(
self,
f'_{self._qualified_prefix}_{self.__class__.__name__.lower()}_remote',
remote_props
)
return remote_props
@property
def db_objects(self) -> typing.Dict[str, Property]:
"""
dictionary of properties that are stored or loaded from the database
(`db_init`, `db_persist` or `db_commit` set to True)
"""
try:
return getattr(self, f'_{self._qualified_prefix}_{self.__class__.__name__.lower()}_db')
except AttributeError:
propdict = self.descriptors
db_props = {}
for name, desc in propdict.items():
if not isinstance(desc, Property):
continue
if desc.db_init or desc.db_persist or desc.db_commit:
db_props[name] = desc
setattr(
self,
f'_{self._qualified_prefix}_{self.__class__.__name__.lower()}_db',
db_props
)
return db_props
@property
def db_init_objects(self) -> typing.Dict[str, Property]:
"""dictionary of properties that are initialized from the database (`db_init` or `db_persist` set to True)"""
try:
return getattr(self, f'_{self._qualified_prefix}_{self.__class__.__name__.lower()}_db_init')
except AttributeError:
propdict = self.db_objects
db_init_props = {}
for name, desc in propdict.items():
if desc.db_init or desc.db_persist:
db_init_props[name] = desc
setattr(
self,
f'_{self._qualified_prefix}_{self.__class__.__name__.lower()}_db_init',
db_init_props
)
return db_init_props
@property
def db_commit_objects(self) -> typing.Dict[str, Property]:
"""dictionary of properties that are committed to the database (`db_commit` or `db_persist` set to True)"""
try:
return getattr(self, f'_{self._qualified_prefix}_{self.__class__.__name__.lower()}_db_commit')
except AttributeError:
propdict = self.db_objects
db_commit_props = {}
for name, desc in propdict.items():
if desc.db_commit or desc.db_persist:
db_commit_props[name] = desc
setattr(
self,
f'_{self._qualified_prefix}_{self.__class__.__name__.lower()}_db_commit',
db_commit_props
)
return db_commit_props
@property
def db_persisting_objects(self) -> typing.Dict[str, Property]:
"""dictionary of properties that are persisted through the database (`db_persist` set to True)"""
try:
return getattr(self, f'_{self._qualified_prefix}_{self.__class__.__name__.lower()}_db_persisting')
except AttributeError:
propdict = self.db_objects
db_persisting_props = {}
for name, desc in propdict.items():
if desc.db_persist:
db_persisting_props[name] = desc
setattr(
self,
f'_{self._qualified_prefix}_{self.__class__.__name__.lower()}_db_persisting',
db_persisting_props
)
return db_persisting_props
def get(self, **kwargs: typing.Dict[str, typing.Any]) -> typing.Dict[str, typing.Any]:
"""
read properties from the object, implements WoT operations `readAllProperties` and `readMultipleProperties`
Parameters
----------
**kwargs: typing.Dict[str, typing.Any]
- names: `List[str]`
list of property names to be fetched
- name: `str`
name of the property to be fetched, along with a 'rename' for the property in the response.
For example { 'foo_prop' : 'fooProp' } will return the property 'foo_prop' as 'fooProp' in the response.
Returns
-------
typing.Dict[str, typing.Any]
dictionary of property names and their values
Raises
------
TypeError
if property name is not a string or requested new name is not a string
AttributeError
if property does not exist or is not remote accessible
"""
data = {}
if len(kwargs) == 0:
# read all properties
for name, prop in self.remote_objects.items():
if self.owner_inst is None and not prop.class_member:
continue
data[name] = prop.__get__(self.owner_inst, self.owner_cls)
return data
elif 'names' in kwargs:
names = kwargs.get('names')
if not isinstance(names, (list, tuple, str)):
raise TypeError("Specify properties to be fetched as a list, tuple or comma separated names. " +
f"Given type {type(names)}")
if isinstance(names, str):
names = names.split(',')
kwargs = {name: name for name in names}
for requested_prop, rename in kwargs.items():
if not isinstance(requested_prop, str):
raise TypeError(f"property name must be a string. Given type {type(requested_prop)}")
if not isinstance(rename, str):
raise TypeError(f"requested new name must be a string. Given type {type(rename)}")
if requested_prop not in self.descriptors:
raise AttributeError(f"property {requested_prop} does not exist")
if requested_prop not in self.remote_objects:
raise AttributeError(f"property {requested_prop} is not remote accessible")
prop = self.descriptors[requested_prop]
if self.owner_inst is None and not prop.class_member:
continue
data[rename] = prop.__get__(self.owner_inst, self.owner_cls)
return data
def set(self, **values : typing.Dict[str, typing.Any]) -> None:
"""
set properties whose name is specified by keys of a dictionary; implements WoT operations `writeMultipleProperties`
or `writeAllProperties`.
Parameters
----------
values: typing.Dict[str, typing.Any]
dictionary of property names and its new values
Raises
------
AttributeError
if property does not exist or is not remote accessible
"""
errors = ''
for name, value in values.items():
try:
if name not in self.descriptors:
raise AttributeError(f"property {name} does not exist")
if name not in self.remote_objects:
raise AttributeError(f"property {name} is not remote accessible")
prop = self.descriptors[name]
if self.owner_inst is None and not prop.class_member:
raise AttributeError(f"property {name} is not a class member and cannot be set at class level")
setattr(self.owner, name, value)
except Exception as ex:
errors += f'{name}: {str(ex)}\n'
if errors:
ex = RuntimeError("Some properties could not be set due to errors. " +
"Check exception notes or server logs for more information.")
ex.__notes__ = errors
raise ex from None
def add(self, name: str, config: JSON) -> None:
"""
add a property to the object
Parameters
----------
name: str
name of the property
config: JSON
configuration of the property, i.e. keyword arguments to the `__init__` method of the property class
"""
prop = self.get_type_from_name(**config)
setattr(self.owner_cls, name, prop)
prop.__set_name__(self.owner_cls, name)
if prop.deepcopy_default:
self._deep_copy_param_descriptor(prop)
self._deep_copy_param_default(prop)
self.clear()
def clear(self):
super().clear()
self._instance_params = {}
for attr in ['_db', '_db_init', '_db_persisting', '_remote']:
try:
delattr(self, f'_{self._qualified_prefix}_{self.__class__.__name__.lower()}{attr}')
except AttributeError:
pass
@supports_only_instance_access("database operations are only supported at instance level")
def get_from_DB(self) -> typing.Dict[str, typing.Any]:
"""
get all properties (i.e. their values) currently stored in the database
Returns
-------
Dict[str, typing.Any]
dictionary of property names and their values
"""
if not hasattr(self.owner_inst, 'db_engine'):
raise AttributeError("database engine not set, this object is not connected to a database")
props = self.owner_inst.db_engine.get_all_properties() # type: typing.Dict
final_list = {}
for name, prop in props.items():
try:
# serializer = Serializers.for_object(self.owner_inst.id, self.owner_cls.__name__, name)
# if name in self.db_commit_objects:
# continue
final_list[name] = prop
except Exception as ex:
self.owner_inst.logger.error(
f"could not deserialize property {name} due to error - {str(ex)}, skipping this property"
)
return final_list
@supports_only_instance_access("database operations are only supported at instance level")
def load_from_DB(self):
"""
Load and apply property values from database which have `db_init` or `db_persist` set to `True`
"""
if not hasattr(self.owner_inst, 'db_engine'):
return
# raise AttributeError("database engine not set, this object is not connected to a database")
missing_properties = self.owner_inst.db_engine.create_missing_properties(
self.db_init_objects,
get_missing_property_names=True
)
# 4. read db_init and db_persist objects
with edit_constant_parameters(self.owner_inst):
for db_prop, value in self.get_from_DB().items():
try:
if db_prop not in missing_properties:
setattr(self.owner_inst, db_prop, value) # type: ignore
except Exception as ex:
self.owner_inst.logger.error(f"could not set attribute {db_prop} due to error {str(ex)}")
@classmethod
def get_type_from_name(cls, name: str) -> typing.Type[Property]:
return Property
@supports_only_instance_access("additional property setup is required only for instances")
def _setup_parameters(self, **parameters):
"""
Initialize default and keyword parameter values.
First, ensures that all Parameters with 'deepcopy_default=True'
(typically used for mutable Parameters) are copied directly
into each object, to ensure that there is an independent copy
(to avoid surprising aliasing errors). Then sets each of the
keyword arguments, warning when any of them are not defined as
parameters.
Constant Parameters can be set during calls to this method.
"""
## Deepcopy all 'deepcopy_default=True' parameters
# (building a set of names first to avoid redundantly
# instantiating a later-overridden parent class's parameter)
param_default_values_to_deepcopy = {}
param_descriptors_to_deepcopy = {}
for (k, v) in self.owner_cls.properties.descriptors.items():
if v.deepcopy_default and k != "name":
# (avoid replacing name with the default of None)
param_default_values_to_deepcopy[k] = v
if v.per_instance_descriptor and k != "name":
param_descriptors_to_deepcopy[k] = v
for p in param_default_values_to_deepcopy.values():
self._deep_copy_param_default(p)
for p in param_descriptors_to_deepcopy.values():
self._deep_copy_param_descriptor(p)
## keyword arg setting
if len(parameters) > 0:
descs = self.descriptors
for name, val in parameters.items():
desc = descs.get(name, None) # pylint: disable-msg=E1101
if desc:
setattr(self.owner_inst, name, val)
# Its erroneous to set a non-descriptor (& non-param-descriptor) with a value from init.
# we dont know what that value even means, so we silently ignore
@supports_only_instance_access("additional property setup is required only for instances")
def _deep_copy_param_default(self, param_obj : 'Parameter') -> None:
# deepcopy param_obj.default into self.__dict__ (or dict_ if supplied)
# under the parameter's _internal_name (or key if supplied)
_old = self.owner_inst.__dict__.get(param_obj._internal_name, NotImplemented)
_old = _old if _old is not NotImplemented else param_obj.default
new_object = copy.deepcopy(_old)
# remember : simply setting in the dict does not activate post setter and remaining logic which is sometimes important
self.owner_inst.__dict__[param_obj._internal_name] = new_object
@supports_only_instance_access("additional property setup is required only for instances")
def _deep_copy_param_descriptor(self, param_obj : Parameter):
param_obj_copy = copy.deepcopy(param_obj)
self._instance_params[param_obj.name] = param_obj_copy