refactor: use classmethods

Signed-off-by: Alexander Tkachev <sasha64sasha@gmail.com>
This commit is contained in:
Alexander Tkachev 2022-07-25 01:26:10 +03:00
parent a8872b9808
commit fbc063244b
3 changed files with 24 additions and 28 deletions

View File

@ -25,8 +25,8 @@ class CloudEvent:
""" """
raise NotImplementedError() raise NotImplementedError()
@property @classmethod
def attributes(self) -> typing.Dict[str, typing.Any]: def get_attributes(cls, event: "CloudEvent") -> typing.Dict[str, typing.Any]:
""" """
:return: Attributes of this event. :return: Attributes of this event.
@ -35,8 +35,8 @@ class CloudEvent:
""" """
raise NotImplementedError() raise NotImplementedError()
@property @classmethod
def data(self) -> typing.Optional[typing.Any]: def get_data(cls, event: "CloudEvent") -> typing.Optional[typing.Any]:
""" """
:return: Data value of the event. :return: Data value of the event.
You MUST NOT mutate this dict. You MUST NOT mutate this dict.
@ -46,7 +46,9 @@ class CloudEvent:
def __eq__(self, other: typing.Any) -> bool: def __eq__(self, other: typing.Any) -> bool:
if isinstance(other, CloudEvent): if isinstance(other, CloudEvent):
return self.data == other.data and self.attributes == other.attributes same_data = self.get_data(self) == other.get_data(other)
same_attributes = self.get_attributes(self) == other.get_attributes(other)
return same_data and same_attributes
return False return False
def __getitem__(self, key: str) -> typing.Any: def __getitem__(self, key: str) -> typing.Any:
@ -56,7 +58,7 @@ class CloudEvent:
:param key: The event attribute name. :param key: The event attribute name.
:return: The event attribute value. :return: The event attribute value.
""" """
return self.attributes[key] return self.get_attributes(self)[key]
def get( def get(
self, key: str, default: typing.Optional[typing.Any] = None self, key: str, default: typing.Optional[typing.Any] = None
@ -72,19 +74,21 @@ class CloudEvent:
no attribute with the given key exists. no attribute with the given key exists.
:returns: The event attribute value if exists, default value otherwise. :returns: The event attribute value if exists, default value otherwise.
""" """
return self.attributes.get(key, default) return self.get_attributes(self).get(key, default)
def __iter__(self) -> typing.Iterator[typing.Any]: def __iter__(self) -> typing.Iterator[typing.Any]:
return iter(self.attributes) return iter(self.get_attributes(self))
def __len__(self) -> int: def __len__(self) -> int:
return len(self.attributes) return len(self.get_attributes(self))
def __contains__(self, key: str) -> bool: def __contains__(self, key: str) -> bool:
return key in self.attributes return key in self.get_attributes(self)
def __repr__(self) -> str: def __repr__(self) -> str:
return str({"attributes": self.attributes, "data": self.data}) return str(
{"attributes": self.get_attributes(self), "data": self.get_data(self)}
)
AnyCloudEvent = TypeVar("AnyCloudEvent", bound=CloudEvent) AnyCloudEvent = TypeVar("AnyCloudEvent", bound=CloudEvent)

View File

@ -52,7 +52,7 @@ class CloudEvent(abstract.CloudEvent):
:type data: typing.Any :type data: typing.Any
""" """
self._attributes = {k.lower(): v for k, v in attributes.items()} self._attributes = {k.lower(): v for k, v in attributes.items()}
self._data = data self.data = data
if "specversion" not in self._attributes: if "specversion" not in self._attributes:
self._attributes["specversion"] = "1.0" self._attributes["specversion"] = "1.0"
if "id" not in self._attributes: if "id" not in self._attributes:
@ -74,21 +74,13 @@ class CloudEvent(abstract.CloudEvent):
f"Missing required keys: {required_set - self._attributes.keys()}" f"Missing required keys: {required_set - self._attributes.keys()}"
) )
@property @classmethod
def attributes(self) -> typing.Dict[str, typing.Any]: def get_attributes(cls, event: "CloudEvent") -> typing.Dict[str, typing.Any]:
return self._attributes return event._attributes
@property @classmethod
def data(self) -> typing.Optional[typing.Any]: def get_data(cls, event: "CloudEvent") -> typing.Optional[typing.Any]:
return self._data return event.data
@data.setter
def data(self, value) -> None:
"""
Exists for backwards compatibility
:param value: new data vale
"""
self._data = value
def __setitem__(self, key: str, value: typing.Any) -> None: def __setitem__(self, key: str, value: typing.Any) -> None:
self._attributes[key] = value self._attributes[key] = value

View File

@ -16,7 +16,7 @@ def test_data_is_abstract():
exists mainly for coverage reasons exists mainly for coverage reasons
""" """
with pytest.raises(NotImplementedError): with pytest.raises(NotImplementedError):
CloudEvent().data CloudEvent.get_data(CloudEvent())
def test_attributes_is_abstract(): def test_attributes_is_abstract():
@ -24,4 +24,4 @@ def test_attributes_is_abstract():
exists mainly for coverage reasons exists mainly for coverage reasons
""" """
with pytest.raises(NotImplementedError): with pytest.raises(NotImplementedError):
CloudEvent().attributes CloudEvent.get_attributes(CloudEvent())