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

1""" 

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

3broker. 

4""" 

5 

6from typing import List, Union, Optional 

7from datetime import datetime 

8from aenum import Enum 

9from pydantic import ConfigDict, BaseModel, Field 

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

11 

12 

13class ForwardingMode(str, Enum): 

14 _init_ = "value __doc__" 

15 

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

17 QUERY = "query", "This provider only supports request forwarding to " "query data." 

18 UPDATE = ( 

19 "update", 

20 "This provider only supports request forwarding to " "update data.", 

21 ) 

22 ALL = ( 

23 "all", 

24 "This provider supports both query and update forwarding " 

25 "requests. (Default value)", 

26 ) 

27 

28 

29class Provider(BaseModel): 

30 http: Http = Field( 

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

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

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

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

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

36 "include the protocol specific part (for instance " 

37 "/v2/entities). " 

38 ) 

39 supportedForwardingMode: ForwardingMode = Field( 

40 default=ForwardingMode.ALL, 

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

42 "this context provider. By default all.", 

43 ) 

44 

45 

46class ForwardingInformation(BaseModel): 

47 model_config = ConfigDict(frozen=True) 

48 

49 timesSent: int = Field( 

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

51 "Number of forwarding requests sent due to this " 

52 "registration." 

53 ) 

54 lastForwarding: datetime = Field( 

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

56 "Last forwarding timestamp in ISO8601 format." 

57 ) 

58 lastFailure: Optional[datetime] = Field( 

59 default=None, 

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

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

62 "if registration has never had a problem with forwarding.", 

63 ) 

64 lastSuccess: Optional[datetime] = Field( 

65 default=None, 

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

67 "Timestamp in ISO8601 format for last successful " 

68 "request forwarding. Not present if registration has " 

69 "never had a successful notification.", 

70 ) 

71 

72 

73class DataProvided(BaseModel): 

74 """ 

75 Model for provided data 

76 """ 

77 

78 entities: List[EntityPattern] = Field( 

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

80 ) 

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

82 default=None, 

83 description="List of attributes to be provided " 

84 "(if not specified, all attributes)", 

85 ) 

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

87 default=None, 

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

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

90 "geographical scopes are supported ", 

91 ) 

92 

93 

94class Registration(BaseModel): 

95 """ 

96 A Context Registration allows to bind external context information 

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

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

99 located at specific geographical areas. 

100 """ 

101 

102 id: Optional[str] = Field( 

103 default=None, 

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

105 "Automatically generated at creation time.", 

106 ) 

107 description: Optional[str] = Field( 

108 default=None, 

109 description="A free text used by the client to describe the " "registration.", 

110 json_schema_extra={"example": "Relative Humidity Context Source"}, 

111 ) 

112 provider: Provider = Field( 

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

114 json_schema_extra={"example": '"http": {"url": "http://localhost:1234"}'}, 

115 ) 

116 dataProvided: DataProvided = Field( 

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

118 json_schema_extra={ 

119 "example": "{" 

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

121 ' "attrs": ["relativeHumidity"]' 

122 "}," 

123 }, 

124 ) 

125 status: Optional[Status] = Field( 

126 default=Status.ACTIVE, 

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

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

129 "provided at registration creation time, new registration " 

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

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

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

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

134 "experiencing problems with notifications, the status is " 

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

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

137 ) 

138 expires: Optional[datetime] = Field( 

139 default=None, 

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

141 "Permanent registrations must omit this field.", 

142 ) 

143 forwardingInformation: Optional[ForwardingInformation] = Field( 

144 default=None, 

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

146 "against the provider. Automatically provided by the " 

147 "implementation, in the case such implementation supports " 

148 "forwarding capabilities.", 

149 )