Coverage for filip/models/ngsi_v2/registrations.py: 100%

1""" 

2This module contains NGSIv2 models for context registrations in the context 

3broker. 

4""" 

5from typing import List, Union, Optional 

6from datetime import datetime 

7from aenum import Enum 

8from pydantic import ConfigDict, BaseModel, Field 

9from filip.models.ngsi_v2.base import EntityPattern, Expression, Http, Status 

10 

11 

12class ForwardingMode(str, Enum): 

13 _init_ = 'value __doc__' 

14 

15 NONE = "none", "This provider does not support request forwarding." 

16 QUERY = "query", "This provider only supports request forwarding to " \ 

17 "query data." 

18 UPDATE = "update", "This provider only supports request forwarding to " \ 

19 "update data." 

20 ALL = "all", "This provider supports both query and update forwarding " \ 

21 "requests. (Default value)" 

22 

23 

24class Provider(BaseModel): 

25 http: Http = Field( 

26 description="It is used to convey parameters for providers that " 

27 "deliver information through the HTTP protocol. (Only " 

28 "protocol supported nowadays). It must contain a subfield " 

29 "named url with the URL that serves as the endpoint that " 

30 "offers the providing interface. The endpoint must not " 

31 "include the protocol specific part (for instance " 

32 "/v2/entities). " 

33 ) 

34 supportedForwardingMode: ForwardingMode = Field( 

35 default=ForwardingMode.ALL, 

36 description="It is used to convey the forwarding mode supported by " 

37 "this context provider. By default all." 

38 ) 

39 

40 

41class ForwardingInformation(BaseModel): 

42 model_config = ConfigDict(frozen=True) 

43 

44 timesSent: int = Field( 

45 description="(not editable, only present in GET operations): " 

46 "Number of forwarding requests sent due to this " 

47 "registration." 

48 ) 

49 lastForwarding: datetime = Field( 

50 description="(not editable, only present in GET operations): " 

51 "Last forwarding timestamp in ISO8601 format." 

52 ) 

53 lastFailure: Optional[datetime] = Field( 

54 default=None, 

55 description="(not editable, only present in GET operations): " 

56 "Last failure timestamp in ISO8601 format. Not present " 

57 "if registration has never had a problem with forwarding." 

58 ) 

59 lastSuccess: Optional[datetime] = Field( 

60 default=None, 

61 description="(not editable, only present in GET operations): " 

62 "Timestamp in ISO8601 format for last successful " 

63 "request forwarding. Not present if registration has " 

64 "never had a successful notification." 

65 ) 

66 

67 

68class DataProvided(BaseModel): 

69 """ 

70 Model for provided data 

71 """ 

72 entities: List[EntityPattern] = Field( 

73 description="A list of objects, each one composed by an entity object" 

74 ) 

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

76 default=None, 

77 description="List of attributes to be provided " 

78 "(if not specified, all attributes)" 

79 ) 

80 expression: Optional[Union[str, Expression]] = Field( 

81 default=None, 

82 description="By means of a filtering expression, allows to express " 

83 "what is the scope of the data provided. Currently only " 

84 "geographical scopes are supported " 

85 ) 

86 

87 

88class Registration(BaseModel): 

89 """ 

90 A Context Registration allows to bind external context information 

91 sources so that they can play the role of providers of certain subsets 

92 (entities, attributes) of the context information space, including those 

93 located at specific geographical areas. 

94 """ 

95 id: Optional[str] = Field( 

96 default=None, 

97 description="Unique identifier assigned to the registration. " 

98 "Automatically generated at creation time." 

99 ) 

100 description: Optional[str] = Field( 

101 default=None, 

102 description="A free text used by the client to describe the " 

103 "registration.", 

104 example="Relative Humidity Context Source" 

105 ) 

106 provider: Provider = Field( 

107 description="Object that describes the context source registered.", 

108 example='"http": {"url": "http://localhost:1234"}' 

109 ) 

110 dataProvided: DataProvided = Field( 

111 description="Object that describes the data provided by this source", 

112 example='{' 

113 ' "entities": [{"id": "room2", "type": "Room"}],' 

114 ' "attrs": ["relativeHumidity"]' 

115 '},' 

116 ) 

117 status: Optional[Status] = Field( 

118 default=Status.ACTIVE, 

119 description="Either active (for active registration) or inactive " 

120 "(for inactive registration). If this field is not " 

121 "provided at registration creation time, new registration " 

122 "are created with the active status, which can be changed" 

123 " by clients afterwards. For expired registration, this " 

124 "attribute is set to expired (no matter if the client " 

125 "updates it to active/inactive). Also, for subscriptions " 

126 "experiencing problems with notifications, the status is " 

127 "set to failed. As soon as the notifications start working " 

128 "again, the status is changed back to active." 

129 ) 

130 expires: Optional[datetime] = Field( 

131 default=None, 

132 description="Registration expiration date in ISO8601 format. " 

133 "Permanent registrations must omit this field." 

134 ) 

135 forwardingInformation: Optional[ForwardingInformation] = Field( 

136 default=None, 

137 description="Information related to the forwarding operations made " 

138 "against the provider. Automatically provided by the " 

139 "implementation, in the case such implementation supports " 

140 "forwarding capabilities." 

141 )