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