Coverage for filip/models/ngsi_v2/subscriptions.py: 96%
141 statements
« prev ^ index » next coverage.py v7.4.4, created at 2025-02-19 11:48 +0000
1"""
2This module contains NGSIv2 models for context subscription in the context
3broker.
4"""
6from typing import Any, List, Dict, Union, Optional
7from datetime import datetime
8from aenum import Enum
9from pydantic import (
10 field_validator,
11 model_validator,
12 ConfigDict,
13 BaseModel,
14 conint,
15 Field,
16 Json,
17)
18from .base import AttrsFormat, EntityPattern, Http, Status, Expression
19from filip.utils.validators import validate_mqtt_url, validate_mqtt_topic
20from filip.models.ngsi_v2.context import ContextEntity
21from filip.models.ngsi_v2.base import EntityPattern, Expression, DataType
22from filip.custom_types import AnyMqttUrl
23import warnings
25# The pydantic models still have a .json() function, but this method is deprecated.
26warnings.filterwarnings(
27 "ignore",
28 category=UserWarning,
29 message='Field name "json" shadows an attribute in parent "Http"',
30)
31warnings.filterwarnings(
32 "ignore",
33 category=UserWarning,
34 message='Field name "json" shadows an attribute in parent "Mqtt"',
35)
38class NgsiPayloadAttr(BaseModel):
39 """
40 Model for NGSI V2 type payload in httpCustom/mqttCustom notifications.
41 The difference between this model and the usual BaseValueAttribute model is that
42 a metadata field is not allowed.
43 In the absence of type/value in some attribute field, one should resort to partial
44 representations ( as specified in the orion api manual), done by the BaseValueAttr.
45 model.
46 """
48 model_config = ConfigDict(extra="forbid")
49 type: Union[DataType, str] = Field(
50 default=DataType.TEXT,
51 description="The attribute type represents the NGSI value type of the "
52 "attribute value. Note that FIWARE NGSI has its own type "
53 "system for attribute values, so NGSI value types are not "
54 "the same as JSON types. Allowed characters "
55 "are the ones in the plain ASCII set, except the following "
56 "ones: control characters, whitespace, &, ?, / and #.",
57 max_length=256,
58 min_length=1,
59 )
60 value: Optional[Any] = Field(
61 default=None, title="Attribute value", description="the actual data"
62 )
65class NgsiPayload(BaseModel):
66 """
67 Model for NGSI V2 type payload in httpCustom/mqttCustom notifications.
68 Differences between this model and the usual Context entity models include:
69 - id and type are not mandatory
70 - an attribute metadata field is not allowed
71 """
73 model_config = ConfigDict(extra="allow", validate_default=True)
74 id: Optional[str] = Field(default=None, max_length=256, min_length=1, frozen=True)
75 type: Optional[Union[str, Enum]] = Field(
76 default=None,
77 max_length=256,
78 min_length=1,
79 frozen=True,
80 )
82 @model_validator(mode="after")
83 def validate_notification_attrs(self):
84 for v in self.model_dump(exclude={"id", "type"}).values():
85 assert isinstance(NgsiPayloadAttr.model_validate(v), NgsiPayloadAttr)
86 return self
89class Message(BaseModel):
90 """
91 Model for a notification message, when sent to other NGSIv2-APIs
92 """
94 subscriptionId: Optional[str] = Field(
95 default=None,
96 description="Id of the subscription the notification comes from",
97 )
98 data: List[ContextEntity] = Field(
99 description="is an array with the notification data itself which "
100 "includes the entity and all concerned attributes. Each "
101 "element in the array corresponds to a different entity. "
102 "By default, the entities are represented in normalized "
103 "mode. However, using the attrsFormat modifier, a "
104 "simplified representation mode can be requested."
105 )
108class HttpMethods(str, Enum):
109 _init_ = "value __doc__"
111 POST = "POST", "Post Method"
112 PUT = "PUT", "Put Method"
113 PATCH = "PATCH", "Patch Method"
116class HttpCustom(Http):
117 """
118 Model for custom notification patterns sent via HTTP
119 """
121 headers: Optional[Dict[str, Union[str, Json]]] = Field(
122 default=None,
123 description="a key-map of HTTP headers that are included in "
124 "notification messages.",
125 )
126 qs: Optional[Dict[str, Union[str, Json]]] = Field(
127 default=None,
128 description="a key-map of URL query parameters that are included in "
129 "notification messages.",
130 )
131 method: str = Field(
132 default=HttpMethods.POST,
133 description="the method to use when sending the notification "
134 "(default is POST). Only valid HTTP methods are allowed. "
135 "On specifying an invalid HTTP method, a 400 Bad Request "
136 "error is returned.",
137 )
138 payload: Optional[str] = Field(
139 default=None,
140 description="the payload to be used in notifications. If omitted, the "
141 'default payload (see "Notification Messages" sections) '
142 "is used.",
143 )
144 json: Optional[Dict[str, Union[str, Json]]] = Field(
145 default=None,
146 description="get a json as notification. If omitted, the default"
147 'payload (see "Notification Messages" sections) is used.',
148 )
149 ngsi: Optional[NgsiPayload] = Field(
150 default=None,
151 description="get an NGSI-v2 normalized entity as notification.If omitted, "
152 'the default payload (see "Notification Messages" sections) is used.',
153 )
154 timeout: Optional[int] = Field(
155 default=None,
156 description="Maximum time (in milliseconds) the subscription waits for the "
157 "response. The maximum value allowed for this parameter is 1800000 "
158 "(30 minutes). If timeout is defined to 0 or omitted, then the value "
159 "passed as -httpTimeout CLI parameter is used. See section in the "
160 "'Command line options' for more details.",
161 )
163 @model_validator(mode="after")
164 def validate_notification_payloads(self):
165 fields = [self.payload, self.json, self.ngsi]
166 filled_fields = [field for field in fields if field is not None]
168 if len(filled_fields) > 1:
169 raise ValueError(
170 "Only one of payload, json or ngsi fields accepted at the "
171 "same time in httpCustom."
172 )
174 return self
177class Mqtt(BaseModel):
178 """
179 Model for notifications sent via MQTT
180 https://fiware-orion.readthedocs.io/en/3.8.0/user/mqtt_notifications/index.html
181 """
183 url: Union[AnyMqttUrl, str] = Field(
184 description="to specify the MQTT broker endpoint to use. URL must "
185 "start with mqtt:// and never contains a path (i.e. it "
186 "only includes host and port)"
187 )
188 topic: str = Field(
189 description="to specify the MQTT topic to use",
190 )
191 valid_type = field_validator("topic")(validate_mqtt_topic)
192 qos: Optional[int] = Field(
193 default=0,
194 description="to specify the MQTT QoS value to use in the "
195 "notifications associated to the subscription (0, 1 or 2). "
196 "This is an optional field, if omitted then QoS 0 is used.",
197 ge=0,
198 le=2,
199 )
200 user: Optional[str] = Field(default=None, description="username if required")
201 passwd: Optional[str] = Field(default=None, description="password if required")
203 @field_validator("url")
204 @classmethod
205 def check_url(cls, value):
206 """
207 Check if url has a valid structure
208 Args:
209 value: url to validate
210 Returns:
211 validated url
212 """
213 return validate_mqtt_url(url=value)
216class MqttCustom(Mqtt):
217 """
218 Model for custom notification patterns sent via MQTT
219 https://fiware-orion.readthedocs.io/en/3.8.0/user/mqtt_notifications/index.html
220 """
222 payload: Optional[str] = Field(
223 default=None,
224 description="the payload to be used in notifications. If omitted, the "
225 'default payload (see "Notification Messages" sections) '
226 "is used.",
227 )
228 json: Optional[Dict[str, Any]] = Field(
229 default=None,
230 description="get a json as notification. If omitted, the default"
231 'payload (see "Notification Messages" sections) is used.',
232 )
233 ngsi: Optional[NgsiPayload] = Field(
234 default=None,
235 description="get an NGSI-v2 normalized entity as notification.If omitted, "
236 'the default payload (see "Notification Messages" sections) is used.',
237 )
239 @model_validator(mode="after")
240 def validate_payload_type(self):
241 assert (
242 len(
243 [
244 v
245 for k, v in self.model_dump().items()
246 if ((v is not None) and (k in ["payload", "ngsi", "json"]))
247 ]
248 )
249 <= 1
250 )
251 return self
254class Notification(BaseModel):
255 """
256 If the notification attributes are left empty, all attributes will be
257 included in the notifications. Otherwise, only the specified ones will
258 be included.
259 """
261 model_config = ConfigDict(validate_assignment=True)
262 timesSent: Optional[Any] = Field(
263 default=None,
264 description="Not editable, only present in GET operations. "
265 "Number of notifications sent due to this subscription.",
266 )
267 http: Optional[Http] = Field(
268 default=None,
269 description="It is used to convey parameters for notifications "
270 "delivered through the HTTP protocol. Cannot be used "
271 'together with "httpCustom, mqtt, mqttCustom"',
272 )
273 httpCustom: Optional[HttpCustom] = Field(
274 default=None,
275 description="It is used to convey parameters for notifications "
276 "delivered through the HTTP protocol. Cannot be used "
277 'together with "http"',
278 )
279 mqtt: Optional[Mqtt] = Field(
280 default=None,
281 description="It is used to convey parameters for notifications "
282 "delivered through the MQTT protocol. Cannot be used "
283 'together with "http, httpCustom, mqttCustom"',
284 )
285 mqttCustom: Optional[MqttCustom] = Field(
286 default=None,
287 description="It is used to convey parameters for notifications "
288 "delivered through the MQTT protocol. Cannot be used "
289 'together with "http, httpCustom, mqtt"',
290 )
291 attrs: Optional[List[str]] = Field(
292 default=None,
293 description="List of attributes to be included in notification "
294 "messages. It also defines the order in which attributes "
295 "must appear in notifications when attrsFormat value is "
296 'used (see "Notification Messages" section). An empty list '
297 "means that all attributes are to be included in "
298 'notifications. See "Filtering out attributes and '
299 'metadata" section for more detail.',
300 )
301 exceptAttrs: Optional[List[str]] = Field(
302 default=None,
303 description="List of attributes to be excluded from the notification "
304 "message, i.e. a notification message includes all entity "
305 "attributes except the ones listed in this field.",
306 )
307 attrsFormat: Optional[AttrsFormat] = Field(
308 default=AttrsFormat.NORMALIZED,
309 description="specifies how the entities are represented in "
310 "notifications. Accepted values are normalized (default), "
311 "keyValues or values. If attrsFormat takes any value "
312 "different than those, an error is raised. See detail in "
313 '"Notification Messages" section.',
314 )
315 metadata: Optional[Any] = Field(
316 default=None,
317 description="List of metadata to be included in notification messages. "
318 'See "Filtering out attributes and metadata" section for '
319 "more detail.",
320 )
321 onlyChangedAttrs: Optional[bool] = Field(
322 default=False,
323 description="Only supported by Orion Context Broker!"
324 "If set to true then notifications associated to the "
325 "subscription include only attributes that changed in the "
326 "triggering update request, in combination with the attrs "
327 "or exceptAttrs field. For instance, if attrs is "
328 "[A=1, B=2, C=3] and A=0 is updated. In case "
329 "onlyChangedAttrs=false, CB notifies [A=0, B=2, C=3]."
330 "In case onlyChangedAttrs=true, CB notifies "
331 "[A=0, B=null, C=null]. This ",
332 )
333 covered: Optional[bool] = Field(
334 default=False,
335 description="A flag to decide whether to include not existing attribute in "
336 "notifications. It can be useful for those notification endpoints "
337 "that are not flexible enough for a variable set of attributes and "
338 "needs always the same set of incoming attributes in every received"
339 " notification "
340 "https://fiware-orion.readthedocs.io/en/master/orion-api.html#covered-subscriptions",
341 )
343 @model_validator(mode="after")
344 def validate_http(self):
345 if self.httpCustom is not None:
346 assert self.http is None
347 return self
349 @model_validator(mode="after")
350 def validate_attr(self):
351 if self.exceptAttrs is not None:
352 assert self.attrs is None
353 return self
355 @model_validator(mode="after")
356 def validate_endpoints(self):
357 if self.http is not None:
358 assert all(
359 (
360 v is None
361 for k, v in self.model_dump().items()
362 if k in ["httpCustom", "mqtt", "mqttCustom"]
363 )
364 )
365 elif self.httpCustom is not None:
366 assert all(
367 (
368 v is None
369 for k, v in self.model_dump().items()
370 if k in ["http", "mqtt", "mqttCustom"]
371 )
372 )
373 elif self.mqtt is not None:
374 assert all(
375 (
376 v is None
377 for k, v in self.model_dump().items()
378 if k in ["http", "httpCustom", "mqttCustom"]
379 )
380 )
381 else:
382 assert all(
383 (
384 v is None
385 for k, v in self.model_dump().items()
386 if k in ["http", "httpCustom", "mqtt"]
387 )
388 )
389 return self
391 @model_validator(mode="after")
392 def validate_covered_attrs(self):
393 if self.covered is True:
394 if isinstance(self.attrs, list) and len(self.attrs) > 0:
395 return self
396 else:
397 raise ValueError("Covered notification need an explicit list of attrs.")
398 return self
401class Response(Notification):
402 """
403 Server response model for notifications
404 """
406 timesSent: int = Field(
407 description="(not editable, only present in GET operations): "
408 "Number of notifications sent due to this subscription."
409 )
410 lastNotification: datetime = Field(
411 description="(not editable, only present in GET operations): "
412 "Last notification timestamp in ISO8601 format."
413 )
414 lastFailure: Optional[datetime] = Field(
415 default=None,
416 description="(not editable, only present in GET operations): "
417 "Last failure timestamp in ISO8601 format. Not present if "
418 "subscription has never had a problem with notifications.",
419 )
420 lastSuccess: Optional[datetime] = Field(
421 default=None,
422 description="(not editable, only present in GET operations): "
423 "Timestamp in ISO8601 format for last successful "
424 "notification. Not present if subscription has never "
425 "had a successful notification.",
426 )
429class Condition(BaseModel):
430 """
431 Notification rules are as follow:
432 If attrs and expression are used, a notification is sent whenever one of
433 the attributes in the attrs list changes and at the same time expression
434 matches.
435 If attrs is used and expression is not used, a notification is sent
436 whenever any of the attributes in the attrs list changes.
437 If attrs is not used and expression is used, a notification is sent
438 whenever any of the attributes of the entity changes and at the same time
439 expression matches.
440 If neither attrs nor expression are used, a notification is sent whenever
441 any of the attributes of the entity changes.
442 alterationTypes: for more information about this field, see
443 https://github.com/telefonicaid/fiware-orion/blob/3.8.0/doc/manuals/orion-api.md#subscriptions-based-in-alteration-type
445 """
447 attrs: Optional[Union[str, List[str]]] = Field(
448 default=None, description="array of attribute names"
449 )
450 expression: Optional[Union[str, Expression]] = Field(
451 default=None,
452 description="an expression composed of q, mq, georel, geometry and "
453 'coords (see "List entities" operation above about this '
454 "field).",
455 )
456 alterationTypes: Optional[List[str]] = Field(
457 default=None, description="list of alteration types triggering the subscription"
458 )
460 @field_validator("attrs")
461 def check_attrs(cls, v):
462 if isinstance(v, list):
463 return v
464 elif isinstance(v, str):
465 return [v]
466 else:
467 raise TypeError()
469 @field_validator("alterationTypes")
470 def check_alteration_types(cls, v):
471 allowed_types = {"entityCreate", "entityDelete", "entityUpdate", "entityChange"}
473 if v is None:
474 return None
475 elif isinstance(v, list):
476 for item in v:
477 if item not in allowed_types:
478 raise ValueError(
479 f"{item} is not a valid alterationType"
480 f" allowed values are {allowed_types}"
481 )
482 return v
483 else:
484 raise ValueError("alterationTypes must be a list of strings")
487class Subject(BaseModel):
488 """
489 Model for subscription subject
490 """
492 entities: List[EntityPattern] = Field(
493 description="A list of objects, each one composed of by an Entity " "Object:"
494 )
495 condition: Optional[Condition] = Field(
496 default=None,
497 )
500class Subscription(BaseModel):
501 """
502 Subscription payload validations
503 https://fiware-orion.readthedocs.io/en/master/user/ngsiv2_implementation_notes/index.html#subscription-payload-validations
504 """
506 model_config = ConfigDict(validate_assignment=True)
508 id: Optional[str] = Field(
509 default=None,
510 description="Subscription unique identifier. Automatically created at "
511 "creation time.",
512 )
513 description: Optional[str] = Field(
514 default=None,
515 description="A free text used by the client to describe the " "subscription.",
516 )
517 status: Optional[Status] = Field(
518 default=Status.ACTIVE,
519 description="Either active (for active subscriptions) or inactive "
520 "(for inactive subscriptions). If this field is not "
521 "provided at subscription creation time, new subscriptions "
522 "are created with the active status, which can be changed"
523 " by clients afterwards. For expired subscriptions, this "
524 "attribute is set to expired (no matter if the client "
525 "updates it to active/inactive). Also, for subscriptions "
526 "experiencing problems with notifications, the status is "
527 "set to failed. As soon as the notifications start working "
528 "again, the status is changed back to active.",
529 )
530 subject: Subject = Field(
531 description="An object that describes the subject of the subscription.",
532 json_schema_extra={
533 "example": {
534 "entities": [{"idPattern": ".*", "type": "Room"}],
535 "condition": {
536 "attrs": ["temperature"],
537 "expression": {"q": "temperature>40"},
538 },
539 }
540 },
541 )
542 notification: Notification = Field(
543 description="An object that describes the notification to send when "
544 "the subscription is triggered.",
545 json_schema_extra={
546 "example": {
547 "http": {"url": "http://localhost:1234"},
548 "attrs": ["temperature", "humidity"],
549 }
550 },
551 )
552 expires: Optional[datetime] = Field(
553 default=None,
554 description="Subscription expiration date in ISO8601 format. "
555 "Permanent subscriptions must omit this field.",
556 )
558 throttling: Optional[
559 conint(
560 strict=True,
561 ge=0,
562 )
563 ] = Field(
564 default=None,
565 strict=True,
566 description="Minimal period of time in seconds which "
567 "must elapse between two consecutive notifications. "
568 "It is optional.",
569 )