Coverage for postrfp / shared / serial / models.py: 99%

654 statements  

« prev     ^ index     » next       coverage.py v7.12.0, created at 2025-12-03 01:35 +0000

1from enum import Enum 

2from datetime import datetime 

3from typing import List, Union, Optional, Any, Annotated 

4from typing_extensions import Self 

5from decimal import Decimal 

6 

7from pydantic import ( 

8 field_validator, 

9 model_validator, 

10 StringConstraints, 

11 ConfigDict, 

12 BaseModel, 

13 EmailStr, 

14 Field, 

15 PositiveInt, 

16 AnyHttpUrl, 

17 RootModel, 

18) 

19 

20from postrfp.shared.serial.common import Id, Pagination 

21 

22from .qmodels import Question, QElement, Choice, ElGrid 

23 

24from postrfp.model import notes 

25from postrfp.shared.constants import CEL_EXPRESSION_MAX_LENGTH 

26 

27 

28CONSTRAINED_ID = Annotated[int, Field(lt=4294967295, gt=0)] 

29 

30CONSTRAINED_ID_LIST = Annotated[List[CONSTRAINED_ID], Field(max_length=1000)] 

31 

32 

33class OrgType(str, Enum): 

34 RESPONDENT = "RESPONDENT" 

35 BUYER = "BUYER" 

36 CONSULTANT = "CONSULTANT" 

37 

38 

39class Error(BaseModel): 

40 error: str 

41 description: str 

42 

43 

44class ErrorList(BaseModel): 

45 description: str 

46 errors: List[Error] 

47 

48 

49class IdList(BaseModel): 

50 ids: List[int] 

51 

52 

53class Count(BaseModel): 

54 """ 

55 Generic count model 

56 """ 

57 

58 description: str 

59 count: int 

60 

61 

62class ShortName(BaseModel): 

63 name: Annotated[str, StringConstraints(min_length=1, max_length=255)] 

64 

65 

66class AnswerAttachmentIds(BaseModel): 

67 attachment_id: CONSTRAINED_ID 

68 answer_id: CONSTRAINED_ID 

69 

70 

71class NewProjectIds(Id): 

72 section_id: int = Field( 

73 ..., description="The ID of the Root section for the newly created project" 

74 ) 

75 

76 

77class NewClient(BaseModel): 

78 org_name: Annotated[str, StringConstraints(min_length=4, max_length=50)] 

79 domain_name: Optional[Annotated[str, StringConstraints(max_length=256)]] = None 

80 administrator_email: EmailStr 

81 administrator_name: Annotated[str, StringConstraints(min_length=4, max_length=50)] 

82 

83 

84class BaseOrganisation(BaseModel): 

85 model_config = ConfigDict(from_attributes=True) 

86 

87 id: Annotated[str, StringConstraints(min_length=3, max_length=50)] 

88 name: Annotated[str, StringConstraints(min_length=3, max_length=50)] 

89 type: OrgType | None = None 

90 public: bool = False 

91 domain_name: Optional[str] = None 

92 

93 @field_validator("type", mode="before") 

94 @classmethod 

95 def set_type(cls, v): 

96 # If from ORM then v is an enum value of OrganisationType 

97 # if so, serialise to the name 

98 return getattr(v, "name", v) 

99 

100 

101class Organisation(BaseOrganisation): 

102 password_expiry: int = 0 

103 

104 

105class Participant(BaseModel): 

106 model_config = ConfigDict(from_attributes=True) 

107 organisation: Organisation 

108 role: Annotated[str, StringConstraints(max_length=255)] 

109 

110 

111class ParticipantList(RootModel): 

112 root: list[Participant] 

113 

114 

115class UpdateParticipant(BaseModel): 

116 org_id: Annotated[str, StringConstraints(max_length=50)] 

117 role: Annotated[str, StringConstraints(max_length=255)] 

118 

119 

120class UpdateParticipantList(RootModel): 

121 model_config = ConfigDict(json_schema_extra={"maxItems": 20}) 

122 root: list[UpdateParticipant] 

123 

124 

125class UserType(str, Enum): 

126 standard = "standard" 

127 restricted = "restricted" 

128 

129 

130class UserId(BaseModel): 

131 id: Annotated[str, StringConstraints(max_length=50)] 

132 

133 

134class EditableUser(UserId): 

135 model_config = ConfigDict(from_attributes=True) 

136 id: Annotated[str, StringConstraints(min_length=5, max_length=50)] 

137 org_id: Annotated[str, StringConstraints(min_length=4, max_length=50)] 

138 fullname: Annotated[str, StringConstraints(min_length=5, max_length=50)] 

