Coverage for filip/models/ngsi_v2/base.py: 86%
177 statements
« prev ^ index » next coverage.py v7.4.4, created at 2024-11-20 16:54 +0000
1"""
2Shared models that are used by multiple submodules
3"""
4import json
6from aenum import Enum
7from geojson_pydantic import (
8 Point,
9 MultiPoint,
10 LineString,
11 MultiLineString,
12 Polygon,
13 MultiPolygon,
14 Feature,
15 FeatureCollection,
16)
17from pydantic import (
18 field_validator,
19 model_validator,
20 ConfigDict,
21 AnyHttpUrl,
22 BaseModel,
23 Field,
24 model_serializer,
25 SerializationInfo,
26 ValidationInfo,
27)
29from typing import Union, Optional, Pattern, List, Dict, Any
31from filip.models.base import DataType
32from filip.models.ngsi_v2.units import validate_unit_data, Unit
33from filip.utils.simple_ql import QueryString, QueryStatement
34from filip.utils.validators import (
35 validate_escape_character_free,
36 validate_fiware_datatype_string_protect,
37 validate_fiware_datatype_standard,
38)
41class Http(BaseModel):
42 """
43 Model for notification and registrations sent or retrieved via HTTP
44 """
46 url: AnyHttpUrl = Field(
47 description="URL referencing the service to be invoked when a "
48 "notification is generated. An NGSIv2 compliant server "
49 "must support the http URL schema. Other schemas could "
50 "also be supported."
51 )
54class EntityPattern(BaseModel):
55 """
56 Entity pattern used to create subscriptions or registrations
57 """
59 id: Optional[str] = Field(default=None, pattern=r"\w")
60 idPattern: Optional[Pattern] = None
61 type: Optional[str] = Field(default=None, pattern=r"\w")
62 typePattern: Optional[Pattern] = None
64 @model_validator(mode="after")
65 def validate_conditions(self):
66 assert (self.id and not self.idPattern) or (not self.id and self.idPattern), (
67 "Both cannot be used at the same time, but one of 'id' or "
68 "'idPattern must' be present."
69 )
70 if self.type or self.model_dump().get("typePattern", None):
71 assert (self.type and not self.typePattern) or (
72 not self.type and self.typePattern
73 ), (
74 "Type or pattern of the affected entities. "
75 "Both cannot be used at the same time."
76 )
77 return self
80class Status(str, Enum):
81 """
82 Current status of a subscription or registrations
83 """
85 _init_ = "value __doc__"
87 ACTIVE = "active", "for active subscriptions"
88 INACTIVE = "inactive", "for inactive subscriptions"
89 FAILED = "failed", "for failed subscription"
90 EXPIRED = "expired", "for expired subscription"
93class Expression(BaseModel):
94 """
95 By means of a filtering expression, allows to express what is the scope
96 of the data provided.
97 https://telefonicaid.github.io/fiware-orion/api/v2/stable
98 """
100 model_config = ConfigDict(arbitrary_types_allowed=True)
102 q: Optional[Union[str, QueryString]] = Field(
103 default=None,
104 title="Simple Query Language: filter",
105 description="If filtering by attribute value (i.e. the expression is "
106 "used in a q query), the rest of tokens (if present) "
107 "represent the path to a sub-property of the target NGSI "
108 "attribute value (which should be a JSON object). Such "
109 "sub-property is defined as the target property.",
110 )
111 mq: Optional[Union[str, QueryString]] = Field(
112 default=None,
113 title="Simple Query Language: metadata filters",
114 description="If filtering by metadata (i.e. the expression is used in "
115 "a mq query), the second token represents a metadata name "
116 "associated to the target NGSI attribute, target "
117 "metadata, and the rest of tokens (if present) represent "
118 "the path to a sub-property of the target metadata value "
119 "(which should be a JSON object). Such sub-property is "
120 "defined as the target property. ",
121 )
122 georel: Optional[Union[str, QueryString]] = Field(
123 default=None,
124 title="Metadata filters",
125 description="Any of the geographical relationships as specified by "
126 "the Geoqueries section of this specification.",
127 )
128 geometry: Optional[Union[str, QueryString]] = Field(
129 default=None,
130 title="Metadata filters",
131 description="Any of the supported geometries as specified by the "
132 "Geoqueries section of this specification.",
133 )
134 coords: Optional[Union[str, QueryString]] = Field(
135 default=None,
136 title="Metadata filters",
137 description="String representation of coordinates as specified by the "
138 "Geoqueries section of the specification.",
139 )
141 @field_validator("q", "mq")
142 @classmethod
143 def validate_expressions(cls, v):
144 if isinstance(v, str):
145 return QueryString.parse_str(v)
147 @model_serializer(mode="wrap")
148 def serialize(self, serializer: Any, info: SerializationInfo):
149 if isinstance(self.q, (QueryString, QueryStatement)):
150 self.q = self.q.to_str()
151 if isinstance(self.mq, (QueryString, QueryStatement)):
152 self.mq = self.mq.to_str()
153 if isinstance(self.coords, (QueryString, QueryStatement)):
154 self.coords = self.coords.to_str()
155 if isinstance(self.georel, (QueryString, QueryStatement)):
156 self.georel = self.georel.to_str()
157 if isinstance(self.geometry, (QueryString, QueryStatement)):
158 self.geometry = self.geometry.to_str()
159 return serializer(self)
162class AttrsFormat(str, Enum):
163 """
164 Allowed options for attribute formats
165 """
167 _init_ = "value __doc__"
169 NORMALIZED = "normalized", "Normalized message representation"
170 KEY_VALUES = (
171 "keyValues",
172 "Key value message representation."
173 "This mode represents the entity "
174 "attributes by their values only, leaving out "
175 "the information about type and metadata. "
176 "See example below."
177 "Example: "
178 "{"
179 " 'id': 'R12345',"
180 " 'type': 'Room',"
181 " 'temperature': 22"
182 "}",
183 )
184 VALUES = (
185 "values",
186 "Key value message representation. "
187 "This mode represents the entity as an array of "
188 "attribute values. Information about id and type is "
189 "left out. See example below. The order of the "
190 "attributes in the array is specified by the attrs "
191 "URI param (e.g. attrs=branch,colour,engine). "
192 "If attrs is not used, the order is arbitrary. "
193 "Example:"
194 "[ 'Ford', 'black', 78.3 ]",
195 )
198# NGSIv2 entity models
199class Metadata(BaseModel):
200 """
201 Context metadata is used in FIWARE NGSI in several places, one of them being
202 an optional part of the attribute value as described above. Similar to
203 attributes, each piece of metadata has.
205 Note:
206 In NGSI it is not foreseen that metadata may contain nested metadata.
207 """
209 type: Optional[Union[DataType, str]] = Field(
210 default=None,
211 title="metadata type",
212 description="a metadata type, describing the NGSI value type of the "
213 "metadata value Allowed characters "
214 "are the ones in the plain ASCII set, except the following "
215 "ones: control characters, whitespace, &, ?, / and #.",
216 max_length=256,
217 min_length=1,
218 )
219 valid_type = field_validator("type")(validate_fiware_datatype_standard)
220 value: Optional[Any] = Field(
221 default=None,
222 title="metadata value",
223 description="a metadata value containing the actual metadata",
224 )
226 @field_validator("value")
227 def validate_value(cls, value, info: ValidationInfo):
228 assert json.dumps(value), "metadata not serializable"
230 if info.data.get("type").casefold() == "unit":
231 value = Unit.model_validate(value)
232 return value
235class NamedMetadata(Metadata):
236 """
237 Model for metadata including a name
238 """
240 name: str = Field(
241 titel="metadata name",
242 description="a metadata name, describing the role of the metadata in "
243 "the place where it occurs; for example, the metadata name "
244 "accuracy indicates that the metadata value describes how "
245 "accurate a given attribute value is. Allowed characters "
246 "are the ones in the plain ASCII set, except the following "
247 "ones: control characters, whitespace, &, ?, / and #.",
248 max_length=256,
249 min_length=1,
250 )
251 valid_name = field_validator("name")(validate_fiware_datatype_standard)
253 @model_validator(mode="after")
254 def validate_data(self):
255 if self.model_dump().get("name", "").casefold() in [
256 "unit",
257 "unittext",
258 "unitcode",
259 ]:
260 valide_dict = self.model_dump()
261 valide_dict.update(validate_unit_data(self.model_dump()))
262 return self
263 return self
265 def to_context_metadata(self):
266 return {self.name: Metadata(**self.model_dump())}
269class BaseAttribute(BaseModel):
270 """
271 Model for an attribute is represented by a JSON object with the following
272 syntax:
274 The attribute NGSI type is specified by the type property, whose value
275 is a string containing the NGSI type.
277 The attribute metadata is specified by the metadata property. Its value
278 is another JSON object which contains a property per metadata element
279 defined (the name of the property is the name of the metadata element).
280 Each metadata element, in turn, is represented by a JSON object
281 containing the following properties:
283 Values of entity attributes. For adding it you need to nest it into a
284 dict in order to give it a name.
286 Example:
288 >>> data = {"type": <...>,
289 "metadata": <...>}
290 >>> attr = BaseAttribute(**data)
292 """
294 model_config = ConfigDict(validate_assignment=True)
295 type: Union[DataType, str] = Field(
296 default=DataType.TEXT,
297 description="The attribute type represents the NGSI value type of the "
298 "attribute value. Note that FIWARE NGSI has its own type "
299 "system for attribute values, so NGSI value types are not "
300 "the same as JSON types. Allowed characters "
301 "are the ones in the plain ASCII set, except the following "
302 "ones: control characters, whitespace, &, ?, / and #.",
303 max_length=256,
304 min_length=1,
305 )
306 valid_type = field_validator("type")(validate_fiware_datatype_string_protect)
307 metadata: Optional[
308 Union[
309 Dict[str, Metadata],
310 NamedMetadata,
311 List[NamedMetadata],
312 Dict[str, Dict[str, str]],
313 ]
314 ] = Field(
315 default={},
316 title="Metadata",
317 description="optional metadata describing properties of the attribute "
318 "value like e.g. accuracy, provider, or a timestamp",
319 )
321 @field_validator("metadata")
322 @classmethod
323 def validate_metadata_type(cls, value):
324 """validator for field 'metadata'"""
325 if type(value) == NamedMetadata:
326 value = [value]
327 elif isinstance(value, dict):
328 if all(isinstance(item, Metadata) for item in value.values()):
329 value = [
330 NamedMetadata(name=key, **item.model_dump())
331 for key, item in value.items()
332 ]
333 else:
334 json.dumps(value)
335 value = [NamedMetadata(name=key, **item) for key, item in value.items()]
337 if isinstance(value, list):
338 if all(isinstance(item, dict) for item in value):
339 value = [NamedMetadata(**item) for item in value]
340 if all(isinstance(item, NamedMetadata) for item in value):
341 return {
342 item.name: Metadata(**item.model_dump(exclude={"name"}))
343 for item in value
344 }
346 raise TypeError(f"Invalid type {type(value)}")
349class BaseNameAttribute(BaseModel):
350 """
351 Model to add the name property to an BaseAttribute Model.
352 The attribute name describes what kind of property the
353 attribute value represents of the entity
354 """
356 name: str = Field(
357 titel="Attribute name",
358 description="The attribute name describes what kind of property the "
359 "attribute value represents of the entity, for example "
360 "current_speed. Allowed characters "
361 "are the ones in the plain ASCII set, except the following "
362 "ones: control characters, whitespace, &, ?, / and #.",
363 max_length=256,
364 min_length=1,
365 # Make it FIWARE-Safe
366 )
367 valid_name = field_validator("name")(validate_fiware_datatype_string_protect)
370class BaseValueAttribute(BaseModel):
371 """
372 Model to add the value property to an BaseAttribute Model. The Model
373 is represented by a JSON object with the following syntax:
376 The attribute value is specified by the value property, whose value may
377 be any JSON datatype.
378 """
380 type: Union[DataType, str] = Field(
381 default=DataType.TEXT,
382 description="The attribute type represents the NGSI value type of the "
383 "attribute value. Note that FIWARE NGSI has its own type "
384 "system for attribute values, so NGSI value types are not "
385 "the same as JSON types. Allowed characters "
386 "are the ones in the plain ASCII set, except the following "
387 "ones: control characters, whitespace, &, ?, / and #.",
388 max_length=256,
389 min_length=1,
390 )
391 valid_type = field_validator("type")(validate_fiware_datatype_string_protect)
392 value: Optional[Any] = Field(
393 default=None, title="Attribute value", description="the actual data"
394 )
396 @field_validator("value")
397 def validate_value_type(cls, value, info: ValidationInfo):
398 """
399 Validator for field 'value'
400 The validator will try autocast the value based on the given type.
401 If `DataType.STRUCTUREDVALUE` is used for type it will also serialize
402 pydantic models. With latter operation all additional features of the
403 original pydantic model will be dumped.
404 If the type is unknown it will check json-serializable.
405 """
407 type_ = info.data.get("type")
408 value_ = value
409 if isinstance(value, BaseModel):
410 value_ = value.model_dump()
411 validate_escape_character_free(value_)
413 if value not in (None, "", " "):
414 if type_ == DataType.TEXT:
415 if isinstance(value, list):
416 return [str(item) for item in value]
417 return str(value)
418 if type_ == DataType.BOOLEAN:
419 if isinstance(value, list):
420 return [bool(item) for item in value]
421 return bool(value)
422 if type_ in (DataType.NUMBER, DataType.FLOAT):
423 if isinstance(value, list):
424 return [float(item) for item in value]
425 return float(value)
426 if type_ == DataType.INTEGER:
427 if isinstance(value, list):
428 return [int(item) for item in value]
429 return int(value)
430 if type_ == DataType.DATETIME:
431 return value
432 # allows list
433 if type_ == DataType.ARRAY:
434 if isinstance(value, list):
435 return value
436 raise TypeError(f"{type(value)} does not match " f"{DataType.ARRAY}")
437 # allows dict and BaseModel as object
438 if type_ == DataType.OBJECT:
439 if isinstance(value, dict):
440 value = json.dumps(value)
441 return json.loads(value)
442 elif isinstance(value, BaseModel):
443 value.model_dump_json()
444 return value
445 raise TypeError(
446 f"{type(value)} does not match " f"{DataType.OBJECT}"
447 )
449 # allows geojson as structured value
450 if type_ == DataType.GEOJSON:
451 if isinstance(
452 value,
453 (
454 Point,
455 MultiPoint,
456 LineString,
457 MultiLineString,
458 Polygon,
459 MultiPolygon,
460 Feature,
461 FeatureCollection,
462 ),
463 ):
464 return value
465 if isinstance(value, dict):
466 _geo_json_type = value.get("type", None)
467 if _geo_json_type == "Point":
468 return Point(**value)
469 elif _geo_json_type == "MultiPoint":
470 return MultiPoint(**value)
471 elif _geo_json_type == "LineString":
472 return LineString(**value)
473 elif _geo_json_type == "MultiLineString":
474 return MultiLineString(**value)
475 elif _geo_json_type == "Polygon":
476 return Polygon(**value)
477 elif _geo_json_type == "MultiPolygon":
478 return MultiPolygon(**value)
479 elif _geo_json_type == "Feature":
480 return Feature(**value)
481 elif _geo_json_type == "FeatureCollection":
482 return FeatureCollection(**value)
483 raise TypeError(f"{type(value)} does not match "
484 f"{DataType.GEOJSON}")
486 # allows list, dict and BaseModel as structured value
487 if type_ == DataType.STRUCTUREDVALUE:
488 if isinstance(value, (dict, list)):
489 value = json.dumps(value)
490 return json.loads(value)
491 elif isinstance(value, BaseModel):
492 value.model_dump_json()
493 return value
494 raise TypeError(
495 f"{type(value)} does not match " f"{DataType.STRUCTUREDVALUE}"
496 )
498 # if none of the above, check if it is a pydantic model
499 if isinstance(value, BaseModel):
500 value.model_dump_json()
501 return value
503 # if none of the above, check if serializable. Hence, no further
504 # type check is performed
505 value = json.dumps(value)
506 return json.loads(value)
508 return value