12
12
# License for the specific language governing permissions and limitations
13
13
# under the License.
14
14
15
- from typing import Any , Optional
15
+ from typing import Any , Optional , Final
16
16
from datetime import datetime
17
17
import re
18
18
19
- REQUIRED_ATTRIBUTES = {"id" , "source" , "type" , "specversion" }
20
- OPTIONAL_ATTRIBUTES = {"datacontenttype" , "dataschema" , "subject" , "time" }
19
+ REQUIRED_ATTRIBUTES : Final [set [str ]] = {"id" , "source" , "type" , "specversion" }
20
+ OPTIONAL_ATTRIBUTES : Final [set [str ]] = {
21
+ "datacontenttype" ,
22
+ "dataschema" ,
23
+ "subject" ,
24
+ "time" ,
25
+ }
21
26
22
27
23
28
class CloudEvent :
24
- def __init__ (self , attributes : dict , data : Optional [dict ] = None ) -> None :
29
+ """
30
+ The CloudEvent Python wrapper contract exposing generically-available
31
+ properties and APIs.
32
+
33
+ Implementations might handle fields and have other APIs exposed but are
34
+ obliged to follow this contract.
35
+ """
36
+
37
+ def __init__ (self , attributes : dict [str , Any ], data : Optional [dict ] = None ) -> None :
25
38
"""
26
39
Create a new CloudEvent instance.
27
40
28
41
:param attributes: The attributes of the CloudEvent instance.
29
- :type attributes: dict
30
42
:param data: The payload of the CloudEvent instance.
31
- :type data: Optional[dict]
32
43
33
44
:raises ValueError: If any of the required attributes are missing or have invalid values.
34
45
:raises TypeError: If any of the attributes have invalid types.
35
46
"""
36
47
self ._validate_attribute (attributes )
37
- self ._attributes = attributes
38
- self ._data = data
48
+ self ._attributes : dict = attributes
49
+ self ._data : Optional [ dict ] = data
39
50
40
- def _validate_attribute (self , attributes : dict ) -> None :
51
+ @staticmethod
52
+ def _validate_attribute (attributes : dict ) -> None :
41
53
"""
42
- Private method that validates the attributes of the CloudEvent as per the CloudEvents specification.
54
+ Validates the attributes of the CloudEvent as per the CloudEvents specification.
55
+
56
+ See https://github.com/cloudevents/spec/blob/main/cloudevents/spec.md#required-attributes
43
57
"""
44
58
missing_attributes = [
45
59
attr for attr in REQUIRED_ATTRIBUTES if attr not in attributes
@@ -95,38 +109,31 @@ def _validate_attribute(self, attributes: dict) -> None:
95
109
if not attributes ["dataschema" ]:
96
110
raise ValueError ("Attribute 'dataschema' must not be empty" )
97
111
98
- for custom_extension in (
112
+ for extension_attributes in (
99
113
set (attributes .keys ()) - REQUIRED_ATTRIBUTES - OPTIONAL_ATTRIBUTES
100
114
):
101
- if custom_extension == "data" :
115
+ if extension_attributes == "data" :
102
116
raise ValueError (
103
117
"Extension attribute 'data' is reserved and must not be used"
104
118
)
105
119
106
- if not custom_extension [0 ].isalpha ():
107
- raise ValueError (
108
- f"Extension attribute '{ custom_extension } ' should start with a letter"
109
- )
110
-
111
- if not (5 <= len (custom_extension ) <= 20 ):
120
+ if not (1 <= len (extension_attributes ) <= 20 ):
112
121
raise ValueError (
113
- f"Extension attribute '{ custom_extension } ' should be between 5 and 20 characters long"
122
+ f"Extension attribute '{ extension_attributes } ' should be between 1 and 20 characters long"
114
123
)
115
124
116
- if not re .match (r"^[a-z0-9]+$" , custom_extension ):
125
+ if not re .match (r"^[a-z0-9]+$" , extension_attributes ):
117
126
raise ValueError (
118
- f"Extension attribute '{ custom_extension } ' should only contain lowercase letters and numbers"
127
+ f"Extension attribute '{ extension_attributes } ' should only contain lowercase letters and numbers"
119
128
)
120
129
121
130
def get_attribute (self , attribute : str ) -> Optional [Any ]:
122
131
"""
123
132
Retrieve a value of an attribute of the event denoted by the given `attribute`.
124
133
125
134
:param attribute: The name of the event attribute to retrieve the value for.
126
- :type attribute: str
127
135
128
136
:return: The event attribute value.
129
- :rtype: Optional[Any]
130
137
"""
131
138
return self ._attributes [attribute ]
132
139
@@ -135,6 +142,5 @@ def get_data(self) -> Optional[dict]:
135
142
Retrieve data of the event.
136
143
137
144
:return: The data of the event.
138
- :rtype: Optional[dict]
139
145
"""
140
146
return self ._data
0 commit comments