139 email: EmailStr 

140 type: UserType = UserType.standard 

141 roles: Annotated[ 

142 List[Annotated[str, StringConstraints(min_length=3, max_length=255)]], 

143 Field(max_length=20), 

144 ] 

145 

146 

147class BaseUser(BaseModel): 

148 model_config = ConfigDict(from_attributes=True, populate_by_name=True) 

149 id: str 

150 org_id: str 

151 type: str 

152 fullname: str 

153 email: str 

154 previous_login_date: Optional[datetime] = None 

155 

156 

157class UserList(RootModel): 

158 root: list[BaseUser] 

159 

160 

161class User(BaseUser): 

162 organisation: Organisation 

163 type: UserType = UserType.standard 

164 roles: List[str] 

165 

166 @field_validator("roles", mode="before") 

167 def get_roles(cls, val_list): 

168 if len(val_list) and isinstance(val_list[0], str): 

169 return val_list 

170 else: 

171 return [role.role_id for role in val_list] 

172 

173 

174class FullUser(User): 

175 """ 

176 Don't use for incoming data - permissions are derived from user's roles 

177 thus should not be set directly. 

178 """ 

179 

180 permissions: set[str] = Field( 

181 validation_alias="sorted_permissions", 

182 description="List of this user's permissions, based on their roles", 

183 ) 

184 

185 # Note - if using a validation_alias then json_schema_mode_override must 

186 # be set to "serialization" or the json schema for this model will use 

187 # the validation alias instead of the field name. 

188 model_config = ConfigDict( 

189 from_attributes=True, json_schema_mode_override="serialization" 

190 ) 

191 

192 

193class ProjectUser(User): 

194 effectivePermissions: list[str] = Field(default_factory=list) 

195 userPermissions: list[str] = Field(default_factory=list) 

196 participantPermissions: list[str] = Field(default_factory=list) 

197 participantRole: Optional[str] = None 

198 

199 

200class OrgWithUsers(BaseOrganisation): 

201 model_config = ConfigDict(from_attributes=True) 

202 users: List[BaseUser] 

203 

204 

205class BaseIssue(BaseModel): 

206 label: Annotated[str, StringConstraints(max_length=250)] | None = None 

207 deadline: datetime | None = None 

208 

209 

210class NewIssue(BaseIssue): 

211 respondent_id: Optional[Annotated[str, StringConstraints(max_length=50)]] = None 

212 respondent_email: Optional[EmailStr] = None 

213 

214 @model_validator(mode="after") 

215 def either_email_or_respondent(self) -> Self: 

216 if not self.respondent_email and not self.respondent_id: 

217 raise ValueError("Either respondent_id or respondent_email must be set") 

218 elif self.respondent_id and self.respondent_email: 

219 raise ValueError("Not permitted to set both respondent or respondent_email") 

220 return self 

221 

222 @field_validator("respondent_email", "respondent_id") 

223 def nonify(cls, value): 

224 if value is not None and value.strip() == "": 

225 return None 

226 return value 

227 

228 

229class ListIssue(NewIssue): 

230 model_config = ConfigDict(from_attributes=True) 

231 issue_id: int 

232 winloss_exposed: bool = False 

233 winloss_expiry: Optional[datetime] = None 

234 project_title: str 

235 project_id: int 

236 status: str 

237 issue_date: Optional[datetime] = None 

238 submitted_date: Optional[datetime] = None 

239 

240 

241class IssuesList(BaseModel): 

242 data: List[ListIssue] 

243 pagination: Pagination 

244 

245 

246class UpdateableIssue(BaseIssue): 

247 award_status: Optional[int] = None 

248 internal_comments: Optional[str] = None 

249 feedback: Optional[str] = None 

250 winloss_exposed: bool = False 

251 winloss_expiry: Optional[datetime] = None 

252 

253 

254class IssueStatuses(str, Enum): 

255 NOT_SENT = ("Not Sent",) 

256 OPPORTUNITY = ("Opportunity",) 

257 ACCEPTED = ("Accepted",) 

258 UPDATEABLE = ("Updateable",) 

259 DECLINED = ("Declined",) 

260 SUBMITTED = ("Submitted",) 

261 RETRACTED = ("Retracted",) 

262 

263 

264class Issue(UpdateableIssue): 

265 """ 

266 Represents the directly editable fields of an Issue 

267 

268 Either respondent or respondent_email must be set, but not both 

269 """ 

270 

271 model_config = ConfigDict(from_attributes=True) 

272 id: int 

273 project_id: int 

274 respondent: Optional[Organisation] = None 

275 respondent_email: Optional[EmailStr] = None 

