refactor: convert get_data and get_attributes to private member functions

instead of classmethods

Signed-off-by: Alexander Tkachev <sasha64sasha@gmail.com>
This commit is contained in:
Alexander Tkachev 2022-08-05 17:09:17 +03:00
parent 0c2bafc423
commit 4e00b55062
3 changed files with 26 additions and 51 deletions

View File

@ -1,4 +1,5 @@
import typing
from abc import abstractmethod
from typing import TypeVar
@ -25,29 +26,34 @@ class CloudEvent:
"""
raise NotImplementedError()
@classmethod
def get_attributes(cls, event: "CloudEvent") -> typing.Dict[str, typing.Any]:
@abstractmethod
def _get_attributes(self) -> typing.Dict[str, typing.Any]:
"""
:return: Attributes of this event.
You MUST NOT mutate this dict.
Implementation MAY assume the dict will not be mutated.
Implementation MUST assume the returned value MAY be mutated.
The reason this function is not a property is to prevent possible issues
with different frameworks that assumes behaviours for properties
"""
raise NotImplementedError()
@classmethod
def get_data(cls, event: "CloudEvent") -> typing.Optional[typing.Any]:
@abstractmethod
def _get_data(self) -> typing.Optional[typing.Any]:
"""
:return: Data value of the event.
You MUST NOT mutate this dict.
Implementation MAY assume the dict will not be mutated.
Implementation MUST assume the returned value MAY be mutated.
The reason this function is not a property is to prevent possible issues
with different frameworks that assumes behaviours for properties
"""
raise NotImplementedError()
def __eq__(self, other: typing.Any) -> bool:
if isinstance(other, CloudEvent):
same_data = self.get_data(self) == other.get_data(other)
same_attributes = self.get_attributes(self) == other.get_attributes(other)
same_data = self._get_data() == other._get_data()
same_attributes = self._get_attributes() == other._get_attributes()
return same_data and same_attributes
return False
@ -58,7 +64,7 @@ class CloudEvent:
:param key: The event attribute name.
:return: The event attribute value.
"""
return self.get_attributes(self)[key]
return self._get_attributes()[key]
def get(
self, key: str, default: typing.Optional[typing.Any] = None
@ -74,21 +80,19 @@ class CloudEvent:
no attribute with the given key exists.
:returns: The event attribute value if exists, default value otherwise.
"""
return self.get_attributes(self).get(key, default)
return self._get_attributes().get(key, default)
def __iter__(self) -> typing.Iterator[typing.Any]:
return iter(self.get_attributes(self))
return iter(self._get_attributes())
def __len__(self) -> int:
return len(self.get_attributes(self))
return len(self._get_attributes())
def __contains__(self, key: str) -> bool:
return key in self.get_attributes(self)
return key in self._get_attributes()
def __repr__(self) -> str:
return str(
{"attributes": self.get_attributes(self), "data": self.get_data(self)}
)
return str({"attributes": self._get_attributes(), "data": self._get_data()})
AnyCloudEvent = TypeVar("AnyCloudEvent", bound=CloudEvent)

View File

@ -74,13 +74,11 @@ class CloudEvent(abstract.CloudEvent):
f"Missing required keys: {required_set - self._attributes.keys()}"
)
@classmethod
def get_attributes(cls, event: "CloudEvent") -> typing.Dict[str, typing.Any]:
return event._attributes
def _get_attributes(self) -> typing.Dict[str, typing.Any]:
return self._attributes
@classmethod
def get_data(cls, event: "CloudEvent") -> typing.Optional[typing.Any]:
return event.data
def _get_data(self) -> typing.Optional[typing.Any]:
return self.data
def __setitem__(self, key: str, value: typing.Any) -> None:
self._attributes[key] = value

View File

@ -1,27 +0,0 @@
import pytest
from cloudevents.abstract import CloudEvent
def test_create_is_abstract():
"""
exists mainly for coverage reasons
"""
with pytest.raises(NotImplementedError):
assert CloudEvent.create({}, None) is None
def test_data_is_abstract():
"""
exists mainly for coverage reasons
"""
with pytest.raises(NotImplementedError):
CloudEvent.get_data(CloudEvent())
def test_attributes_is_abstract():
"""
exists mainly for coverage reasons
"""
with pytest.raises(NotImplementedError):
CloudEvent.get_attributes(CloudEvent())