Coverage for filip/models/ngsi_ld/context.py: 91%
254 statements
« prev ^ index » next coverage.py v7.10.2, created at 2025-08-05 11:07 +0000
1"""
2NGSI LD models for context broker interaction
3"""
5import logging
6from typing import Any, List, Dict, Union, Optional
7import re
8from geojson_pydantic import (
9 Point,
10 MultiPoint,
11 LineString,
12 MultiLineString,
13 Polygon,
14 MultiPolygon,
15)
16from typing_extensions import Self
17from aenum import Enum
18from pydantic import field_validator, ConfigDict, BaseModel, Field, model_validator
19from filip.models.ngsi_v2 import ContextEntity
20from filip.utils.validators import (
21 FiwareRegex,
22 validate_fiware_datatype_string_protect,
23 validate_fiware_standard_regex,
24)
25from pydantic_core import ValidationError
28class DataTypeLD(str, Enum):
29 """
30 In NGSI-LD the data types on context entities are only divided into properties and relationships.
31 """
33 _init_ = "value __doc__"
34 GEOPROPERTY = "GeoProperty", "A property that represents a geometry value"
35 PROPERTY = "Property", "All attributes that do not represent a relationship"
36 RELATIONSHIP = (
37 "Relationship",
38 "Reference to another context entity, which can be identified with a URN.",
39 )
42# NGSI-LD entity models
43class ContextProperty(BaseModel):
44 """
45 The model for a property is represented by a JSON object with the following syntax:
47 The attribute value is specified by the value, whose value can be any data type. This does not need to be
48 specified further.
50 The NGSI type of the attribute is fixed and does not need to be specified.
51 Example:
53 >>> data = {"value": <...>}
55 >>> attr = ContextProperty(**data)
57 """
59 model_config = ConfigDict(extra="allow") # In order to allow nested properties
60 type: Optional[str] = Field(default="Property", title="type", frozen=True)
61 value: Optional[
62 Union[
63 Union[float, int, bool, str, List, Dict[str, Any]],
64 List[Union[float, int, bool, str, List, Dict[str, Any]]],
65 ]
66 ] = Field(default=None, title="Property value", description="the actual data")
67 observedAt: Optional[str] = Field(
68 None,
69 title="Timestamp",
70 description="Representing a timestamp for the "
71 "incoming value of the property.",
72 max_length=256,
73 min_length=1,
74 )
75 field_validator("observedAt")(validate_fiware_datatype_string_protect)
77 createdAt: Optional[str] = Field(
78 None,
79 title="Timestamp",
80 description="Representing a timestamp for the "
81 "creation time of the property.",
82 max_length=256,
83 min_length=1,
84 )
85 field_validator("createdAt")(validate_fiware_datatype_string_protect)
87 modifiedAt: Optional[str] = Field(
88 None,
89 title="Timestamp",
90 description="Representing a timestamp for the "
91 "last modification of the property.",
92 max_length=256,
93 min_length=1,
94 )
95 field_validator("modifiedAt")(validate_fiware_datatype_string_protect)
97 unitCode: Optional[str] = Field(
98 None,
99 title="Unit Code",
100 description="Representing the unit of the value. "
101 "Should be part of the defined units "
102 "by the UN/ECE Recommendation No. 21"
103 "https://unece.org/fileadmin/DAM/cefact/recommendations/rec20/rec20_rev3_Annex2e.pdf ",
104 max_length=256,
105 min_length=1,
106 )
107 field_validator("unitCode")(validate_fiware_datatype_string_protect)
109 datasetId: Optional[str] = Field(
110 None,
111 title="dataset Id",
112 description="It allows identifying a set or group of property values",
113 max_length=256,
114 min_length=1,
115 )
116 field_validator("datasetId")(validate_fiware_datatype_string_protect)
118 @classmethod
119 def get_model_fields_set(cls):
120 """
121 Get all names and aliases of the model fields.
122 """
123 return set(
124 [field.validation_alias for (_, field) in cls.model_fields.items()]
125 + [field_name for field_name in cls.model_fields]
126 )
128 @field_validator("type")
129 @classmethod
130 def check_property_type(cls, value):
131 """
132 Force property type to be "Property"
133 Args:
134 value: value field
135 Returns:
136 value
137 """
138 valid_property_types = ["Property", "Relationship", "TemporalProperty"]
139 if value not in valid_property_types:
140 msg = (
141 f"NGSI_LD Properties must have type {valid_property_types}, "
142 f'not "{value}"'
143 )
144 logging.warning(msg=msg)
145 raise ValueError(msg)
146 return value
149class NamedContextProperty(ContextProperty):
150 """
151 Context properties are properties of context entities. For example, the current speed of a car could be modeled
152 as the current_speed property of the car-104 entity.
154 In the NGSI-LD data model, properties have a name, the type "property" and a value.
155 """
157 name: str = Field(
158 title="Property name",
159 description="The property name describes what kind of property the "
160 "attribute value represents of the entity, for example "
161 "current_speed. Allowed characters "
162 "are the ones in the plain ASCII set, except the following "
163 "ones: control characters, whitespace, &, ?, / and #.",
164 max_length=256,
165 min_length=1,
166 )
167 field_validator("name")(validate_fiware_datatype_string_protect)
170class ContextGeoPropertyValue(BaseModel):
171 """
172 The value for a Geo property is represented by a JSON object with the following syntax:
174 A type with value "Point" and the
175 coordinates with a list containing the coordinates as value
177 Example:
178 "value": {
179 "type": "Point",
180 "coordinates": [
181 -3.80356167695194,
182 43.46296641666926
183 ]
184 }
185 }
187 """
189 type: Optional[str] = Field(default=None, title="type", frozen=True)
190 model_config = ConfigDict(extra="allow")
192 @model_validator(mode="after")
193 def check_geoproperty_value(self) -> Self:
194 """
195 Check if the value is a valid GeoProperty
196 """
197 if self.model_dump().get("type") == "Point":
198 return Point(**self.model_dump())
199 elif self.model_dump().get("type") == "LineString":
200 return LineString(**self.model_dump())
201 elif self.model_dump().get("type") == "Polygon":
202 return Polygon(**self.model_dump())
203 elif self.model_dump().get("type") == "MultiPoint":
204 return MultiPoint(**self.model_dump())
205 elif self.model_dump().get("type") == "MultiLineString":
206 return MultiLineString(**self.model_dump())
207 elif self.model_dump().get("type") == "MultiPolygon":
208 return MultiPolygon(**self.model_dump())
209 elif self.model_dump().get("type") == "GeometryCollection":
210 raise ValueError("GeometryCollection is not supported")
213class ContextGeoProperty(BaseModel):
214 """
215 The model for a Geo property is represented by a JSON object with the following syntax:
217 The attribute value is a JSON object with two contents.
219 Example:
221 {
222 "type": "GeoProperty",
223 "value": {
224 "type": "Point",
225 "coordinates": [
226 -3.80356167695194,
227 43.46296641666926
228 ]
229 }
231 """
233 model_config = ConfigDict(extra="allow")
234 type: Optional[str] = Field(default="GeoProperty", title="type", frozen=True)
235 value: Optional[
236 Union[
237 ContextGeoPropertyValue,
238 Point,
239 LineString,
240 Polygon,
241 MultiPoint,
242 MultiPolygon,
243 MultiLineString,
244 ]
245 ] = Field(default=None, title="GeoProperty value", description="the actual data")
246 observedAt: Optional[str] = Field(
247 default=None,
248 title="Timestamp",
249 description="Representing a timestamp for the "
250 "incoming value of the property.",
251 max_length=256,
252 min_length=1,
253 )
254 field_validator("observedAt")(validate_fiware_datatype_string_protect)
256 datasetId: Optional[str] = Field(
257 None,
258 title="dataset Id",
259 description="It allows identifying a set or group of property values",
260 max_length=256,
261 min_length=1,
262 )
263 field_validator("datasetId")(validate_fiware_datatype_string_protect)
266class NamedContextGeoProperty(ContextGeoProperty):
267 """
268 Context GeoProperties are geo properties of context entities. For example, the coordinates of a building .
270 In the NGSI-LD data model, properties have a name, the type "Geoproperty" and a value.
271 """
273 name: str = Field(
274 title="Property name",
275 description="The property name describes what kind of property the "
276 "attribute value represents of the entity, for example "
277 "current_speed. Allowed characters "
278 "are the ones in the plain ASCII set, except the following "
279 "ones: control characters, whitespace, &, ?, / and #.",
280 max_length=256,
281 min_length=1,
282 )
283 field_validator("name")(validate_fiware_datatype_string_protect)
286class ContextRelationship(BaseModel):
287 """
288 The model for a relationship is represented by a JSON object with the following syntax:
290 The attribute value is specified by the object, whose value can be a reference to another context entity. This
291 should be specified as the URN. The existence of this entity is not assumed.
293 The NGSI type of the attribute is fixed and does not need to be specified.
295 Example:
297 >>> data = {"object": <...>}
299 >>> attr = ContextRelationship(**data)
301 """
303 model_config = ConfigDict(extra="allow") # In order to allow nested relationships
304 type: Optional[str] = Field(default="Relationship", title="type", frozen=True)
305 object: Optional[
306 Union[
307 Union[float, int, bool, str, List, Dict[str, Any]],
308 List[Union[float, int, bool, str, List, Dict[str, Any]]],
309 ]
310 ] = Field(
311 default=None, title="Realtionship object", description="the actual object id"
312 )
314 datasetId: Optional[str] = Field(
315 None,
316 title="dataset Id",
317 description="It allows identifying a set or group of property values",
318 max_length=256,
319 min_length=1,
320 )
321 field_validator("datasetId")(validate_fiware_datatype_string_protect)
323 observedAt: Optional[str] = Field(
324 None,
325 titel="Timestamp",
326 description="Representing a timestamp for the "
327 "incoming value of the property.",
328 max_length=256,
329 min_length=1,
330 )
331 field_validator("observedAt")(validate_fiware_datatype_string_protect)
333 @field_validator("type")
334 @classmethod
335 def check_relationship_type(cls, value):
336 """
337 Force property type to be "Relationship"
338 Args:
339 value: value field
340 Returns:
341 value
342 """
343 if not value == "Relationship":
344 logging.warning(msg='NGSI_LD relationships must have type "Relationship"')
345 value = "Relationship"
346 return value
349class NamedContextRelationship(ContextRelationship):
350 """
351 Context Relationship are relations of context entities to each other.
352 For example, the current_speed of the entity car-104 could be modeled.
353 The location could be modeled as located_in the entity Room-001.
355 In the NGSI-LD data model, relationships have a name, the type "relationship" and an object.
356 """
358 name: str = Field(
359 title="Attribute name",
360 description="The attribute name describes what kind of property the "
361 "attribute value represents of the entity, for example "
362 "current_speed. Allowed characters "
363 "are the ones in the plain ASCII set, except the following "
364 "ones: control characters, whitespace, &, ?, / and #.",
365 max_length=256,
366 min_length=1,
367 # pattern=FiwareRegex.string_protect.value,
368 # Make it FIWARE-Safe
369 )
370 field_validator("name")(validate_fiware_datatype_string_protect)
373class ContextLDEntityBase(BaseModel):
374 """
375 Base Model for an entity is represented by a JSON object with the following
376 syntax.
378 The entity id is specified by the object's id property, whose value
379 is a string containing the entity id.
381 The entity type is specified by the object's type property, whose value
382 is a string containing the entity's type name.
383 """
385 model_config = ConfigDict(
386 extra="allow", validate_default=True, validate_assignment=True
387 )
388 id: str = Field(
389 ...,
390 title="Entity Id",
391 description="Id of an entity in an NGSI context broker. Allowed "
392 "characters are the ones in the plain ASCII set, except "
393 "the following ones: control characters, "
394 "whitespace, &, ?, / and #."
395 "the id should be structured according to the urn naming scheme.",
396 json_schema_extra={"example": "urn:ngsi-ld:Room:001"},
397 max_length=256,
398 min_length=1,
399 # pattern=FiwareRegex.standard.value, # Make it FIWARE-Safe
400 frozen=True,
401 )
402 field_validator("id")(validate_fiware_standard_regex)
403 type: str = Field(
404 ...,
405 title="Entity Type",
406 description="Id of an entity in an NGSI context broker. "
407 "Allowed characters are the ones in the plain ASCII set, "
408 "except the following ones: control characters, "
409 "whitespace, &, ?, / and #.",
410 json_schema_extra={"example": "Room"},
411 max_length=256,
412 min_length=1,
413 # pattern=FiwareRegex.standard.value, # Make it FIWARE-Safe
414 frozen=True,
415 )
416 field_validator("type")(validate_fiware_standard_regex)
417 context: Optional[Union[str, List[str], Dict]] = Field(
418 title="@context",
419 default=None,
420 description="The @context in JSON-LD is used to expand terms, provided as short "
421 "hand strings, to concepts, specified as URIs, and vice versa, "
422 "to compact URIs into terms "
423 "The main implication of NGSI-LD API is that if the @context is "
424 "a compound one, i.e. an @context which references multiple "
425 "individual @context, served by resources behind different URIs, "
426 "then a wrapper @context has to be created and hosted.",
427 examples=["https://n5geh.github.io/n5geh.test-context.io/context_saref.jsonld"],
428 alias="@context",
429 validation_alias="@context",
430 frozen=False,
431 )
434class ContextLDEntityKeyValues(ContextLDEntityBase):
435 """
436 Base Model for an entity is represented by a JSON object with the following
437 syntax.
439 The entity id is specified by the object's id property, whose value
440 is a string containing the entity id.
442 The entity type is specified by the object's type property, whose value
443 is a string containing the entity's type name.
445 """
447 model_config = ConfigDict(
448 extra="allow", validate_default=True, validate_assignment=True
449 )
451 def to_entity(self):
452 """
453 Convert the entity to a normalized representation.
454 """
455 return ContextLDEntity(
456 **{
457 "id": self.id,
458 "type": self.type,
459 "context": self.context if self.context else None,
460 **{
461 key: {
462 "type": "Property",
463 "value": value,
464 }
465 for key, value in self.model_dump().items()
466 if key not in ["id", "type", "context"]
467 },
468 }
469 )
472class PropertyFormat(str, Enum):
473 """
474 Format to decide if properties of ContextEntity class are returned as
475 List of NamedContextAttributes or as Dict of ContextAttributes.
476 """
478 LIST = "list"
479 DICT = "dict"
482class ContextLDEntity(ContextLDEntityBase):
483 """
484 Context LD entities, or simply entities, are the center of gravity in the
485 FIWARE NGSI-LD information model. An entity represents a thing, i.e., any
486 physical or logical object (e.g., a sensor, a person, a room, an issue in
487 a ticketing system, etc.). Each entity has an entity id.
488 Furthermore, the type system of FIWARE NGSI enables entities to have an
489 entity type. Entity types are semantic types; they are intended to describe
490 the type of thing represented by the entity. For example, a context
491 entity #with id sensor-365 could have the type temperatureSensor.
493 Each entity is uniquely identified by its id.
495 The entity id is specified by the object's id property, whose value
496 is a string containing the entity id.
498 The entity type is specified by the object's type property, whose value
499 is a string containing the entity's type name.
501 Entity attributes are specified by additional properties and relationships, whose names are
502 the name of the attribute and whose representation is described in the
503 "ContextProperty"/"ContextRelationship"-model. Obviously, id and type are
504 not allowed to be used as attribute names.
506 Example:
508 >>> data = {'id': 'MyId',
509 'type': 'MyType',
510 'my_attr': {'value': 20}}
512 >>> entity = ContextLDEntity(**data)
514 """
516 model_config = ConfigDict(
517 extra="allow",
518 validate_default=True,
519 validate_assignment=True,
520 populate_by_name=True,
521 )
523 observationSpace: Optional[ContextGeoProperty] = Field(
524 default=None,
525 title="Observation Space",
526 description="The geospatial Property representing "
527 "the geographic location that is being "
528 "observed, e.g. by a sensor. "
529 "For example, in the case of a camera, "
530 "the location of the camera and the "
531 "observationspace are different and "
532 "can be disjoint. ",
533 )
535 @field_validator("context")
536 @classmethod
537 def return_context(cls, context):
538 return context
540 operationSpace: Optional[ContextGeoProperty] = Field(
541 default=None,
542 title="Operation Space",
543 description="The geospatial Property representing "
544 "the geographic location in which an "
545 "Entity,e.g. an actuator is active. "
546 "For example, a crane can have a "
547 "certain operation space.",
548 )
550 createdAt: Optional[str] = Field(
551 None,
552 title="Timestamp",
553 description="Representing a timestamp for the "
554 "creation time of the property.",
555 max_length=256,
556 min_length=1,
557 )
558 field_validator("createdAt")(validate_fiware_datatype_string_protect)
560 modifiedAt: Optional[str] = Field(
561 None,
562 title="Timestamp",
563 description="Representing a timestamp for the "
564 "last modification of the property.",
565 max_length=256,
566 min_length=1,
567 )
568 field_validator("modifiedAt")(validate_fiware_datatype_string_protect)
570 def __init__(self, **data):
571 # There is currently no validation for extra fields
572 data.update(self._validate_attributes(data))
573 super().__init__(**data)
575 @classmethod
576 def get_model_fields_set(cls):
577 """
578 Get all names and aliases of the model fields.
579 """
580 return set(
581 [field.validation_alias for (_, field) in cls.model_fields.items()]
582 + [field_name for field_name in cls.model_fields]
583 )
585 @classmethod
586 def _validate_single_property(
587 cls, attr
588 ) -> Union[ContextProperty, ContextRelationship, ContextGeoProperty]:
589 # skip validation if pre-defined model is already used
590 if type(attr) in [ContextProperty, ContextRelationship, ContextGeoProperty]:
591 return attr
592 property_fields = ContextProperty.get_model_fields_set()
593 property_fields.remove(None)
594 # subattrs = {}
595 if attr.get("type") == "Relationship":
596 attr_instance = ContextRelationship.model_validate(attr)
597 elif attr.get("type") == "GeoProperty":
598 try:
599 attr_instance = ContextGeoProperty.model_validate(attr)
600 except Exception as e:
601 pass
602 elif attr.get("type") == "Property" or attr.get("type") is None:
603 attr_instance = ContextProperty.model_validate(attr)
604 else:
605 raise ValueError(f"Attribute {attr.get('type')} " "is not a valid type")
606 for subkey, subattr in attr.items():
607 if isinstance(subattr, dict) and subkey not in property_fields:
608 attr_instance.model_extra.update(
609 {subkey: cls._validate_single_property(attr=subattr)}
610 )
611 return attr_instance
613 @classmethod
614 def _validate_attributes(cls, data: Dict):
615 entity_fields = cls.get_model_fields_set()
616 entity_fields.remove(None)
617 # Initialize the attribute dictionary
618 attrs = {}
619 # Iterate through the data
620 for key, attr in data.items():
621 # Check if the keyword is not already present in the fields
622 if key not in entity_fields:
623 attrs[key] = cls._validate_single_property(attr=attr)
624 return attrs
626 def model_dump(self, *args, by_alias: bool = True, **kwargs):
627 return super().model_dump(*args, by_alias=by_alias, **kwargs)
629 @field_validator("id")
630 @classmethod
631 def _validate_id(cls, id: str):
632 if not id.startswith("urn:ngsi-ld:"):
633 logging.warning(
634 msg="It is recommended that the entity id to be a URN,"
635 'starting with the namespace "urn:ngsi-ld:"'
636 )
637 return id
639 def get_properties(
640 self, response_format: Union[str, PropertyFormat] = PropertyFormat.LIST
641 ) -> Union[List[NamedContextProperty], Dict[str, ContextProperty]]:
642 """
643 Get all properties of the entity.
644 Args:
645 response_format:
647 Returns:
649 """
650 response_format = PropertyFormat(response_format)
651 # response format dict:
652 if response_format == PropertyFormat.DICT:
653 final_dict = {}
654 for key, value in self.model_dump(exclude_unset=True).items():
655 if key not in ContextLDEntity.get_model_fields_set():
656 if value.get("type") != DataTypeLD.RELATIONSHIP:
657 if value.get("type") == DataTypeLD.GEOPROPERTY:
658 final_dict[key] = ContextGeoProperty(**value)
659 elif value.get("type") == DataTypeLD.PROPERTY:
660 final_dict[key] = ContextProperty(**value)
661 else: # named context property by default
662 final_dict[key] = ContextProperty(**value)
663 return final_dict
664 # response format list:
665 final_list = []
666 for key, value in self.model_dump(exclude_unset=True).items():
667 if key not in ContextLDEntity.get_model_fields_set():
668 if value.get("type") != DataTypeLD.RELATIONSHIP:
669 if value.get("type") == DataTypeLD.GEOPROPERTY:
670 final_list.append(NamedContextGeoProperty(name=key, **value))
671 elif value.get("type") == DataTypeLD.PROPERTY:
672 final_list.append(NamedContextProperty(name=key, **value))
673 else: # named context property by default
674 final_list.append(NamedContextProperty(name=key, **value))
675 return final_list
677 def delete_relationships(self, relationships: List[str]):
678 """
679 Delete the given relationships from the entity
681 Args:
682 relationships: List of relationship names
684 Returns:
686 """
687 all_relationships = self.get_relationships(response_format="dict")
688 for relationship in relationships:
689 # check they are relationships
690 if relationship not in all_relationships:
691 raise ValueError(f"Relationship {relationship} does not exist")
692 delattr(self, relationship)
694 def delete_properties(
695 self,
696 props: Union[Dict[str, ContextProperty], List[NamedContextProperty], List[str]],
697 ):
698 """
699 Delete the given properties from the entity
701 Args:
702 props: can be given in multiple forms
703 1) Dict: {"<property_name>": ContextProperty, ...}
704 2) List: [NamedContextProperty, ...]
705 3) List: ["<property_name>", ...]
707 Returns:
709 """
710 names: List[str] = []
711 if isinstance(props, list):
712 for entry in props:
713 if isinstance(entry, str):
714 names.append(entry)
715 elif isinstance(entry, NamedContextProperty):
716 names.append(entry.name)
717 else:
718 names.extend(list(props.keys()))
720 # check there are no relationships
721 relationship_names = [rel.name for rel in self.get_relationships()]
722 for name in names:
723 if name in relationship_names:
724 raise TypeError(f"{name} is a relationship")
726 for name in names:
727 delattr(self, name)
729 def add_geo_properties(
730 self, attrs: Union[Dict[str, ContextGeoProperty], List[NamedContextGeoProperty]]
731 ) -> None:
732 """
733 Add property to entity
734 Args:
735 attrs:
736 Returns:
737 None
738 """
739 if isinstance(attrs, list):
740 attrs = {
741 attr.name: ContextGeoProperty(
742 **attr.model_dump(exclude={"name"}, exclude_unset=True)
743 )
744 for attr in attrs
745 }
746 for key, attr in attrs.items():
747 self.__setattr__(name=key, value=attr)
749 def add_properties(
750 self, attrs: Union[Dict[str, ContextProperty], List[NamedContextProperty]]
751 ) -> None:
752 """
753 Add property to entity
754 Args:
755 attrs:
756 Returns:
757 None
758 """
759 if isinstance(attrs, list):
760 attrs = {
761 attr.name: ContextProperty(
762 **attr.model_dump(exclude={"name"}, exclude_unset=True)
763 )
764 for attr in attrs
765 }
766 for key, attr in attrs.items():
767 self.__setattr__(name=key, value=attr)
769 def add_relationships(
770 self,
771 relationships: Union[
772 Dict[str, ContextRelationship], List[NamedContextRelationship]
773 ],
774 ) -> None:
775 """
776 Add relationship to entity
777 Args:
778 relationships:
779 Returns:
780 None
781 """
782 if isinstance(relationships, list):
783 relationships = {
784 attr.name: ContextRelationship(**attr.dict(exclude={"name"}))
785 for attr in relationships
786 }
787 for key, attr in relationships.items():
788 self.__setattr__(name=key, value=attr)
790 def get_relationships(
791 self, response_format: Union[str, PropertyFormat] = PropertyFormat.LIST
792 ) -> Union[List[NamedContextRelationship], Dict[str, ContextRelationship]]:
793 """
794 Get all relationships of the context entity
796 Args:
797 response_format:
799 Returns:
801 """
802 response_format = PropertyFormat(response_format)
803 # response format dict:
804 if response_format == PropertyFormat.DICT:
805 final_dict = {}
806 for key, value in self.model_dump(exclude_unset=True).items():
807 if key not in ContextLDEntity.get_model_fields_set():
808 try:
809 if value.get("type") == DataTypeLD.RELATIONSHIP:
810 final_dict[key] = ContextRelationship(**value)
811 except AttributeError: # if context attribute
812 if isinstance(value, list):
813 pass
814 return final_dict
815 # response format list:
816 final_list = []
817 for key, value in self.model_dump(exclude_unset=True).items():
818 if key not in ContextLDEntity.get_model_fields_set():
819 if value.get("type") == DataTypeLD.RELATIONSHIP:
820 final_list.append(NamedContextRelationship(name=key, **value))
821 return final_list
823 def get_context(self):
824 """
825 Args:
826 response_format:
828 Returns: context of the entity as list
830 """
831 _, context = self.model_dump(include={"context"}).popitem()
832 if not context:
833 logging.warning("No context in entity")
834 return None
835 else:
836 return context
838 def to_keyvalues(self) -> ContextLDEntityKeyValues:
839 props = self.get_properties()
840 rels = self.get_relationships()
841 result = dict[str, Any]()
842 for prop in props:
843 result[prop.name] = prop.value
844 for rel in rels:
845 result[rel.name] = rel.object
846 return ContextLDEntityKeyValues(id=self.id, type=self.type, **result)
849class ActionTypeLD(str, Enum):
850 """
851 Options for queries
852 """
854 CREATE = "create"
855 UPSERT = "upsert"
856 UPDATE = "update"
857 DELETE = "delete"
860class UpdateLD(BaseModel):
861 """
862 Model for update action
863 """
865 entities: List[Union[ContextLDEntity, ContextLDEntityKeyValues]] = Field(
866 description="an array of entities, each entity specified using the "
867 "JSON entity representation format "
868 )
871class MessageLD(BaseModel):
872 """
873 Model for a notification message, when sent to other NGSIv2-APIs
874 """
876 subscriptionId: Optional[str] = Field(
877 default=None,
878 description="Id of the subscription the notification comes from",
879 )
880 data: List[ContextLDEntity] = Field(
881 description="is an array with the notification data itself which "
882 "includes the entity and all concerned attributes. Each "
883 "element in the array corresponds to a different entity. "
884 "By default, the entities are represented in normalized "
885 "mode. However, using the attrsFormat modifier, a "
886 "simplified representation mode can be requested."
887 )