276 status: str 

277 accepted_date: Optional[datetime] = None 

278 submitted_date: Optional[datetime] = None 

279 issue_date: Optional[datetime] = Field( 

280 None, description="Date the Issue was published" 

281 ) 

282 

283 

284class Issues(RootModel): 

285 root: list[Issue] 

286 

287 

288class VendorIssue(Issue): 

289 title: str = Field(..., description="Title of the parent Project") 

290 owner_org_name: str = Field( 

291 ..., description="Name of the Organisation that owns the Project" 

292 ) 

293 is_watched: bool = Field( 

294 False, description="Shows if the current user is watching this issue" 

295 ) 

296 

297 

298class IssueStatus(BaseModel): 

299 new_status: IssueStatuses 

300 

301 

302class IssueUseWorkflow(BaseModel): 

303 use_workflow: bool 

304 

305 

306class RespondentNote(BaseModel): 

307 note_text: str 

308 private: bool = Field( 

309 False, 

310 alias="internal", 

311 description="If true, visible only to the respondent organisation", 

312 ) 

313 

314 

315target_docs = "ID of the Organisation to who this Note is addressed" 

316 

317 

318class ProjectNote(BaseModel): 

319 target_org_id: Optional[str] = Field(None, description=target_docs) 

320 note_text: Annotated[str, StringConstraints(max_length=16384, min_length=1)] 

321 private: bool = Field( 

322 False, 

323 alias="internal", 

324 description="If true, visible only to buyer (participant) organisations", 

325 ) 

326 

327 

328class ReadNote(ProjectNote): 

329 id: Optional[int] = Field(None, alias="note_id") 

330 note_time: datetime 

331 org_id: Annotated[str, StringConstraints(max_length=50)] = Field( 

332 ..., alias="organisation_id" 

333 ) 

334 distribution: notes.Distribution 

335 model_config = ConfigDict(from_attributes=True, populate_by_name=True) 

336 

337 

338class ReadNotes(RootModel): 

339 root: list[ReadNote] 

340 

341 

342copy_ws_docs = "Copy weightings from an existing Weighting Set with this ID" 

343initial_weight_docs = "Initial weighting value for each question and section" 

344 

345 

346class NewWeightSet(ShortName): 

347 """ 

348 Create a new weighting set. This can be populated with either an initial value 

349 for each section and question (initial_value field) or by copying an existing weightset. 

350 

351 Only one of initial_value and source_weightset_id can be set 

352 """ 

353 

354 initial_value: Optional[int] = Field( 

355 ge=0, le=100, description=initial_weight_docs, default=None 

356 ) 

357 source_weightset_id: Optional[int] = Field( 

358 ge=0, lt=2147483648, description=copy_ws_docs, default=None 

359 ) 

360 

361 @model_validator(mode="before") 

362 @classmethod 

363 def check_src_weightset(cls, values): 

364 initial_value = values.get("initial_value", None) 

365 source_weightset_id = values.get("source_weightset_id", None) 

366 if source_weightset_id is None: 

367 if initial_value is None: 

368 raise ValueError( 

369 "Both of initial_value & source_weightset_id cannot be null" 

370 ) 

371 else: 

372 if initial_value is not None: 

373 raise ValueError( 

374 "initial_value must be null if source_weightset_id is defined" 

375 ) 

376 return values 

377 

378 

379class WeightSet(ShortName): 

380 id: Optional[int] = None # default weightset 

381 

382 

383class QWeight(BaseModel): 

384 model_config = ConfigDict(populate_by_name=True) 

385 question_instance_id: CONSTRAINED_ID = Field(alias="question_id") 

386 weight: Annotated[float, Field(ge=0.0)] 

387 

388 

389class SecWeight(BaseModel): 

390 section_id: CONSTRAINED_ID 

391 weight: Annotated[float, Field(ge=0.0)] 

392 

393 

394class Weightings(BaseModel): 

395 questions: List[QWeight] 

396 sections: List[SecWeight] 

397 

398 

399class WeightingsDoc(Weightings): 

400 weightset_id: Optional[CONSTRAINED_ID] = None 

401 

402 

403class ProjectWeightings(BaseModel): 

404 weightset: WeightSet 

405 total: Weightings 

406 instance: Weightings 

407 

408 

409class ParentedWeighting(Weightings): 

410 parent_absolute_weight: float 

411 

412 

413class Score(BaseModel): 

414 issue_id: CONSTRAINED_ID 

415 score_value: Optional[Decimal] = None 

416 scoreset_id: str = "" 

417 comment: Optional[str] = None 

418 

419 

