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 

7from .constants import ( 

8 DESC_DEFAULT_VALUE, 

9 DESC_FIELD_SCHEMA, 

10 DESC_FIELD_TYPE, 

11 DESC_JSON_POINTER_PATH, 

12 DESC_JSON_POINTER_PATH_DELETE, 

13 DESC_JSON_POINTER_PATH_EXISTING, 

14 DESC_JSON_POINTER_PATH_RENAMED, 

15 DESC_JSON_POINTER_PATH_FROM, 

16 DESC_JSON_POINTER_PATH_TO, 

17) 

18 

19# COMMON TYPES 

20 

21AUTH_POLICY_TYPE = Annotated[ 

22 str, 

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

24 StringConstraints(max_length=4000), 

25] 

26 

27JSON_TYPES = str | int | float | bool | list | dict 

28 

29 

30# UNIFIED MODELS 

31 

32 

33class ContentSchema(TimestampedId): 

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

35 

36 # Required/core fields 

37 name: str 

38 description: Optional[str] = None 

39 version: int | None = None 

40 spec_doc: dict[str, Any] 

41 is_draft: bool | None = None 

42 auth_policy: Optional[AUTH_POLICY_TYPE] = None 

43 is_validated: bool | None = None 

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

45 

46 # Optional for input, present in responses 

47 org_id: Optional[str] = None 

48 

49 model_config = ConfigDict(from_attributes=True) 

50 

51 

52class TagSummary(BaseModel): 

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

54 

55 id: int 

56 name: Optional[str] = None 

57 description: Optional[str] = None 

58 

59 model_config = ConfigDict(from_attributes=True) 

60 

61 

62class SubjectSummary(BaseModel): 

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

64 

65 id: int 

66 parent_id: Optional[int] = None 

67 name: Optional[str] = None 

68 code: Optional[str] = None 

69 description: Optional[str] = None 

70 subject_type: Optional[str] = None 

71 

72 model_config = ConfigDict(from_attributes=True) 

73 

74 

75class ContentDocument(TimestampedId): 

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

77 

78 # Required/core fields 

79 title: str 

80 schema_id: int 

81 is_validated: bool | None = None 

82 

83 # Optional fields 

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

85 primary_subject_id: Optional[int] = None 

86 auth_policy: Optional[AUTH_POLICY_TYPE] = None 

87 

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

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

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

91 

92 # Response-only fields 

93 author_org_id: Optional[str] = None 

94 last_updated_by_id: Optional[str] = None 

95 

96 model_config = ConfigDict(from_attributes=True) 

97 

98 

99# PERMISSION MODELS 

100 

101 

102class PolicyRequest(BaseModel): 

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

104 

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

106 entity_id: int 

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

108 

109 

110class PolicyResponse(PolicyRequest): 

111 """Response from policy operations""" 

112 

113 success: bool 

114 entity_name: Optional[str] = None 

115 error: Optional[str] = None 

116 operation_performed: Optional[str] = None 

117 

118 

119class PermissionUpdated(BaseModel): 

120 """Response for individual permission update operations""" 

121 

122 operation_performed: str 

123 permission_name: str 

124 entity_type: str 

125 entity_name: str 

126 target_org: str 

127 

128 model_config = ConfigDict(from_attributes=True) 

129 

130 

131class SubjectDocument(TimestampedId): 

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

133 

134 # Required/core fields 

135 name: str 

136 subject_type: str # Using string to handle enum serialization 

137 

138 # Optional fields 

139 code: Optional[str] = None 

140 description: Optional[str] = None 

141 auth_policy: Optional[AUTH_POLICY_TYPE] = None 

142 parent_id: Optional[int] = None 

143 managing_org_id: Optional[str] = None 

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

145 

146 model_config = ConfigDict(from_attributes=True) 

147 

148 

149class ContentQElementPairDocument(BaseModel): 

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

151 

152 # Optional for input, required for responses 

153 id: Optional[int] = None 

154 

155 # Required fields 

156 question_element_id: int 

157 content_reference: str = Field( 

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

159 ) 

160 

161 # Response-only field (parent relationship) 

162 content_map_id: Optional[int] = None 

163 

164 model_config = ConfigDict(from_attributes=True) 

165 

166 

167class ContentSpecMapDocument(BaseModel): 

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

