Coverage for filip/models/ngsi_v2/context.py: 96%

210 statements  

« prev     ^ index     » next       coverage.py v7.4.4, created at 2025-04-17 14:42 +0000

1""" 

2NGSIv2 models for context broker interaction 

3""" 

4 

5import json 

6from typing import Any, List, Dict, Union, Optional, Set, Tuple 

7 

8from aenum import Enum 

9from pydantic import ( 

10 field_validator, 

11 ConfigDict, 

12 BaseModel, 

13 Field, 

14 model_validator, 

15 SerializeAsAny, 

16) 

17from pydantic_core.core_schema import ValidationInfo 

18from pydantic.types import OnErrorOmit 

19from filip.models.ngsi_v2.base import ( 

20 EntityPattern, 

21 Expression, 

22 BaseAttribute, 

23 BaseValueAttribute, 

24 BaseNameAttribute, 

25) 

26from filip.models.base import DataType 

27from filip.utils.validators import ( 

28 validate_fiware_datatype_standard, 

29 validate_fiware_datatype_string_protect, 

30 validate_fiware_attribute_value_regex, 

31 validate_fiware_attribute_name_regex, 

32) 

33 

34 

35class GetEntitiesOptions(str, Enum): 

36 """Options for queries""" 

37 

38 _init_ = "value __doc__" 

39 

40 NORMALIZED = "normalized", "Normalized message representation" 

41 KEY_VALUES = ( 

42 "keyValues", 

43 "Key value message representation." 

44 "This mode represents the entity " 

45 "attributes by their values only, leaving out " 

46 "the information about type and metadata. " 

47 "See example " 

48 "below." 

49 "Example: " 

50 "{" 

51 " 'id': 'R12345'," 

52 " 'type': 'Room'," 

53 " 'temperature': 22" 

54 "}", 

55 ) 

56 VALUES = ( 

57 "values", 

58 "Key value message representation. " 

59 "This mode represents the entity as an array of " 

60 "attribute values. Information about id and type is " 

61 "left out. See example below. The order of the " 

62 "attributes in the array is specified by the attrs " 

63 "URI param (e.g. attrs=branch,colour,engine). " 

64 "If attrs is not used, the order is arbitrary. " 

65 "Example:" 

66 "[ 'Ford', 'black', 78.3 ]", 

67 ) 

68 UNIQUE = ( 

69 "unique", 

70 "unique mode. This mode is just like values mode, " 

71 "except that values are not repeated", 

72 ) 

73 

74 

75class PropertyFormat(str, Enum): 

76 """ 

77 Format to decide if properties of ContextEntity class are returned as 

78 List of NamedContextAttributes or as Dict of ContextAttributes. 

79 """ 

80 

81 LIST = "list" 

82 DICT = "dict" 

83 

84 

85class ContextAttribute(BaseAttribute, BaseValueAttribute): 

86 """ 

87 Model for an attribute is represented by a JSON object with the following 

88 syntax: 

89 

90 The attribute value is specified by the value property, whose value may 

91 be any JSON datatype. 

92 

93 The attribute NGSI type is specified by the type property, whose value 

94 is a string containing the NGSI type. 

95 

96 The attribute metadata is specified by the metadata property. Its value 

97 is another JSON object which contains a property per metadata element 

98 defined (the name of the property is the name of the metadata element). 

99 Each metadata element, in turn, is represented by a JSON object 

100 containing the following properties: 

101 

102 Values of entity attributes. For adding it you need to nest it into a 

103 dict in order to give it a name. 

104 

105 Example: 

106 

107 >>> data = {"value": <...>, 

108 "type": <...>, 

109 "metadata": <...>} 

110 >>> attr = ContextAttribute(**data) 

111 

112 """ 

113 

114 # although `type` is a required field in the NGSIv2 specification, it is 

115 # set to optional here to allow for the possibility of setting 

116 # default-types in child classes. Pydantic will raise the correct error 

117 # and also exports the correct json-schema. 

118 def __init__(self, type: str = None, **data): 

119 if type is None and self.model_fields["type"].default: 

120 type = self.model_fields["type"].default 

121 super().__init__(type=type, **data) 

122 

123 