420class SectionScore(BaseModel): 

421 issue_id: CONSTRAINED_ID 

422 score_value: Decimal 

423 

424 

425class SectionScoreDoc(BaseModel): 

426 question_id: CONSTRAINED_ID 

427 scores: List[SectionScore] 

428 

429 

430class SectionScoreDocs(RootModel): 

431 root: Annotated[List[SectionScoreDoc], Field(min_length=1, max_length=100)] 

432 

433 

434class ScoreSet(BaseModel): 

435 model_config = ConfigDict(from_attributes=True) 

436 scoreset_id: str 

437 fullname: str 

438 

439 

440class ScoringData(BaseModel): 

441 scoreset_id: str 

442 scores: List[dict] 

443 

444 

445class ProjectPermission(BaseModel): 

446 user: str 

447 permissions: List[int] 

448 

449 

450class TargetUser(BaseModel): 

451 targetUser: Optional[str] = Field(None, min_length=3) 

452 

453 

454class TargetUserList(RootModel): 

455 root: List[TargetUser] 

456 

457 

458class TreeNode(Id): 

459 parent_id: Optional[int] 

460 number: str 

461 title: str 

462 

463 

464class SummaryEvent(Id): 

465 timestamp: datetime 

466 user_id: str 

467 event_type: str 

468 

469 

470class FullEvent(SummaryEvent): 

471 model_config = ConfigDict(from_attributes=True) 

472 

473 id: int 

474 event_class: str 

475 question_id: Optional[int] 

476 project_id: Optional[int] 

477 issue_id: Optional[int] 

478 org_id: str 

479 object_id: Any 

480 changes: List[dict] 

481 

482 

483class EvIssue(BaseModel): 

484 respondent_id: str 

485 label: Optional[str] 

486 

487 

488class AuditEvent(FullEvent): 

489 issue: Optional[EvIssue] = None 

490 user: Optional[User] 

491 question_number: Optional[str] = None 

492 section_id: Optional[int] = None 

493 project_title: Optional[str] = None 

494 

495 

496class AnsweredQElement(QElement): 

497 answer: Annotated[str, StringConstraints(max_length=65536)] 

498 

499 

500class ElementAnswer(BaseModel): 

501 element_id: int = Field( 

502 ..., description="ID of the question Element for this answer" 

503 ) 

504 answer: Annotated[str, StringConstraints(max_length=65536)] 

505 

506 

507class ElementAnswerList(RootModel): 

508 root: Annotated[List[ElementAnswer], Field(min_length=0, max_length=100)] 

509 

510 

511class Answer(ElementAnswer): 

512 issue_id: int = Field(..., description="ID of the associated Issue for this answer") 

513 question_id: int 

514 

515 

516ldoc = "Mapping of Issue ID to Question Element ID to answer" 

517 

518 

519class AnswerLookup(RootModel): 

520 model_config = ConfigDict( 

521 json_schema_extra={ 

522 "example": { 

523 "523": { 

524 "234872": "Answer for Issue 523, Element 234872", 

525 "234875": "Answer for Issue 523, Element 234875", 

526 }, 

527 "529": { 

528 "234872": "Answer for Issue 529, Element 234872", 

529 "234875": "Answer for Issue 529, Element 234875", 

530 }, 

531 } 

532 } 

533 ) 

534 root: dict[str, dict[str, str]] = Field(..., description=ldoc) 

535 

536 

537class AnswerResponseState(BaseModel): 

538 status: str 

539 allocated_by: Optional[str] 

540 allocated_to: Optional[str] 

541 approved_by: Optional[str] 

542 updated_by: Optional[str] 

543 date_updated: Optional[datetime] 

544 question_instance_id: int 

545 

546 

547class AllocatedTo(BaseModel): 

548 allocated_to: str 

549 question_instance_id: int 

550 

551 

552class AllocatedToList(RootModel): 

553 root: List[AllocatedTo] 

554 

555 def __iter__(self): 

556 return iter(self.root) 

557 

558 

559class AnswerStats(BaseModel): 

560 model_config = ConfigDict(from_attributes=True) 

561 allocated_to: Optional[str] = None 

562 status: str 

563 question_count: int 

564 

565 

566class ImportableAnswers(BaseModel): 

567 model_config = ConfigDict(from_attributes=True) 

568 title: str 

569 issue_date: Optional[datetime] 

570 submitted_date: Optional[datetime] 

571 question_count: int 

572 issue_id: int 

573 

574 

575class ImportableAnswersList(RootModel): 

576 root: List[ImportableAnswers] 

577 

578 

579class AnsweredQuestion(Question): 