169 

170 # Optional for input, required for responses 

171 id: Optional[int] = Field( 

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

173 ) 

174 

175 # Required fields 

176 name: str 

177 description: str 

178 content_spec_id: int 

179 

180 content_spec_name: Optional[str] = Field( 

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

182 ) 

183 

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

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

186 

187 model_config = ConfigDict(from_attributes=True) 

188 

189 

190# GENERIC RESPONSE TYPES 

191 

192 

193class ListResponse(BaseModel): 

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

195 

196 items: ( 

197 list[ContentDocument] 

198 | list[SubjectSummary] 

199 | list[ContentSchema] 

200 | list[ContentSpecMapDocument] 

201 ) 

202 pagination: Pagination 

203 

204 model_config = ConfigDict(from_attributes=True) 

205 

206 

207class DeletionResponse(BaseModel): 

208 """Standard deletion response""" 

209 

210 success: bool 

211 message: str 

212 

213 model_config = ConfigDict(from_attributes=True) 

214 

215 

216class SchemaPointersResponse(BaseModel): 

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

218 

219 content_spec_id: int 

220 content_spec_name: str 

221 pointers: list[str] = Field( 

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

223 ) 

224 

225 model_config = ConfigDict(from_attributes=True) 

226 

227 

228class JsonPatchOp(BaseModel): 

229 """JSON Patch operation model""" 

230 

231 op: Literal["add", "remove", "replace", "move", "copy", "test"] 

232 path: str 

233 value: JSON_TYPES | None = None 

234 from_: str | None = Field( 

235 default=None, alias="from", exclude_if=lambda v: v is None 

236 ) 

237 

238 model_config = {"populate_by_name": True, "serialize_by_alias": True} 

239 

240 

241class ContentPatch(BaseModel): 

242 """List of JSON Patch operations""" 

243 

244 patches: list[JsonPatchOp] 

245 comment: str 

246 

247 

248# Request models for each migration operation 

249 

250 

251class AddOptionalFieldRequest(BaseModel): 

252 """Request to add an optional field to a JSON schema""" 

253 

254 path: str = Field(description=DESC_JSON_POINTER_PATH) 

255 field_type: str = Field(description=DESC_FIELD_TYPE) 

256 field_schema: dict | None = Field( 

257 default=None, 

258 description=DESC_FIELD_SCHEMA, 

259 ) 

260 

261 

262class AddRequiredFieldRequest(BaseModel): 

263 """Request to add a required field to a JSON schema and update existing documents""" 

264 

265 path: str = Field(description=DESC_JSON_POINTER_PATH) 

266 field_type: str = Field(description=DESC_FIELD_TYPE) 

267 default_value: JSON_TYPES = Field(description=DESC_DEFAULT_VALUE) 

268 field_schema: dict | None = Field( 

269 default=None, 

270 description=DESC_FIELD_SCHEMA, 

271 ) 

272 

273 

274class DeleteFieldRequest(BaseModel): 

275 """Request to delete a field from a JSON schema and existing documents""" 

276 

277 path: str = Field(description=DESC_JSON_POINTER_PATH_DELETE) 

278 

279 

280class RenameFieldRequest(BaseModel): 

281 """Request to rename a field in a JSON schema and existing documents""" 

282 

283 old_path: str = Field(description=DESC_JSON_POINTER_PATH_EXISTING) 

284 new_path: str = Field(description=DESC_JSON_POINTER_PATH_RENAMED) 

285 

286 

287class MoveFieldRequest(BaseModel): 

288 """Request to move a field to a different location in the schema""" 

289 

290 from_path: str = Field(description=DESC_JSON_POINTER_PATH_FROM) 

291 to_path: str = Field(description=DESC_JSON_POINTER_PATH_TO) 

292 

293 

294class SchemaUpdate(BaseModel): 

295 """Schema update with optional migration operation""" 

296 

297 request_type: Literal[ 

298 "add_optional_field", 

299 "add_required_field", 

300 "delete_field", 

301 "rename_field", 

302 "move_field", 

303 ] 

304 request: ( 

305 AddOptionalFieldRequest 

306 | AddRequiredFieldRequest 

307 | DeleteFieldRequest 

308 | RenameFieldRequest 

309 | MoveFieldRequest 

310 )