124class NamedContextAttribute(ContextAttribute, BaseNameAttribute): 

125 """ 

126 Context attributes are properties of context entities. For example, the 

127 current speed of a car could be modeled as attribute current_speed of entity 

128 car-104. 

129 

130 In the NGSI data model, attributes have an attribute name, an attribute type 

131 an attribute value and metadata. 

132 """ 

133 

134 pass 

135 

136 

137class ContextEntityKeyValues(BaseModel): 

138 """ 

139 Base Model for an entity is represented by a JSON object with the following 

140 syntax. 

141 

142 The entity id is specified by the object's id property, whose value 

143 is a string containing the entity id. 

144 

145 The entity type is specified by the object's type property, whose value 

146 is a string containing the entity's type name. 

147 

148 """ 

149 

150 model_config = ConfigDict( 

151 extra="allow", validate_default=True, validate_assignment=True 

152 ) 

153 id: str = Field( 

154 ..., 

155 title="Entity Id", 

156 description="Id of an entity in an NGSI context broker. Allowed " 

157 "characters are the ones in the plain ASCII set, except " 

158 "the following ones: control characters, " 

159 "whitespace, &, ?, / and #.", 

160 json_schema_extra={"example": "Bcn-Welt"}, 

161 max_length=256, 

162 min_length=1, 

163 frozen=True, 

164 ) 

165 valid_id = field_validator("id")(validate_fiware_datatype_standard) 

166 type: Union[str, Enum] = Field( 

167 ..., 

168 title="Entity Type", 

169 description="Id of an entity in an NGSI context broker. " 

170 "Allowed characters are the ones in the plain ASCII set, " 

171 "except the following ones: control characters, " 

172 "whitespace, &, ?, / and #.", 

173 json_schema_extra={"example": "Room"}, 

174 max_length=256, 

175 min_length=1, 

176 frozen=True, 

177 ) 

178 valid_type = field_validator("type")(validate_fiware_datatype_standard) 

179 

180 # although `type` is a required field in the NGSIv2 specification, it is 

181 # set to optional here to allow for the possibility of setting 

182 # default-types in child classes. Pydantic will raise the correct error 

183 # and also exports the correct json-schema. 

184 def __init__(self, id: str, type: Union[str, Enum] = None, **data): 

185 # this allows to set the type of the entity in child classes 

186 if type is None: 

187 if isinstance(self.model_fields["type"].default, str): 

188 type = self.model_fields["type"].default 

189 else: 

190 # if this statement is reached not proper default-value for 

191 # `type` was found and pydantic will raise the correct error 

192 super().__init__(id=id, **data) 

193 # This will result in usual behavior 

194 data.update(self._validate_attributes(data)) 

195 super().__init__(id=id, type=type, **data) 

196 

197 # Validation of attributes 

198 @classmethod 

199 def _validate_attributes(cls, data: dict): 

200 """ 

201 Validate attribute name and value of the entity in keyvalues format 

202 """ 

203 for attr_name, attr_value in data.items(): 

204 if isinstance(attr_value, str): 

205 validate_fiware_attribute_value_regex(attr_value) 

206 validate_fiware_attribute_name_regex(attr_name) 

207 return data 

208 

209 def get_attributes(self) -> dict: 

210 """ 

211 Get the attribute of the entity with the given name in 

212 dict format 

213 

214 Returns: 

215 dict 

216 """ 

217 return self.model_dump(exclude={"id", "type"}) 

218 

219 def to_normalized(self): 

220 attrs = [] 

221 for key, value in self.get_attributes().items(): 

222 attr_type = ( 

223 DataType.NUMBER.value 

224 if isinstance(value, int) or isinstance(value, float) 

225 else ( 

226 DataType.TEXT.value 

227 if isinstance(value, str) 

228 else DataType.OBJECT.value 

229 ) 

230 ) 

231 attr = NamedContextAttribute(name=key, value=value, type=attr_type) 

232 attrs.append(attr) 

233 entity = ContextEntity(self.id, self.type) 

234 entity.add_attributes(attrs) 

235 return entity 

236 

237 

238class ContextEntity(ContextEntityKeyValues): 