580 response_state: List[AnswerResponseState] 

581 elements: ElGrid 

582 

583 

584class SingleRespondentQuestion(Question): 

585 model_config = ConfigDict(extra="ignore") 

586 elements: ElGrid 

587 respondent: Organisation 

588 

589 

590class Node(Id): 

591 model_config = ConfigDict(from_attributes=True) 

592 number: str 

593 title: str 

594 

595 

596class NodeTypeEnum(str, Enum): 

597 section = "section" 

598 question = "question" 

599 

600 

601class ProjectNode(Node): 

602 type: NodeTypeEnum 

603 parent_id: Optional[int] = Field( 

604 ..., description="ID of this node's parent section" 

605 ) 

606 position: int = Field(..., description="Position of the node within the parent") 

607 depth: int = Field( 

608 ..., description="The nested depth of this node within the questionnaire" 

609 ) 

610 description: Optional[str] = None 

611 

612 

613class QI(BaseModel): 

614 model_config = ConfigDict(from_attributes=True) 

615 id: int 

616 project_id: int 

617 number: str 

618 

619 

620class QuestionInstance(QI): 

621 section_id: int 

622 project_title: Optional[str] = None 

623 

624 

625class ScoreGaps(BaseModel): 

626 question_id: int 

627 number: str 

628 title: str 

629 node_type: str 

630 score_gap: float 

631 weight: float 

632 

633 

634class EditableSection(BaseModel): 

635 title: Annotated[str, StringConstraints(min_length=1, max_length=255)] 

636 description: Optional[str] = Field(None, max_length=1024) 

637 

638 

639class Section(EditableSection): 

640 model_config = ConfigDict(from_attributes=True) 

641 

642 id: Optional[PositiveInt] = None 

643 parent_id: Optional[PositiveInt] = Field(...) 

644 number: Optional[str] = None 

645 

646 

647class FullSection(Section): 

648 id: int 

649 subsections: List[TreeNode] 

650 questions: List[Question] 

651 

652 

653class SummarySection(Section): 

654 model_config = ConfigDict(from_attributes=True) 

655 questions: List[Node] = Field(default_factory=list) 

656 subsections: List[Node] = Field(default_factory=list) 

657 

658 

659class ParentId(BaseModel): 

660 new_parent_id: CONSTRAINED_ID 

661 

662 

663class MoveSection(ParentId): 

664 section_id: CONSTRAINED_ID 

665 

666 

667class SectionChildNodes(BaseModel): 

668 question_ids: Annotated[List[CONSTRAINED_ID], Field(max_length=100)] = Field( 

669 default_factory=list 

670 ) 

671 section_ids: Annotated[List[CONSTRAINED_ID], Field(max_length=100)] = Field( 

672 default_factory=list 

673 ) 

674 

675 delete_orphans: bool = False 

676 

677 @model_validator(mode="before") 

678 def either_or(cls, values): 

679 sec_ids, qids = ( 

680 values.get("section_ids", None), 

681 values.get("question_ids", None), 

682 ) 

683 if sec_ids is None and qids is None: 

684 raise ValueError("Either section_ids or question_ids must be set") 

685 if ( 

686 sec_ids is not None 

687 and qids is not None 

688 and len(sec_ids) > 0 

689 and len(qids) > 0 

690 ): 

691 raise ValueError( 

692 "Cannot assign values to both question_ids and section_ids" 

693 ) 

694 return values 

695 

696 

697class WorkflowSection(BaseModel): 

698 section: Section 

699 questions: List[AnswerResponseState] 

700 count: int 

701 

702 

703class Nodes(Node): 

704 type: str 

705 subsections: Optional[list["Nodes"]] = None 

706 questions: Optional[list["Nodes"]] = None 

707 

708 

709class NodesList(RootModel): 

710 root: List[Nodes] 

711 

712 

713class QElementStats(BaseModel): 

714 TX: int = Field(..., description="Count of Text Input elements") 

715 CR: int = Field(..., description="Count of Radio Choices elements") 

716 CC: int = Field(..., description="Count of Combon/Select elements") 

717 CB: int = Field(..., description="Count of Checkbox elements") 

718 LB: int = Field(..., description="Count of Label elements") 

719 QA: int = Field(..., description="Count of Question Attachment elements") 

720 AT: int = Field(..., description="Count of Upload Field Attachment elements") 

721 MD: int = Field(..., description="Count of Media elements") 

722 answerable_elements: int = Field( 

723 ..., description="Total number of answerable elements" 

724 ) 

725 

726 

727class QuestionnaireStats(BaseModel): 

728 questions: int = Field(..., description="Count of Questions in the Project") 

