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
« 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
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)
20from postrfp.shared.serial.common import Id, Pagination
22from .qmodels import Question, QElement, Choice, ElGrid
24from postrfp.model import notes
25from postrfp.shared.constants import CEL_EXPRESSION_MAX_LENGTH
28CONSTRAINED_ID = Annotated[int, Field(lt=4294967295, gt=0)]
30CONSTRAINED_ID_LIST = Annotated[List[CONSTRAINED_ID], Field(max_length=1000)]
33class OrgType(str, Enum):
34 RESPONDENT = "RESPONDENT"
35 BUYER = "BUYER"
36 CONSULTANT = "CONSULTANT"
39class Error(BaseModel):
40 error: str
41 description: str
44class ErrorList(BaseModel):
45 description: str
46 errors: List[Error]
49class IdList(BaseModel):
50 ids: List[int]
53class Count(BaseModel):
54 """
55 Generic count model
56 """
58 description: str
59 count: int
62class ShortName(BaseModel):
63 name: Annotated[str, StringConstraints(min_length=1, max_length=255)]
66class AnswerAttachmentIds(BaseModel):
67 attachment_id: CONSTRAINED_ID
68 answer_id: CONSTRAINED_ID
71class NewProjectIds(Id):
72 section_id: int = Field(
73 ..., description="The ID of the Root section for the newly created project"
74 )
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)]
84class BaseOrganisation(BaseModel):
85 model_config = ConfigDict(from_attributes=True)
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
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)
101class Organisation(BaseOrganisation):
102 password_expiry: int = 0
105class Participant(BaseModel):
106 model_config = ConfigDict(from_attributes=True)
107 organisation: Organisation
108 role: Annotated[str, StringConstraints(max_length=255)]
111class ParticipantList(RootModel):
112 root: list[Participant]
115class UpdateParticipant(BaseModel):
116 org_id: Annotated[str, StringConstraints(max_length=50)]
117 role: Annotated[str, StringConstraints(max_length=255)]
120class UpdateParticipantList(RootModel):
121 model_config = ConfigDict(json_schema_extra={"maxItems": 20})
122 root: list[UpdateParticipant]
125class UserType(str, Enum):
126 standard = "standard"
127 restricted = "restricted"
130class UserId(BaseModel):
131 id: Annotated[str, StringConstraints(max_length=50)]
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 ]
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
157class UserList(RootModel):
158 root: list[BaseUser]
161class User(BaseUser):
162 organisation: Organisation
163 type: UserType = UserType.standard
164 roles: List[str]
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]
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 """
180 permissions: set[str] = Field(
181 validation_alias="sorted_permissions",
182 description="List of this user's permissions, based on their roles",
183 )
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 )
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
200class OrgWithUsers(BaseOrganisation):
201 model_config = ConfigDict(from_attributes=True)
202 users: List[BaseUser]
205class BaseIssue(BaseModel):
206 label: Annotated[str, StringConstraints(max_length=250)] | None = None
207 deadline: datetime | None = None
210class NewIssue(BaseIssue):
211 respondent_id: Optional[Annotated[str, StringConstraints(max_length=50)]] = None
212 respondent_email: Optional[EmailStr] = None
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
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
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
241class IssuesList(BaseModel):
242 data: List[ListIssue]
243 pagination: Pagination
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
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",)
264class Issue(UpdateableIssue):
265 """
266 Represents the directly editable fields of an Issue
268 Either respondent or respondent_email must be set, but not both
269 """
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 )
284class Issues(RootModel):
285 root: list[Issue]
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 )
298class IssueStatus(BaseModel):
299 new_status: IssueStatuses
302class IssueUseWorkflow(BaseModel):
303 use_workflow: bool
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 )
315target_docs = "ID of the Organisation to who this Note is addressed"
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 )
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)
338class ReadNotes(RootModel):
339 root: list[ReadNote]
342copy_ws_docs = "Copy weightings from an existing Weighting Set with this ID"
343initial_weight_docs = "Initial weighting value for each question and section"
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.
351 Only one of initial_value and source_weightset_id can be set
352 """
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 )
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
379class WeightSet(ShortName):
380 id: Optional[int] = None # default weightset
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)]
389class SecWeight(BaseModel):
390 section_id: CONSTRAINED_ID
391 weight: Annotated[float, Field(ge=0.0)]
394class Weightings(BaseModel):
395 questions: List[QWeight]
396 sections: List[SecWeight]
399class WeightingsDoc(Weightings):
400 weightset_id: Optional[CONSTRAINED_ID] = None
403class ProjectWeightings(BaseModel):
404 weightset: WeightSet
405 total: Weightings
406 instance: Weightings
409class ParentedWeighting(Weightings):
410 parent_absolute_weight: float
413class Score(BaseModel):
414 issue_id: CONSTRAINED_ID
415 score_value: Optional[Decimal] = None
416 scoreset_id: str = ""
417 comment: Optional[str] = None
420class SectionScore(BaseModel):
421 issue_id: CONSTRAINED_ID
422 score_value: Decimal
425class SectionScoreDoc(BaseModel):
426 question_id: CONSTRAINED_ID
427 scores: List[SectionScore]
430class SectionScoreDocs(RootModel):
431 root: Annotated[List[SectionScoreDoc], Field(min_length=1, max_length=100)]
434class ScoreSet(BaseModel):
435 model_config = ConfigDict(from_attributes=True)
436 scoreset_id: str
437 fullname: str
440class ScoringData(BaseModel):
441 scoreset_id: str
442 scores: List[dict]
445class ProjectPermission(BaseModel):
446 user: str
447 permissions: List[int]
450class TargetUser(BaseModel):
451 targetUser: Optional[str] = Field(None, min_length=3)
454class TargetUserList(RootModel):
455 root: List[TargetUser]
458class TreeNode(Id):
459 parent_id: Optional[int]
460 number: str
461 title: str
464class SummaryEvent(Id):
465 timestamp: datetime
466 user_id: str
467 event_type: str
470class FullEvent(SummaryEvent):
471 model_config = ConfigDict(from_attributes=True)
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]
483class EvIssue(BaseModel):
484 respondent_id: str
485 label: Optional[str]
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
496class AnsweredQElement(QElement):
497 answer: Annotated[str, StringConstraints(max_length=65536)]
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)]
507class ElementAnswerList(RootModel):
508 root: Annotated[List[ElementAnswer], Field(min_length=0, max_length=100)]
511class Answer(ElementAnswer):
512 issue_id: int = Field(..., description="ID of the associated Issue for this answer")
513 question_id: int
516ldoc = "Mapping of Issue ID to Question Element ID to answer"
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)
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
547class AllocatedTo(BaseModel):
548 allocated_to: str
549 question_instance_id: int
552class AllocatedToList(RootModel):
553 root: List[AllocatedTo]
555 def __iter__(self):
556 return iter(self.root)
559class AnswerStats(BaseModel):
560 model_config = ConfigDict(from_attributes=True)
561 allocated_to: Optional[str] = None
562 status: str
563 question_count: int
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
575class ImportableAnswersList(RootModel):
576 root: List[ImportableAnswers]
579class AnsweredQuestion(Question):
580 response_state: List[AnswerResponseState]
581 elements: ElGrid
584class SingleRespondentQuestion(Question):
585 model_config = ConfigDict(extra="ignore")
586 elements: ElGrid
587 respondent: Organisation
590class Node(Id):
591 model_config = ConfigDict(from_attributes=True)
592 number: str
593 title: str
596class NodeTypeEnum(str, Enum):
597 section = "section"
598 question = "question"
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
613class QI(BaseModel):
614 model_config = ConfigDict(from_attributes=True)
615 id: int
616 project_id: int
617 number: str
620class QuestionInstance(QI):
621 section_id: int
622 project_title: Optional[str] = None
625class ScoreGaps(BaseModel):
626 question_id: int
627 number: str
628 title: str
629 node_type: str
630 score_gap: float
631 weight: float
634class EditableSection(BaseModel):
635 title: Annotated[str, StringConstraints(min_length=1, max_length=255)]
636 description: Optional[str] = Field(None, max_length=1024)
639class Section(EditableSection):
640 model_config = ConfigDict(from_attributes=True)
642 id: Optional[PositiveInt] = None
643 parent_id: Optional[PositiveInt] = Field(...)
644 number: Optional[str] = None
647class FullSection(Section):
648 id: int
649 subsections: List[TreeNode]
650 questions: List[Question]
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)
659class ParentId(BaseModel):
660 new_parent_id: CONSTRAINED_ID
663class MoveSection(ParentId):
664 section_id: CONSTRAINED_ID
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 )
675 delete_orphans: bool = False
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
697class WorkflowSection(BaseModel):
698 section: Section
699 questions: List[AnswerResponseState]
700 count: int
703class Nodes(Node):
704 type: str
705 subsections: Optional[list["Nodes"]] = None
706 questions: Optional[list["Nodes"]] = None
709class NodesList(RootModel):
710 root: List[Nodes]
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 )
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")
733class ProjectField(BaseModel):
734 """
735 ProjectFields allow the user to associate custom data fields with their project.
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 """
741 model_config = ConfigDict(from_attributes=True)
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 )
750class NewCategory(BaseModel):
751 """Represents a category which can be assigned to a Project for classification and filtering"""
753 name: Annotated[str, StringConstraints(max_length=50, min_length=3)]
754 description: Annotated[str, StringConstraints(max_length=250)]
757class Category(NewCategory):
758 model_config = ConfigDict(from_attributes=True)
760 id: Optional[int] = None
763class UpdateableProject(BaseModel):
764 model_config = ConfigDict(from_attributes=True, populate_by_name=True)
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 )
776 maximum_score: Annotated[int, Field(gt=0, lt=2147483648)] = 10
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] = []
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] = []
800class FullProject(UpdateableProject):
801 model_config = ConfigDict(from_attributes=True)
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
815 def set_permissions(self, user):
816 self.permissions = user.sorted_permissions
818 @field_validator("permissions", mode="before")
819 def perms(cls, v):
820 return []
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
837class ProjectList(BaseModel):
838 data: List[ListProject]
839 pagination: Pagination
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
851class Supplier(BaseModel):
852 issues: List[Issue]
853 organisation: Organisation
854 users: List[User]
857class AnswerAttachment(Id):
858 model_config = ConfigDict(from_attributes=True)
859 filename: str
860 mimetype: str
861 size: str
862 size_bytes: int
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
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)]
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
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
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
899 model_config = ConfigDict(from_attributes=True)
902class IssueWatchList(BaseModel):
903 issue_id: int
904 watchlist: List[Watcher]
907class AnswerImportResult(BaseModel):
908 imported: List[str]
909 errors: List[str]
910 unchanged: List[str]
913class ImportAnswers(BaseModel):
914 source_issue_id: int
915 section_number: str
918class SectionImportDoc(BaseModel):
919 project_id: int
920 section_ids: List[int] = []
921 question_ids: List[int] = []
922 clone: bool = True
925class SectionImportResult(BaseModel):
926 section_count: int = 0
927 question_count: int = 0
930class TextReplace(BaseModel):
931 search_term: str
932 replace_term: str
933 dry_run: bool = True
936class HitTypes(str, Enum):
937 questions = "questions"
938 choices = "choices"
939 answers = "answers"
940 notes = "notes"
941 scoreComments = "scoreComments"
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
953class RelationshipType(BaseModel):
954 model_config = ConfigDict(from_attributes=True)
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)
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)]
967class NetworkRelationship(BaseModel):
968 model_config = ConfigDict(from_attributes=True, populate_by_name=True)
970 # relationship: str
971 relationship: RelationshipType = Field(..., alias="relationship_type")
972 from_org: BaseOrganisation
973 to_org: BaseOrganisation
976class NewTag(BaseModel):
977 """
978 A Tag is a keyword used to categorize questions
979 """
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"])
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()
995class Tag(NewTag):
996 model_config = ConfigDict(from_attributes=True)
998 id: Optional[Annotated[int, Field(gt=0, lt=2147483648)]] = None
1001class TagAssigns(BaseModel):
1002 question_instance_ids: CONSTRAINED_ID_LIST
1003 section_ids: CONSTRAINED_ID_LIST
1004 recursive: bool = False
1007class TagGroup(TagAssigns):
1008 tag_id: Annotated[int, Field(gt=0, lt=2147483648)]
1011class MatchedElement(BaseModel):
1012 title: str
1013 number: str
1014 el_type: str
1015 label: str
1016 choices: List[Choice]
1019class QSearchResult(BaseModel):
1020 distinct_question_count: int = Field(
1021 ..., description="Number of questions that match the query"
1022 )
1023 matched_elements: List[MatchedElement]
1026class AnswerSearch(BaseModel):
1027 number: str
1028 question_id: int
1031class AnswerSearchList(BaseModel):
1032 matches: list[AnswerSearch]
1035class ReplacedItem(BaseModel):
1036 change_type: str
1037 question_number: str
1038 old: str
1039 new: str
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 ]
1049class PublishResult(BaseModel):
1050 issue_id: int
1051 respondent_id: str
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)]
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 )
1080 @field_validator("event_type")
1081 @classmethod
1082 def validate_event_type(cls, event_type):
1083 from postrfp.model.audit import evt_types
1085 if not hasattr(evt_types, event_type):
1086 raise ValueError(f"{event_type} is not a valid Event Type")
1087 return event_type
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
1099class WebhookExecList(BaseModel):
1100 data: list[WebhookExec]
1101 pagination: Pagination
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 )
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
1134class DatafeedExecList(BaseModel):
1135 data: list[DatafeedExec]
1136 pagination: Pagination