239 """ 

240 Context entities, or simply entities, are the center of gravity in the 

241 FIWARE NGSI information model. An entity represents a thing, i.e., any 

242 physical or logical object (e.g., a sensor, a person, a room, an issue in 

243 a ticketing system, etc.). Each entity has an entity id. 

244 Furthermore, the type system of FIWARE NGSI enables entities to have an 

245 entity type. Entity types are semantic types; they are intended to describe 

246 the type of thing represented by the entity. For example, a context 

247 entity #with id sensor-365 could have the type temperatureSensor. 

248 

249 Each entity is uniquely identified by the combination of its id and type. 

250 

251 The entity id is specified by the object's id property, whose value 

252 is a string containing the entity id. 

253 

254 The entity type is specified by the object's type property, whose value 

255 is a string containing the entity's type name. 

256 

257 Entity attributes are specified by additional properties, whose names are 

258 the name of the attribute and whose representation is described by the 

259 "ContextAttribute"-model. Obviously, `id` and `type` are 

260 not allowed as attribute names. 

261 

262 Example:: 

263 

264 >>> data = {'id': 'MyId', 

265 'type': 'MyType', 

266 'my_attr': {'value': 20, 'type': 'Number'}} 

267 

268 >>> entity = ContextEntity(**data) 

269 

270 """ 

271 

272 model_config = ConfigDict( 

273 extra="allow", validate_default=True, validate_assignment=True 

274 ) 

275 

276 # although `type` is a required field in the NGSIv2 specification, it is 

277 # set to optional here to allow for the possibility of setting 

278 # default-types in child classes. Pydantic will raise the correct error 

279 # and also exports the correct json-schema. 

280 def __init__(self, id: str, type: str = None, **data): 

281 # There is currently no validation for extra fields 

282 data.update(self._validate_attributes(data)) 

283 # case where type is None to raise correct error message 

284 if type is None: 

285 super().__init__(id=id, **data) 

286 else: 

287 super().__init__(id=id, type=type, **data) 

288 

289 # Validation of attributes 

290 @classmethod 

291 def _validate_attributes(cls, data: dict): 

292 """ 

293 Validate attributes of the entity if the attribute is not a model 

294 field and the type is not already a subtype of ContextAttribute 

295 """ 

296 attrs = { 

297 key: ContextAttribute.model_validate(attr) 

298 for key, attr in data.items() 

299 if ( 

300 # validate_fiware_attribute_value_regex(key) not in cls.model_fields 

301 validate_fiware_attribute_name_regex(key) not in cls.model_fields 

302 and not isinstance(attr, ContextAttribute) 

303 # key not in cls.model_fields 

304 # and not isinstance(attr, ContextAttribute) 

305 ) 

306 } 

307 

308 return attrs 

309 

310 @field_validator("*") 

311 @classmethod 

312 def check_attributes(cls, value, info: ValidationInfo): 

313 """ 

314 Check whether all model fields are of subtype of ContextAttribute to 

315 ensure full functionality. 

316 """ 

317 if info.field_name in ["id", "type"]: 

318 return value 

319 

320 if info.field_name in cls.model_fields: 

321 if not ( 

322 isinstance(value, ContextAttribute) 

323 or value == cls.model_fields[info.field_name].default 

324 ): 

325 raise ValueError( 

326 f"Attribute {info.field_name} must be a of " 

327 f"type or subtype ContextAttribute" 

328 ) 

329 return value 

330 

331 @model_validator(mode="after") 

332 @classmethod 

333 def check_attributes_after(cls, values): 

334 try: 

335 for attr in values.model_extra: 

336 if not isinstance(values.__getattr__(attr), ContextAttribute): 

337 raise ValueError( 

338 f"Attribute {attr} must be a of type or " 

339 f"subtype ContextAttribute. You most " 

340 f"likely tried to directly assign an " 

341 f"attribute without converting it to a " 

342 f"proper Attribute-Type!" 

343 ) 

344 except TypeError: 

345 pass 

346 return values 

347 

348 # API for attributes and commands 

349 def add_attributes( 

350 self, attrs: Union[Dict[str, ContextAttribute], List[NamedContextAttribute]] 

351 ) -> None: 