729 sections: int = Field(..., description="Count of Sections in the Project") 

730 elements: QElementStats = Field(..., description="Statistics for Question Elements") 

731 

732 

733class ProjectField(BaseModel): 

734 """ 

735 ProjectFields allow the user to associate custom data fields with their project. 

736 

737 By default such fields are invisible to respondent users - behaviour dictated by the 'private' 

738 property. ProjectFields are ordered when displayed according to the 'position' property. 

739 """ 

740 

741 model_config = ConfigDict(from_attributes=True) 

742 

743 key: Annotated[str, StringConstraints(min_length=1, max_length=64)] 

744 value: Annotated[str, StringConstraints(max_length=8192)] 

745 private: bool = Field( 

746 True, description="Indicates whether this field is visible to Respondents" 

747 ) 

748 

749 

750class NewCategory(BaseModel): 

751 """Represents a category which can be assigned to a Project for classification and filtering""" 

752 

753 name: Annotated[str, StringConstraints(max_length=50, min_length=3)] 

754 description: Annotated[str, StringConstraints(max_length=250)] 

755 

756 

757class Category(NewCategory): 

758 model_config = ConfigDict(from_attributes=True) 

759 

760 id: Optional[int] = None 

761 

762 

763class UpdateableProject(BaseModel): 

764 model_config = ConfigDict(from_attributes=True, populate_by_name=True) 

765 

766 title: str = Field("Untitled Project", min_length=5, max_length=256) 

767 # Default workflow_id is 1 - the standard project workflow 

768 workflow_id: int = 1 

769 description: Optional[str] = Field(None, max_length=20000) 

770 deadline: Optional[datetime] = None 

771 multiscored: bool = Field( 

772 False, 

773 description="Indicates whether Multiple Score Sets are enabled for this project", 

774 ) 

775 

776 maximum_score: Annotated[int, Field(gt=0, lt=2147483648)] = 10 

777 

778 email: Optional[EmailStr] = Field( 

779 None, 

780 alias="from_email_address", 

781 description=( 

782 "Send notification emails pertaining to the current project from this email " 

783 "address" 

784 ), 

785 ) 

786 project_fields: List[ProjectField] = [] 

787 

788 

789class NewProject(UpdateableProject): 

790 # title duplicates that of UpdateableProject but here it is a required field 

791 title: Annotated[ 

792 str, StringConstraints(strip_whitespace=True, min_length=5, max_length=256) 

793 ] 

794 questionnaire_title: Annotated[ 

795 str, StringConstraints(strip_whitespace=True, min_length=5, max_length=64) 

796 ] = "Questionnaire" 

797 category_ids: List[int] = [] 

798 

799 

800class FullProject(UpdateableProject): 

801 model_config = ConfigDict(from_attributes=True) 

802 

803 id: int 

804 status_name: str = Field( 

805 ..., serialization_alias="status", validation_alias="status" 

806 ) 

807 author_id: str 

808 owner_org: Optional[Organisation] = Field(None, alias="owner_organisation") 

809 date_published: Optional[datetime] = None 

810 date_created: datetime 

811 permissions: List[str] = [] 

812 categories: List[Category] 

813 section_id: Optional[int] = None 

814 

815 def set_permissions(self, user): 

816 self.permissions = user.sorted_permissions 

817 

818 @field_validator("permissions", mode="before") 

819 def perms(cls, v): 

820 return [] 

821 

822 

823class ListProject(BaseModel): 

824 model_config = ConfigDict(from_attributes=True) 

825 id: int 

826 workflow_id: int 

827 title: str 

828 owner_org_name: str 

829 owner_org_id: str 

830 status: str 

831 deadline: Optional[datetime] = None 

832 date_created: datetime 

833 watching_since: Optional[datetime] = None 

834 is_watched: bool = False 

835 

836 

837class ProjectList(BaseModel): 

838 data: List[ListProject] 

839 pagination: Pagination 

840 

841 

842class ProjectApproval(BaseModel): 

843 model_config = ConfigDict(from_attributes=True) 

844 id: Optional[int] = None 

845 project_id: int 

846 user_id: str 

847 project_status: str 

848 date_approved: datetime 

849 

850 

851class Supplier(BaseModel): 

852 issues: List[Issue] 

853 organisation: Organisation 

854 users: List[User] 

855 

856 

857class AnswerAttachment(Id): 

858 model_config = ConfigDict(from_attributes=True) 

859 filename: str 

860 mimetype: str 

861 size: str 

862 size_bytes: int 

863 

864 

865class Attachment(AnswerAttachment): 

866 model_config = ConfigDict(from_attributes=True) 

