Coverage for filip/models/base.py: 94%

1""" 

2Shared data models 

3""" 

4 

5from aenum import Enum 

6from pydantic import ConfigDict, BaseModel, Field, field_validator, computed_field 

7 

8from filip.utils.validators import validate_fiware_service_path, validate_fiware_service 

9 

10core_context = "https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.6.jsonld" 

11 

12 

13class NgsiVersion(str, Enum): 

14 """ 

15 Version of NGSI-Specification that should be used within the target system. 

16 Note: 

17 Currently, the library only supports functionality for NGSI-v2 

18 """ 

19 

20 v2 = "v2" 

21 ld = "ld" 

22 

23 

24class DataType(str, Enum): 

25 """ 

26 When possible reuse schema.org data types 

27 (Text, Number, DateTime, StructuredValue, etc.). 

28 Remember that null is not allowed in NGSI-LD and 

29 therefore should be avoided as a value. 

30 

31 https://schema.org/DataType 

32 """ 

33 

34 _init_ = "value __doc__" 

35 

36 BOOLEAN = "Boolean", "True or False." 

37 DATE = "Date", "A date value in ISO 8601 date format." 

38 DATETIME = ( 

39 "DateTime", 

40 "A combination of date and time of day in the form " 

41 "[-]CCYY-MM-DDThh:mm:ss[Z|(+|-)hh:mm] " 

42 "(see Chapter 5.4 of ISO 8601).", 

43 ) 

44 NUMBER = ( 

45 "Number", 

46 "Use values from 0123456789 (Unicode 'DIGIT ZERO' " 

47 "(U+0030) to 'DIGIT NINE' (U+0039)) rather than " 

48 "superficially similar Unicode symbols. Use '.' " 

49 "(Unicode 'FULL STOP' (U+002E)) rather than ',' to " 

50 "indicate a decimal point. Avoid using these symbols " 

51 "as a readability separator.", 

52 ) 

53 INTEGER = "Integer", "Integer number" 

54 FLOAT = "Float", "Floating number. Please check 'DataType.Number'" 

55 TEXT = "Text", "https://schema.org/Text" 

56 TIME = ( 

57 "Time", 

58 "A point in time recurring on multiple days in the form " 

59 "hh:mm:ss[Z|(+|-)hh:mm] (see XML schema for details).", 

60 ) 

61 RELATIONSHIP = "Relationship", "Reference to another context entity" 

62 STRUCTUREDVALUE = "StructuredValue", ( 

63 "Structured datatype must be an " "array or object" "serializable" 

64 ) 

65 ARRAY = "Array", "Array of the types above" 

66 OBJECT = "Object", "JSON-Object of the types above, i.e. a dictionary" 

67 COMMAND = "command", "A command for IoT Devices" 

68 COMMAND_RESULT = ( 

69 "commandResult", 

70 "An entity containing a command, " 

71 "contains an autogenerated attribute" 

72 "of this type", 

73 ) 

74 COMMAND_STATUS = ( 

75 "commandStatus", 

76 "An entity containing a command, " 

77 "contains an autogenerated attribute " 

78 "of this type", 

79 ) 

80 GEOJSON = "geo:json", "GeoJson object" 

81 

82 

83class PaginationMethod(str, Enum): 

84 """ 

85 Options for the internal pagination methods 

86 """ 

87 

88 GET = "GET" 

89 POST = "POST" 

90 

91 

92class FiwareHeader(BaseModel): 

93 """ 

94 Define entity service paths which are supported by the NGSI 

95 Context Brokers to support hierarchical scopes: 

96 https://fiware-orion.readthedocs.io/en/master/user/service_path/index.html 

97 """ 

98 

99 model_config = ConfigDict(populate_by_name=True, validate_assignment=True) 

100 

101 service: str = Field( 

102 alias="fiware-service", 

103 default="", 

104 max_length=50, 

105 description="Fiware service used for multi-tenancy", 

106 pattern=r"\w*$", 

107 ) 

108 service_path: str = Field( 

109 alias="fiware-servicepath", 

110 default="/", 

111 description="Fiware service path", 

112 max_length=51, 

113 ) 

114 valid_service = field_validator("service")(validate_fiware_service) 

115 valid_service_path = field_validator("service_path")(validate_fiware_service_path) 

116 

117 

118class FiwareHeaderSecure(FiwareHeader): 

119 """ 

120 Defines entity service paths and an authorization via Bearer-Token which are 

121 supported by the NGSI 

122 Context Brokers to support hierarchical scopes: 

123 https://fiware-orion.readthedocs.io/en/master/user/service_path/index.html 

124 """ 

125 

126 authorization: str = Field( 

127 alias="authorization", 

128 default="", 

129 max_length=3000, 

130 description="authorization key", 

131 pattern=r".*", 

132 ) 

133 

134 

135class LogLevel(str, Enum): 

136 CRITICAL = "CRITICAL" 

137 ERROR = "ERROR" 

138 WARNING = "WARNING" 

139 INFO = "INFO" 

140 DEBUG = "DEBUG" 

141 NOTSET = "NOTSET" 

142 

143 @classmethod 

144 def _missing_name_(cls, name): 

145 """ 

146 Class method to realize case-insensitive args 

147 

148 Args: 

149 name: missing argument 

150 

151 Returns: 

152 valid member of enum 

153 """ 

154 for member in cls: 

155 if member.value.casefold() == name.casefold(): 

156 return member 

157 

158 

159class FiwareLDHeader(BaseModel): 

160 """ 

161 Define entity service paths which are supported by the NGSI 

162 Context Brokers to support hierarchical scopes: 

163 https://fiware-orion.readthedocs.io/en/master/user/service_path/index.html 

164 """ 

165 

166 model_config = ConfigDict(populate_by_name=True, validate_assignment=True) 

167 ngsild_tenant: str = Field( 

168 alias="NGSILD-Tenant", 

169 default=None, 

170 max_length=50, 

171 description="Alias to the Fiware service to used for multitenancy", 

172 pattern=r"\w*$", 

173 ) 

174 link_header: str = Field( 

175 alias="Link", 

176 default=f"<{core_context}>; " 

177 'rel="http://www.w3.org/ns/json-ld#context"; ' 

178 'type="application/ld+json"', 

179 description="Fiware service used for multi-tenancy", 

180 pattern=r"\w*$", 

181 ) 

182 # @computed_field 

183 # def Link(self) -> str: 

184 # link_header = f'<{self.context}>; ' \ 

185 # 'rel="http://www.w3.org/ns/json-ld#context"; ' \ 

186 # 'type="application/ld+json"' 

187 # return link_header 

188 

189 def set_context(self, context: str): 

190 self.link_header = f'<{context}>; rel="http://www.w3.org/ns/json-ld#context"; type="application/ld+json"'