352 """ 

353 Add attributes (properties, relationships) to entity 

354 

355 Args: 

356 attrs: Dict[str, ContextAttribute]: {NAME for attr : Attribute} or 

357 List[NamedContextAttribute] 

358 

359 Returns: 

360 None 

361 """ 

362 if isinstance(attrs, list): 

363 attrs = { 

364 attr.name: ContextAttribute(**attr.model_dump(exclude={"name"})) 

365 for attr in attrs 

366 } 

367 for key, attr in attrs.items(): 

368 self.__setattr__(name=key, value=attr) 

369 

370 def get_attributes( 

371 self, 

372 whitelisted_attribute_types: Optional[List[DataType]] = None, 

373 blacklisted_attribute_types: Optional[List[DataType]] = None, 

374 response_format: Union[str, PropertyFormat] = PropertyFormat.LIST, 

375 strict_data_type: bool = True, 

376 ) -> Union[List[NamedContextAttribute], Dict[str, ContextAttribute]]: 

377 """ 

378 Get attributes or a subset from the entity. 

379 

380 Args: 

381 whitelisted_attribute_types: Optional list, if given only 

382 attributes matching one of the types are returned 

383 blacklisted_attribute_types: Optional list, if given all 

384 attributes are returned that do not match a list entry 

385 response_format: Wanted result format, 

386 List -> list of NamedContextAttributes 

387 Dict -> dict of {name: ContextAttribute} 

388 strict_data_type: whether to restrict the data type to pre-defined 

389 types, True by default. 

390 True -> Only return the attributes with pre-defined types, 

391 False -> Do not restrict the data type. 

392 Raises: 

393 AssertionError, if both a white and a black list is given 

394 Returns: 

395 List[NamedContextAttribute] or Dict[str, ContextAttribute] 

396 """ 

397 

398 response_format = PropertyFormat(response_format) 

399 

400 assert ( 

401 whitelisted_attribute_types is None or blacklisted_attribute_types is None 

402 ), "Only whitelist or blacklist is allowed" 

403 

404 if whitelisted_attribute_types is not None: 

405 attribute_types = whitelisted_attribute_types 

406 elif blacklisted_attribute_types is not None: 

407 attribute_types = [ 

408 att_type 

409 for att_type in list(DataType) 

410 if att_type not in blacklisted_attribute_types 

411 ] 

412 else: 

413 attribute_types = [att_type for att_type in list(DataType)] 

414 

415 if response_format == PropertyFormat.DICT: 

416 if strict_data_type: 

417 return { 

418 key: ContextAttribute(**value) 

419 for key, value in self.model_dump().items() 

420 if key not in ContextEntity.model_fields 

421 and value.get("type") in [att.value for att in attribute_types] 

422 } 

423 else: 

424 return { 

425 key: ContextAttribute(**value) 

426 for key, value in self.model_dump().items() 

427 if key not in ContextEntity.model_fields 

428 } 

429 else: 

430 if strict_data_type: 

431 return [ 

432 NamedContextAttribute(name=key, **value) 

433 for key, value in self.model_dump().items() 

434 if key not in ContextEntity.model_fields 

435 and value.get("type") in [att.value for att in attribute_types] 

436 ] 

437 else: 

438 return [ 

439 NamedContextAttribute(name=key, **value) 

440 for key, value in self.model_dump().items() 

441 if key not in ContextEntity.model_fields 

442 ] 

443 

444 def update_attribute( 

445 self, attrs: Union[Dict[str, ContextAttribute], List[NamedContextAttribute]] 

446 ) -> None: 

447 """ 

448 Update attributes of an entity. Overwrite the current held value 

449 for the attribute with the value contained in the corresponding given 

450 attribute 

451 

452 Args: 

453 attrs: List of NamedContextAttributes, 

454 Dict of {attribute_name: ContextAttribute} 

455 Raises: 

456 NameError, if the attribute does not currently exist in the entity 

457 Returns: 

458 None 

459 """ 

460 if isinstance(attrs, list): 

461 attrs = { 

462 attr.name: ContextAttribute(**attr.model_dump(exclude={"name"})) 

463 for attr in attrs 

464 } 

465 

466 existing_attribute_names = self.get_attribute_names() 