867 private: bool 

868 description: Annotated[str, StringConstraints(min_length=1, max_length=255)] 

869 date_uploaded: datetime 

870 author_id: str 

871 org_id: str 

872 

873 

874class IssueAttachment(Attachment): 

875 model_config = ConfigDict(from_attributes=True) 

876 issue_id: int 

877 description: Annotated[str, StringConstraints(min_length=1, max_length=1024)] 

878 

879 

880class Watcher(BaseModel): 

881 user_id: str 

882 fullname: str 

883 email: Optional[EmailStr] = None 

884 watching_since: Optional[datetime] = None 

885 is_watching: bool = False 

886 

887 @model_validator(mode="after") 

888 def set_is_watching(self, values) -> Self: 

889 self.is_watching = self.watching_since is not None 

890 return self 

891 

892 @field_validator("email", mode="before") 

893 @classmethod 

894 def none_if_zerolength(cls, email): 

895 if type(email) is str and len(email.strip()) == 0: 

896 return None 

897 return email 

898 

899 model_config = ConfigDict(from_attributes=True) 

900 

901 

902class IssueWatchList(BaseModel): 

903 issue_id: int 

904 watchlist: List[Watcher] 

905 

906 

907class AnswerImportResult(BaseModel): 

908 imported: List[str] 

909 errors: List[str] 

910 unchanged: List[str] 

911 

912 

913class ImportAnswers(BaseModel): 

914 source_issue_id: int 

915 section_number: str 

916 

917 

918class SectionImportDoc(BaseModel): 

919 project_id: int 

920 section_ids: List[int] = [] 

921 question_ids: List[int] = [] 

922 clone: bool = True 

923 

924 

925class SectionImportResult(BaseModel): 

926 section_count: int = 0 

927 question_count: int = 0 

928 

929 

930class TextReplace(BaseModel): 

931 search_term: str 

932 replace_term: str 

933 dry_run: bool = True 

934 

935 

936class HitTypes(str, Enum): 

937 questions = "questions" 

938 choices = "choices" 

939 answers = "answers" 

940 notes = "notes" 

941 scoreComments = "scoreComments" 

942 

943 

944class SearchResult(BaseModel): 

945 klass: HitTypes 

946 project_title: str 

947 project_id: int 

948 object_id: Union[str, int] 

949 object_ref: Union[str, None] = None 

950 snippet: str 

951 

952 

953class RelationshipType(BaseModel): 

954 model_config = ConfigDict(from_attributes=True) 

955 

956 id: Optional[int] = Field(None, gt=0, lt=2147483648) 

957 name: Annotated[str, StringConstraints(min_length=2, max_length=128)] 

958 description: Optional[str] = Field(None, max_length=256) 

959 

960 

961class Relationship(BaseModel): 

962 reltype_id: Annotated[int, Field(gt=0, lt=2147483648)] 

963 from_org_id: Annotated[str, StringConstraints(min_length=2, max_length=50)] 

964 to_org_id: Annotated[str, StringConstraints(min_length=2, max_length=50)] 

965 

966 

967class NetworkRelationship(BaseModel): 

968 model_config = ConfigDict(from_attributes=True, populate_by_name=True) 

969 

970 # relationship: str 

971 relationship: RelationshipType = Field(..., alias="relationship_type") 

972 from_org: BaseOrganisation 

973 to_org: BaseOrganisation 

974 

975 

976class NewTag(BaseModel): 

977 """ 

978 A Tag is a keyword used to categorize questions 

979 """ 

980 

981 name: Annotated[str, StringConstraints(min_length=2, max_length=128)] 

982 description: Optional[Annotated[str, StringConstraints(max_length=256)]] = None 

983 colour: Optional[ 

984 Annotated[str, StringConstraints(max_length=7, pattern=r"^#[0-9A-Fa-f]{6}$")] 

985 ] = Field(None, description="Hex color code (#RRGGBB)", examples=["#1E90FF"]) 

986 

987 @field_validator("colour", mode="before") 

988 @classmethod 

989 def normalise_colour(cls, v): 

990 if v is None: 

991 return v 

992 return v.upper() 

993 

994 

995class Tag(NewTag): 

996 model_config = ConfigDict(from_attributes=True) 

997 

998 id: Optional[Annotated[int, Field(gt=0, lt=2147483648)]] = None 

999 

1000 

1001class TagAssigns(BaseModel): 

1002 question_instance_ids: CONSTRAINED_ID_LIST 

1003 section_ids: CONSTRAINED_ID_LIST 

1004 recursive: bool = False 

1005 

1006 

