Coverage for postrfp/shared/serial/refmodels.py: 100%

1from typing import Any, Optional, Literal, Annotated 

2 

3from pydantic import BaseModel, Field, ConfigDict, StringConstraints 

4 

5from postrfp.shared.serial.common import Pagination, TimestampedId 

6 

7 

8# COMMON TYPES 

9 

10AUTH_POLICY_TYPE = Annotated[ 

11 str, 

12 Field(description="Authorization policy - CEL expression"), 

13 StringConstraints(max_length=4000), 

14] 

15 

16 

17# UNIFIED MODELS 

18 

19 

20class ContentSchema(TimestampedId): 

21 """Unified content schema model - works for create, update, and response""" 

22 

23 # Required/core fields 

24 name: str 

25 description: Optional[str] = None 

26 spec_doc: dict[str, Any] 

27 auth_policy: Optional[AUTH_POLICY_TYPE] = None 

28 tags: list[str] = Field(default_factory=list) 

29 

30 # Optional for input, present in responses 

31 org_id: Optional[str] = None 

32 

33 model_config = ConfigDict(from_attributes=True) 

34 

35 

36class TagSummary(BaseModel): 

37 """Tag reference - can be minimal for lists or full for details""" 

38 

39 id: int 

40 name: Optional[str] = None 

41 description: Optional[str] = None 

42 

43 model_config = ConfigDict(from_attributes=True) 

44 

45 

46class SubjectSummary(BaseModel): 

47 """Subject reference - can be minimal for lists or full for details""" 

48 

49 id: int 

50 parent_id: Optional[int] = None 

51 name: Optional[str] = None 

52 code: Optional[str] = None 

53 description: Optional[str] = None 

54 subject_type: Optional[str] = None 

55 

56 model_config = ConfigDict(from_attributes=True) 

57 

58 

59class ContentDocument(TimestampedId): 

60 """Unified content model - adapts based on usage context""" 

61 

62 # Required/core fields 

63 title: str 

64 schema_id: int 

65 

66 # Optional fields 

67 content_doc: Optional[dict[str, Any]] = None # Full content only in detail views 

68 primary_subject_id: Optional[int] = None 

69 auth_policy: Optional[AUTH_POLICY_TYPE] = None 

70 

71 # Flexible references - can be IDs (input) or objects (output) 

72 tags: list[TagSummary] = Field(default_factory=list) 

73 subjects: list[SubjectSummary] = Field(default_factory=list) 

74 

75 # Response-only fields 

76 author_org_id: Optional[str] = None 

77 last_updated_by_id: Optional[str] = None 

78 

79 model_config = ConfigDict(from_attributes=True) 

80 

81 

82class ContentRelationships(BaseModel): 

83 """Content relationship data using the unified Content model""" 

84 

85 children: list[ContentDocument] = Field(default_factory=list) 

86 parents: list[ContentDocument] = Field(default_factory=list) 

87 references: list[ContentDocument] = Field(default_factory=list) 

88 related: list[ContentDocument] = Field(default_factory=list) 

89 superseded_by: list[ContentDocument] = Field(default_factory=list) 

90 supersedes: list[ContentDocument] = Field(default_factory=list) 

91 derived_from: list[ContentDocument] = Field(default_factory=list) 

92 derived: list[ContentDocument] = Field(default_factory=list) 

93 

94 

95class ContentRelationshipDoc(BaseModel): 

96 """Content with its relationships""" 

97 

98 content_id: int 

99 content_title: str 

100 relationships: ContentRelationships 

101 

102 

103# PERMISSION MODELS 

104 

105 

106class PolicyRequest(BaseModel): 

107 """Request to update/delete a policy""" 

108 

109 entity_type: Literal["Content", "ContentSpec", "Subject"] 

110 entity_id: int 

111 policy: Optional[str] = None # None for delete operations 

112 

113 

114class PolicyResponse(PolicyRequest): 

115 """Response from policy operations""" 

116 

117 success: bool 

118 entity_name: Optional[str] = None 

119 error: Optional[str] = None 

120 operation_performed: Optional[str] = None 

121 

122 

123class PermissionUpdated(BaseModel): 

124 """Response for individual permission update operations""" 

125 

126 operation_performed: str 

127 permission_name: str 

128 entity_type: str 

129 entity_name: str 

130 target_org: str 

131 

132 model_config = ConfigDict(from_attributes=True) 

133 

134 

135class SubjectDocument(TimestampedId): 

136 """Unified subject model - works for create, update, and response""" 

137 

138 # Required/core fields 

139 name: str 

140 subject_type: str # Using string to handle enum serialization 

141 

142 # Optional fields 

143 code: Optional[str] = None 

144 description: Optional[str] = None 

145 auth_policy: Optional[AUTH_POLICY_TYPE] = None 

146 parent_id: Optional[int] = None 

147 managing_org_id: Optional[str] = None 

148 subject_metadata: Optional[dict[str, Any]] = None 

149 

150 model_config = ConfigDict(from_attributes=True) 

151 

152 

153class ContentQElementPairDocument(BaseModel): 

154 """Mapping between a question element and a content field via JSON Pointer""" 

155 

156 # Optional for input, required for responses 

157 id: Optional[int] = None 

158 

159 # Required fields 

160 question_element_id: int 

161 content_reference: str = Field( 

162 description="JSON Pointer expression to locate field in content document (e.g., '$.sla.uptime')" 

163 ) 

164 

165 # Response-only field (parent relationship) 

166 content_map_id: Optional[int] = None 

167 

168 model_config = ConfigDict(from_attributes=True) 

169 

170 

171class ContentSpecMapDocument(BaseModel): 

172 """Mapping between question elements and content fields""" 

173 

174 # Optional for input, required for responses 

175 id: Optional[int] = Field( 

176 None, description="Unique ID of the content spec map, assigned by the system" 

177 ) 

178 

179 # Required fields 

180 name: str 

181 description: str 

182 content_spec_id: int 

183 

184 content_spec_name: Optional[str] = Field( 

185 None, description="Name of the associated content spec, not required for input" 

186 ) 

187 

188 # Nested pairs - can be provided on create/update 

189 pairs: list[ContentQElementPairDocument] = Field(default_factory=list) 

190 

191 model_config = ConfigDict(from_attributes=True) 

192 

193 

194# GENERIC RESPONSE TYPES 

195 

196 

197class ListResponse(BaseModel): 

198 """Generic list response that works with any item type""" 

199 

200 items: ( 

201 list[ContentDocument] 

202 | list[SubjectSummary] 

203 | list[ContentSchema] 

204 | list[ContentSpecMapDocument] 

205 ) 

206 pagination: Pagination 

207 

208 model_config = ConfigDict(from_attributes=True) 

209 

210 

211class DeletionResponse(BaseModel): 

212 """Standard deletion response""" 

213 

214 success: bool 

215 message: str 

216 

217 model_config = ConfigDict(from_attributes=True) 

218 

219 

220class SchemaPointersResponse(BaseModel): 

221 """Response containing valid JSON Pointer paths for a schema""" 

222 

223 content_spec_id: int 

224 content_spec_name: str 

225 pointers: list[str] = Field( 

226 description="List of valid JSON Pointer paths that can be used in ContentQElementPair mappings" 

227 ) 

228 

229 model_config = ConfigDict(from_attributes=True)