467 for key, attr in attrs.items(): 

468 if key not in existing_attribute_names: 

469 raise NameError 

470 self.__setattr__(name=key, value=attr) 

471 

472 def get_attribute_names(self) -> Set[str]: 

473 """ 

474 Returns a set with all attribute names of this entity 

475 

476 Returns: 

477 Set[str] 

478 """ 

479 

480 return { 

481 key for key in self.model_dump() if key not in ContextEntity.model_fields 

482 } 

483 

484 def delete_attributes( 

485 self, 

486 attrs: Union[ 

487 Dict[str, ContextAttribute], List[NamedContextAttribute], List[str] 

488 ], 

489 ): 

490 """ 

491 Delete the given attributes from the entity 

492 

493 Args: 

494 attrs: - Dict {name: ContextAttribute} 

495 - List[NamedContextAttribute] 

496 - List[str] -> names of attributes 

497 Raises: 

498 Exception: if one of the given attrs does not represent an 

499 existing argument 

500 """ 

501 

502 names: List[str] = [] 

503 if isinstance(attrs, list): 

504 for entry in attrs: 

505 if isinstance(entry, str): 

506 names.append(entry) 

507 elif isinstance(entry, NamedContextAttribute): 

508 names.append(entry.name) 

509 else: 

510 names.extend(list(attrs.keys())) 

511 for name in names: 

512 delattr(self, name) 

513 

514 def get_attribute(self, attribute_name: str) -> NamedContextAttribute: 

515 """ 

516 Get the attribute of the entity with the given name 

517 

518 Args: 

519 attribute_name (str): Name of attribute 

520 

521 Raises: 

522 KeyError, if no attribute with given name exists 

523 

524 Returns: 

525 NamedContextAttribute 

526 """ 

527 for attr in self.get_attributes(): 

528 if attr.name == attribute_name: 

529 return attr 

530 raise KeyError(f"Attribute '{attribute_name}' not in entity") 

531 

532 def get_properties( 

533 self, response_format: Union[str, PropertyFormat] = PropertyFormat.LIST 

534 ) -> Union[List[NamedContextAttribute], Dict[str, ContextAttribute]]: 

535 """ 

536 Returns all attributes of the entity that are not of type Relationship, 

537 and are not auto generated command attributes 

538 

539 Args: 

540 response_format: Wanted result format, 

541 List -> list of NamedContextAttributes 

542 Dict -> dict of {name: ContextAttribute} 

543 

544 Returns: 

545 [NamedContextAttribute] or {name: ContextAttribute} 

546 """ 

547 pre_filtered_attrs = self.get_attributes( 

548 blacklisted_attribute_types=[DataType.RELATIONSHIP], 

549 response_format=PropertyFormat.LIST, 

550 ) 

551 

552 all_command_attributes_names = set() 

553 for command in self.get_commands(): 

554 (c, c_status, c_info) = self.get_command_triple(command.name) 

555 all_command_attributes_names.update([c.name, c_status.name, c_info.name]) 

556 

557 property_attributes = [] 

558 for attr in pre_filtered_attrs: 

559 if attr.name not in all_command_attributes_names: 

560 property_attributes.append(attr) 

561 

562 if response_format == PropertyFormat.LIST: 

563 return property_attributes 

564 else: 

565 return { 

566 p.name: ContextAttribute(**p.model_dump(exclude={"name"})) 

567 for p in property_attributes 

568 } 

569 

570 def get_relationships( 

571 self, response_format: Union[str, PropertyFormat] = PropertyFormat.LIST 

572 ) -> Union[List[NamedContextAttribute], Dict[str, ContextAttribute]]: 

573 """ 

574 Get all relationships of the context entity 

575 

576 Args: 

577 response_format: Wanted result format, 

578 List -> list of NamedContextAttributes 

579 Dict -> dict of {name: ContextAttribute} 

580 

581 Returns: 

582 [NamedContextAttribute] or {name: ContextAttribute} 

583 

584 """ 

585 return self.get_attributes( 

586 whitelisted_attribute_types=[DataType.RELATIONSHIP], 

587 response_format=response_format, 

588 ) 

589 