1007class TagGroup(TagAssigns): 

1008 tag_id: Annotated[int, Field(gt=0, lt=2147483648)] 

1009 

1010 

1011class MatchedElement(BaseModel): 

1012 title: str 

1013 number: str 

1014 el_type: str 

1015 label: str 

1016 choices: List[Choice] 

1017 

1018 

1019class QSearchResult(BaseModel): 

1020 distinct_question_count: int = Field( 

1021 ..., description="Number of questions that match the query" 

1022 ) 

1023 matched_elements: List[MatchedElement] 

1024 

1025 

1026class AnswerSearch(BaseModel): 

1027 number: str 

1028 question_id: int 

1029 

1030 

1031class AnswerSearchList(BaseModel): 

1032 matches: list[AnswerSearch] 

1033 

1034 

1035class ReplacedItem(BaseModel): 

1036 change_type: str 

1037 question_number: str 

1038 old: str 

1039 new: str 

1040 

1041 

1042class PublishProject(BaseModel): 

1043 release_issue_ids: Annotated[ 

1044 List[Annotated[int, Field(gt=0, lt=2147483648)]], 

1045 Field(min_length=0, max_length=50), 

1046 ] 

1047 

1048 

1049class PublishResult(BaseModel): 

1050 issue_id: int 

1051 respondent_id: str 

1052 

1053 

1054class HttpHeader(BaseModel): 

1055 header: Annotated[ 

1056 str, StringConstraints(max_length=255, pattern=r"^[a-zA-Z0-9-]+$") 

1057 ] 

1058 value: Annotated[str, StringConstraints(max_length=2048)] 

1059 

1060 

1061class Webhook(BaseModel): 

1062 id: int | None = None 

1063 org_id: Annotated[str, StringConstraints(max_length=50)] 

1064 event_type: Annotated[str, StringConstraints(max_length=50)] 

1065 remote_url: AnyHttpUrl = Field(...) 

1066 guard_policy: ( 

1067 Annotated[str, StringConstraints(max_length=CEL_EXPRESSION_MAX_LENGTH)] | None 

1068 ) = None 

1069 transform_expression: ( 

1070 Annotated[str, StringConstraints(max_length=CEL_EXPRESSION_MAX_LENGTH)] | None 

1071 ) = None 

1072 http_headers: Optional[List[HttpHeader]] = Field( 

1073 default_factory=list, 

1074 description=( 

1075 "Optional list of HTTP headers to include in the webhook POST request" 

1076 ), 

1077 max_length=5, 

1078 ) 

1079 

1080 @field_validator("event_type") 

1081 @classmethod 

1082 def validate_event_type(cls, event_type): 

1083 from postrfp.model.audit import evt_types 

1084 

1085 if not hasattr(evt_types, event_type): 

1086 raise ValueError(f"{event_type} is not a valid Event Type") 

1087 return event_type 

1088 

1089 

1090class WebhookExec(BaseModel): 

1091 webhook_id: int 

1092 event_id: int 

1093 status: str 

1094 dagu_run_id: str 

1095 created_at: datetime 

1096 error_message: str | None = None 

1097 

1098 

1099class WebhookExecList(BaseModel): 

1100 data: list[WebhookExec] 

1101 pagination: Pagination 

1102 

1103 

1104class Datafeed(BaseModel): 

1105 model_config = ConfigDict(from_attributes=True) 

1106 uid: str | None = None 

1107 org_id: Annotated[str, StringConstraints(max_length=50)] 

1108 name: Annotated[str, StringConstraints(max_length=128)] 

1109 description: Annotated[str, StringConstraints(max_length=512)] | None = None 

1110 source_url: AnyHttpUrl = Field(...) 

1111 transform_expression: ( 

1112 Annotated[str, StringConstraints(max_length=CEL_EXPRESSION_MAX_LENGTH)] | None 

1113 ) = None 

1114 http_headers: Optional[List[HttpHeader]] = Field( 

1115 default_factory=list, 

1116 description=( 

1117 "Optional list of HTTP headers to include in the datafeed request" 

1118 ), 

1119 max_length=5, 

1120 ) 

1121 

1122 

1123class DatafeedExec(BaseModel): 

1124 model_config = ConfigDict(from_attributes=True) 

1125 datafeed_uid: str 

1126 status: str 

1127 dagu_run_id: str | None = None 

1128 created_at: datetime 

1129 error_message: str | None = None 

1130 result_data: dict[str, Any] | None = None 

1131 uid: str | None = None 

1132 

1133 

1134class DatafeedExecList(BaseModel): 

1135 data: list[DatafeedExec] 

1136 pagination: Pagination