class Event:
"""
Asynchronously push arbitrary messages to clients. Apart from default events created by the package (like state
change event, observable properties etc.), events are supposed to be created at class level or at ``__init__``
as a instance attribute, otherwise their publishing socket is unbound and will lead to ``AttributeError``.
Parameters
----------
name: str
name of the event, specified name may contain dashes and can be used on client side to subscribe to this event.
doc: str
docstring for the event
schema: JSON
schema of the event, if the event is JSON complaint. HTTP clients can validate the data with this schema. There
is no validation on server side.
"""
# security: Any
# security necessary to access this event.
__slots__ = ['friendly_name', 'name', '_internal_name', '_publisher', '_observable',
'doc', 'schema', 'security', 'label', 'owner']
def __init__(self, friendly_name : str, doc : typing.Optional[str] = None,
schema : typing.Optional[JSON] = None, # security : typing.Optional[BaseSecurityDefinition] = None,
label : typing.Optional[str] = None) -> None:
self.friendly_name = friendly_name
self.doc = doc
if global_config.validate_schemas and schema:
jsonschema.Draft7Validator.check_schema(schema)
self.schema = schema
# self.security = security
self.label = label
self._publisher = None # type: typing.Optional["EventPublisher"]
self._observable = False
def __set_name__(self, owner : ParameterizedMetaclass, name : str) -> None:
self._internal_name = f"{pep8_to_dashed_name(name)}-dispatcher"
self.name = name
self.owner = owner
@typing.overload
def __get__(self, obj, objtype) -> "EventDispatcher":
...
def __get__(self, obj : ParameterizedMetaclass, objtype : typing.Optional[type] = None):
try:
if not obj:
return self
return obj.__dict__[self._internal_name]
except KeyError:
raise AttributeError("Event object not yet initialized, please dont access now." +
" Access after Thing is running.")
def __set__(self, obj : Parameterized, value : typing.Any) -> None:
if isinstance(value, EventDispatcher):
if not obj.__dict__.get(self._internal_name, None):
value._owner_inst = obj
obj.__dict__[self._internal_name] = value
else:
raise AttributeError(f"Event object already assigned for {self.name}. Cannot reassign.")
# may be allowing to reassign is not a bad idea
else:
raise TypeError(f"Supply EventDispatcher object to event {self.name}, not type {type(value)}.")