590 def get_commands( 

591 self, response_format: Union[str, PropertyFormat] = PropertyFormat.LIST 

592 ) -> Union[List[NamedContextAttribute], Dict[str, ContextAttribute]]: 

593 """ 

594 Get all commands of the context entity. Only works if the commands 

595 were autogenerated by Fiware from an Device. 

596 

597 Args: 

598 response_format: Wanted result format, 

599 List -> list of NamedContextAttributes 

600 Dict -> dict of {name: ContextAttribute} 

601 

602 Returns: 

603 [NamedContextAttribute] or {name: ContextAttribute} 

604 """ 

605 

606 # if an attribute with name n is a command, its type does not need to 

607 # be COMMAND. 

608 # But the attributes name_info (type: commandResult) and 

609 # name_status(type: commandStatus) need to exist. (Autogenerated) 

610 

611 # Search all attributes of type commandStatus, check for each if a 

612 # corresponding _info exists and if also a fitting attribute exists 

613 # we know: that is a command. 

614 

615 commands = [] 

616 for status_attribute in self.get_attributes( 

617 whitelisted_attribute_types=[DataType.COMMAND_STATUS] 

618 ): 

619 if not status_attribute.name.split("_")[-1] == "status": 

620 continue 

621 base_name = status_attribute.name[:-7] 

622 

623 try: 

624 info_attribute = self.get_attribute(f"{base_name}_info") 

625 if not info_attribute.type == DataType.COMMAND_RESULT: 

626 continue 

627 

628 attribute = self.get_attribute(base_name) 

629 commands.append(attribute) 

630 except KeyError: 

631 continue 

632 

633 if response_format == PropertyFormat.LIST: 

634 return commands 

635 else: 

636 return { 

637 cmd.name: ContextAttribute(**cmd.model_dump(exclude={"name"})) 

638 for cmd in commands 

639 } 

640 

641 def get_command_triple( 

642 self, command_attribute_name: str 

643 ) -> Tuple[NamedContextAttribute, NamedContextAttribute, NamedContextAttribute]: 

644 """ 

645 Returns for a given command attribute name all three corresponding 

646 attributes as triple 

647 

648 Args: 

649 command_attribute_name: Name of the command attribute 

650 

651 Raises: 

652 KeyError, if the given name does not belong to a command attribute 

653 

654 Returns: 

655 (Command, Command_status, Command_info) 

656 """ 

657 

658 commands = self.get_commands(response_format=PropertyFormat.DICT) 

659 

660 if command_attribute_name not in commands: 

661 raise KeyError(f"Command '{command_attribute_name}' not in commands") 

662 

663 command = self.get_attribute(command_attribute_name) 

664 

665 # as the given name was found as a valid command, we know that the 

666 # status and info attributes exist correctly 

667 command_status = self.get_attribute(f"{command_attribute_name}_status") 

668 command_info = self.get_attribute(f"{command_attribute_name}_info") 

669 

670 return command, command_status, command_info 

671 

672 def to_keyvalues(self): 

673 attrs = { 

674 attr: value.value 

675 for attr, value in self.get_attributes( 

676 response_format=PropertyFormat.DICT 

677 ).items() 

678 } 

679 entity = ContextEntityKeyValues(self.id, self.type, **attrs) 

680 return entity 

681 

682 def to_normalized(self): 

683 raise AttributeError("This method is not available in ContextEntity") 

684 

685 

686class ContextEntityList(BaseModel): 

687 """ 

688 Collection model for a list of context entities 

689 """ 

690 

691 entities: List[OnErrorOmit[ContextEntity]] 

692 

693 

694class ContextEntityKeyValuesList(BaseModel): 

695 """ 

696 Collection model for a list of context entities in key-values format 

697 """ 

698 

699 entities: List[OnErrorOmit[ContextEntityKeyValues]] 

700 

701 

702class ContextEntityValidationList(ContextEntityList): 

703 """ 

704 Collection model for a list of valid and invalid context entities 

705 """ 

706 

707 invalid_entities: List[str] 

708 

709 

710class ContextEntityKeyValuesValidationList(ContextEntityKeyValuesList): 

711 """ 

712 Collection model for a list of valid and invalid context entities in key-values format 

713 """ 

714 

715 invalid_entities: List[str] 

716 

717 

718class Query(BaseModel): 

719 """ 

720 Model for queries 

721 """ 

722 

723 entities: List[EntityPattern] = Field( 

724 description="a list of entities to search for. Each element is " 

725 "represented by a JSON object" 

726 ) 

727 attrs: Optional[List[str]] = Field( 

728 default=None, 

729 description="List of attributes to be provided " 

730 "(if not specified, all attributes).", 

731 ) 

732 expression: Optional[Expression] = Field( 

733 default=None, 

734 description="An expression composed of q, mq, georel, geometry and " "coords", 

735 ) 

736 metadata: Optional[List[str]] = Field( 

737 default=None, 

738 description="a list of metadata names to include in the response. " 

739 'See "Filtering out attributes and metadata" section for ' 

740 "more detail.", 

741 ) 

742 

743 

744class ActionType(str, Enum): 

745 """ 

746 Options for queries 

747 """ 

748 

749 _init_ = "value __doc__" 

750 APPEND = ( 

751 "append", 

752 "maps to POST /v2/entities (if the entity does not " 

753 "already exist) or POST /v2/entities/<id>/attrs (if " 

754 "the entity already exists). ", 

755 ) 

756 APPEND_STRICT = ( 

757 "appendStrict", 

758 "maps to POST /v2/entities (if the " 

759 "entity does not already exist) or POST " 

760 "/v2/entities/<id>/attrs?options=append " 

761 "(if the entity already exists).", 

762 ) 

763 UPDATE = "update", "maps to PATCH /v2/entities/<id>/attrs." 

764 DELETE = ( 

765 "delete", 

766 "maps to DELETE /v2/entities/<id>/attrs/<attrName> on " 

767 "every attribute included in the entity or to DELETE " 

768 "/v2/entities/<id> if no attribute were included in " 

769 "the entity.", 

770 ) 

771 REPLACE = "replace", "maps to PUT /v2/entities/<id>/attrs" 

772 

773 

774class Update(BaseModel): 

775 """ 

776 Model for update action 

777 """ 

778 

779 action_type: Union[ActionType, str] = Field( 

780 alias="actionType", 

781 description="actionType, to specify the kind of update action to do: " 

782 "either append, appendStrict, update, delete, or replace. ", 

783 ) 

784 entities: SerializeAsAny[List[Union[ContextEntity, ContextEntityKeyValues]]] = ( 

785 Field( 

786 description="an array of entities, each entity specified using the " 

787 "JSON entity representation format " 

788 ) 

789 ) 

790 

791 @field_validator("action_type") 

792 @classmethod 

793 def check_action_type(cls, action): 

794 """ 

795 validates action_type 

796 Args: 

797 action: field action_type 

798 Returns: 

799 action_type 

800 """ 

801 return ActionType(action) 

802 

803 

804class Command(BaseModel): 

805 """ 

806 Class for sending commands to IoT Devices. 

807 Note that the command must be registered via an IoT-Agent. Internally 

808 FIWARE uses its registration mechanism in order to connect the command 

809 with an IoT-Device 

810 """ 

811 

812 type: DataType = Field( 

813 default=DataType.COMMAND, 

814 description="Command must have the type command", 

815 # const=True 

816 ) 

817 value: Any = Field( 

818 description="Any json serializable command that will " 

819 "be forwarded to the connected IoT device" 

820 ) 

821 

822 @field_validator("value") 

823 @classmethod 

824 def check_value(cls, value): 

825 """ 

826 Check if value is json serializable 

827 Args: 

828 value: value field 

829 Returns: 

830 value 

831 """ 

832 try: 

833 json.dumps(value) 

834 except: 

835 raise ValueError(f"Command value {value} " f"is not serializable") 

836 return value 

837 

838 

839class NamedCommand(Command): 

840 """ 

841 Class for sending command to IoT-Device. 

842 Extend :class: Command with command Name 

843 """ 

844 

845 name: str = Field( 

846 description="Name of the command", 

847 max_length=256, 

848 min_length=1, 

849 ) 

850 valid_name = field_validator("name")(validate_fiware_attribute_name_regex)