Coverage for rfpy/web/serial/models.py: 99%
636 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-24 10:52 +0000
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-24 10:52 +0000
1import re
2from enum import Enum
3from datetime import datetime
4from typing import List, Union, Optional
5from typing_extensions import Self
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 .qmodels import Question, QElement, Choice, ElGrid
21from rfpy.model import notes
22from typing import Annotated
25CONSTRAINED_ID = Annotated[int, Field(lt=4294967295, gt=0)]
27CONSTRAINED_ID_LIST = Annotated[List[CONSTRAINED_ID], Field(max_length=1000)]
29DOMAIN_REGEX = (
30 r"(([\da-zA-Z])([_\w-]{0,62})\.)+(([\da-zA-Z])[_\w-]{0,61})?([\da-zA-Z]"
31 r"\.((xn\-\-[a-zA-Z\d]+)|([a-zA-Z\d]{2,})))"
32)
35class Pagination(BaseModel):
36 current_page: int
37 page_size: int
38 total_count: int
39 total_pages: int
40 has_next: bool
41 current_page_count: int = Field(
42 0, description="Number of records in the current page"
43 )
46class OrgType(str, Enum):
47 RESPONDENT = "RESPONDENT"
48 BUYER = "BUYER"
49 CONSULTANT = "CONSULTANT"
52class Error(BaseModel):
53 error: str
54 description: str
57class ErrorList(BaseModel):
58 description: str
59 errors: List[Error]
62class Id(BaseModel):
63 model_config = ConfigDict(from_attributes=True)
64 id: Optional[int] = None
67class StringId(BaseModel):
68 id: str
71class IdList(BaseModel):
72 ids: List[int]
75class Count(BaseModel):
76 """
77 Generic count model
78 """
80 description: str
81 count: int
84class ShortName(BaseModel):
85 name: Annotated[str, StringConstraints(min_length=1, max_length=255)]
88class AnswerAttachmentIds(BaseModel):
89 attachment_id: CONSTRAINED_ID
90 answer_id: CONSTRAINED_ID
93class NewProjectIds(Id):
94 section_id: int = Field(
95 ..., description="The ID of the Root section for the newly created project"
96 )
99class NewClient(BaseModel):
100 org_name: Annotated[str, StringConstraints(min_length=4, max_length=50)]
101 domain_name: Optional[Annotated[str, StringConstraints(max_length=256)]] = None
102 administrator_email: EmailStr
103 administrator_name: Annotated[str, StringConstraints(min_length=4, max_length=50)]
106class BaseOrganisation(BaseModel):
107 model_config = ConfigDict(from_attributes=True)
109 id: Annotated[str, StringConstraints(min_length=3, max_length=50)]
110 name: Annotated[str, StringConstraints(min_length=3, max_length=50)]
111 type: OrgType
112 public: bool = False
113 domain_name: Optional[str] = None
115 @field_validator("type", mode="before")
116 @classmethod
117 def set_type(cls, v):
118 # If from ORM then v is an enum value of OrganisationType
119 # if so, serialise to the name
120 return getattr(v, "name", v)
123class Organisation(BaseOrganisation):
124 password_expiry: int = 0
127class Participant(BaseModel):
128 model_config = ConfigDict(from_attributes=True)
129 organisation: Organisation
130 role: Annotated[str, StringConstraints(max_length=255)]
133class ParticipantList(RootModel):
134 root: list[Participant]
137class UpdateParticipant(BaseModel):
138 org_id: Annotated[str, StringConstraints(max_length=50)]
139 role: Annotated[str, StringConstraints(max_length=255)]
142class UpdateParticipantList(RootModel):
143 model_config = ConfigDict(json_schema_extra={"maxItems": 20})
144 root: list[UpdateParticipant]
147class UserType(str, Enum):
148 standard = "standard"
149 restricted = "restricted"
152class UserId(BaseModel):
153 id: Annotated[str, StringConstraints(max_length=50)]
156class EditableUser(UserId):
157 model_config = ConfigDict(from_attributes=True)
158 id: Annotated[str, StringConstraints(min_length=5, max_length=50)]
159 org_id: Annotated[str, StringConstraints(min_length=4, max_length=50)]
160 fullname: Annotated[str, StringConstraints(min_length=5, max_length=50)]
161 email: EmailStr
162 type: UserType = UserType.standard
163 roles: Annotated[
164 List[Annotated[str, StringConstraints(min_length=3, max_length=255)]],
165 Field(max_length=20),
166 ]
169class BaseUser(BaseModel):
170 model_config = ConfigDict(from_attributes=True, populate_by_name=True)
171 id: str
172 fullname: str
173 email: str
174 previous_login_date: Optional[datetime] = None
175 model_config = ConfigDict(from_attributes=True)
178class UserList(RootModel):
179 root: list[BaseUser]
182class User(BaseUser):
183 organisation: Organisation
184 type: UserType = UserType.standard
185 roles: List[str]
187 @field_validator("roles", mode="before")
188 def get_roles(cls, val_list):
189 if len(val_list) and isinstance(val_list[0], str):
190 return val_list
191 else:
192 return [role.role_id for role in val_list]
195class FullUser(User):
196 permissions: set[str] = Field(
197 default_factory=set, validation_alias="sorted_permissions"
198 )
199 model_config = ConfigDict(from_attributes=True)
202class ProjectUser(User):
203 effectivePermissions: list[str] = Field(default_factory=list)
204 userPermissions: list[str] = Field(default_factory=list)
205 participantPermissions: list[str] = Field(default_factory=list)
206 participantRole: Optional[str] = None
209class OrgWithUsers(BaseOrganisation):
210 model_config = ConfigDict(from_attributes=True)
211 users: List[BaseUser]
214class BaseIssue(BaseModel):
215 label: Annotated[str, StringConstraints(max_length=250)] | None = None
216 deadline: datetime | None = None
219class NewIssue(BaseIssue):
220 respondent_id: Optional[Annotated[str, StringConstraints(max_length=50)]] = None
221 respondent_email: Optional[EmailStr] = None
223 @model_validator(mode="after")
224 def either_email_or_respondent(self) -> Self:
225 if not self.respondent_email and not self.respondent_id:
226 raise ValueError("Either respondent_id or respondent_email must be set")
227 elif self.respondent_id and self.respondent_email:
228 raise ValueError("Not permitted to set both respondent or respondent_email")
229 return self
231 @field_validator("respondent_email", "respondent_id")
232 def nonify(cls, value):
233 if value is not None and value.strip() == "":
234 return None
235 return value
238class ListIssue(NewIssue):
239 model_config = ConfigDict(from_attributes=True)
240 issue_id: int
241 winloss_exposed: bool = False
242 winloss_expiry: Optional[datetime] = None
243 project_title: str
244 project_id: int
245 status: str
246 issue_date: Optional[datetime] = None
247 submitted_date: Optional[datetime] = None
250class IssuesList(BaseModel):
251 data: List[ListIssue]
252 pagination: Pagination
255class UpdateableIssue(BaseIssue):
256 award_status: Optional[int] = None
257 internal_comments: Optional[str] = None
258 feedback: Optional[str] = None
259 winloss_exposed: bool = False
260 winloss_expiry: Optional[datetime] = None
263class IssueStatuses(str, Enum):
264 NOT_SENT = ("Not Sent",)
265 OPPORTUNITY = ("Opportunity",)
266 ACCEPTED = ("Accepted",)
267 UPDATEABLE = ("Updateable",)
268 DECLINED = ("Declined",)
269 SUBMITTED = ("Submitted",)
270 RETRACTED = ("Retracted",)
273class Issue(UpdateableIssue):
274 """
275 Represents the directly editable fields of an Issue
277 Either respondent or respondent_email must be set, but not both
278 """
280 model_config = ConfigDict(from_attributes=True)
281 id: int
282 project_id: int
283 respondent: Optional[Organisation] = None
284 respondent_email: Optional[EmailStr] = None
285 status: str
286 accepted_date: Optional[datetime] = None
287 submitted_date: Optional[datetime] = None
288 issue_date: Optional[datetime] = Field(
289 None, description="Date the Issue was published"
290 )
293class Issues(RootModel):
294 root: list[Issue]
297class VendorIssue(Issue):
298 title: str = Field(..., description="Title of the parent Project")
299 owner_org: Organisation = Field(
300 ..., description="Org that created the Project - the Buyer"
301 )
302 is_watched: bool = Field(
303 False, description="Shows if the current user is watching this issue"
304 )
307class IssueStatus(BaseModel):
308 new_status: IssueStatuses
311class IssueUseWorkflow(BaseModel):
312 use_workflow: bool
315class RespondentNote(BaseModel):
316 note_text: str
317 private: bool = Field(
318 False,
319 alias="internal",
320 description="If true, visible only to the respondent organisation",
321 )
324target_docs = "ID of the Organisation to who this Note is addressed"
327class ProjectNote(BaseModel):
328 target_org_id: Optional[str] = Field(None, description=target_docs)
329 note_text: Annotated[str, StringConstraints(max_length=16384, min_length=1)]
330 private: bool = Field(
331 False,
332 alias="internal",
333 description="If true, visible only to buyer (participant) organisations",
334 )
337class ReadNote(ProjectNote):
338 id: Optional[int] = Field(None, alias="note_id")
339 note_time: datetime
340 org_id: Annotated[str, StringConstraints(max_length=50)] = Field(
341 ..., alias="organisation_id"
342 )
343 distribution: notes.Distribution
344 model_config = ConfigDict(from_attributes=True, populate_by_name=True)
347class ReadNotes(RootModel):
348 root: list[ReadNote]
351copy_ws_docs = "Copy weightings from an existing Weighting Set with this ID"
352initial_weight_docs = "Initial weighting value for each question and section"
355class NewWeightSet(ShortName):
356 """
357 Create a new weighting set. This can be populate with either an initial value for each section
358 and question (initial_value field) or by copying an existing weightset.
360 Only one of initial_value and source_weightset_id can be set
361 """
363 initial_value: Optional[int] = Field(
364 ge=0, le=100, description=initial_weight_docs, default=None
365 )
366 source_weightset_id: Optional[int] = Field(
367 ge=0, lt=2147483648, description=copy_ws_docs, default=None
368 )
370 @model_validator(mode="before")
371 @classmethod
372 def check_src_weightset(cls, values):
373 initial_value = values.get("initial_value", None)
374 source_weightset_id = values.get("source_weightset_id", None)
375 if source_weightset_id is None:
376 if initial_value is None:
377 raise ValueError(
378 "Both of initial_value & source_weightset_id cannot be null"
379 )
380 else:
381 if initial_value is not None:
382 raise ValueError(
383 "initial_value must be null if source_weightset_id is defined"
384 )
385 return values
388class WeightSet(ShortName):
389 id: Optional[int] = None # default weightset
392class QWeight(BaseModel):
393 question_id: CONSTRAINED_ID
394 weight: Annotated[float, Field(ge=0.0)]
397class SecWeight(BaseModel):
398 section_id: CONSTRAINED_ID
399 weight: Annotated[float, Field(ge=0.0)]
402class Weightings(BaseModel):
403 questions: List[QWeight]
404 sections: List[SecWeight]
407class WeightingsDoc(Weightings):
408 weightset_id: Optional[CONSTRAINED_ID] = None
411class ProjectWeightings(BaseModel):
412 weightset: WeightSet
413 total: Weightings
414 instance: Weightings
417class ParentedWeighting(Weightings):
418 parent_absolute_weight: float
421class Score(BaseModel):
422 issue_id: CONSTRAINED_ID
423 score_value: Optional[int] = None
424 scoreset_id: str = ""
425 comment: Optional[str] = None
428class SectionScore(BaseModel):
429 issue_id: CONSTRAINED_ID
430 score_value: int
433class SectionScoreDoc(BaseModel):
434 question_id: CONSTRAINED_ID
435 scores: List[SectionScore]
438class SectionScoreDocs(RootModel):
439 root: Annotated[List[SectionScoreDoc], Field(min_length=1, max_length=100)]
442class ScoreSet(BaseModel):
443 model_config = ConfigDict(from_attributes=True)
444 scoreset_id: str
445 fullname: str
448class ScoringData(BaseModel):
449 scoreset_id: str
450 scores: List[dict]
453class ProjectPermission(BaseModel):
454 user: str
455 permissions: List[int]
458class TargetUser(BaseModel):
459 targetUser: Optional[str] = Field(None, min_length=3)
462class TargetUserList(RootModel):
463 root: List[TargetUser]
466class TreeNode(Id):
467 parent_id: Optional[int]
468 number: str
469 title: str
472class SummaryEvent(Id):
473 timestamp: datetime
474 user_id: str
475 event_type: str
478class FullEvent(SummaryEvent):
479 model_config = ConfigDict(from_attributes=True)
481 id: int
482 event_class: str
483 question_id: Optional[int]
484 project_id: Optional[int]
485 issue_id: Optional[int]
486 org_id: str
487 object_id: Optional[Union[str, int]]
488 changes: List[dict]
491class EvIssue(BaseModel):
492 respondent_id: str
493 label: Optional[str]
496class AuditEvent(FullEvent):
497 issue: Optional[EvIssue] = None
498 user: Optional[User]
499 question_number: Optional[str] = None
500 section_id: Optional[int] = None
501 project_title: Optional[str] = None
504class AnsweredQElement(QElement):
505 answer: Annotated[str, StringConstraints(max_length=65536)]
508class ElementAnswer(BaseModel):
509 element_id: int = Field(
510 ..., description="ID of the question Element for this answer"
511 )
512 answer: Annotated[str, StringConstraints(max_length=65536)]
515class ElementAnswerList(RootModel):
516 root: Annotated[List[ElementAnswer], Field(min_length=0, max_length=100)]
519class Answer(ElementAnswer):
520 issue_id: int = Field(..., description="ID of the associated Issue for this answer")
521 question_id: int
524ldoc = "Mapping of Issue ID to Question Element ID to answer"
527class AnswerLookup(RootModel):
528 model_config = ConfigDict(
529 json_schema_extra={
530 "example": {
531 "523": {
532 "234872": "Answer for Issue 523, Element 234872",
533 "234875": "Answer for Issue 523, Element 234875",
534 },
535 "529": {
536 "234872": "Answer for Issue 529, Element 234872",
537 "234875": "Answer for Issue 529, Element 234875",
538 },
539 }
540 }
541 )
542 root: dict[str, dict[str, str]] = Field(..., description=ldoc)
545class AnswerResponseState(BaseModel):
546 status: str
547 allocated_by: Optional[str]
548 allocated_to: Optional[str]
549 approved_by: Optional[str]
550 updated_by: Optional[str]
551 date_updated: Optional[datetime]
552 question_instance_id: int
555class AllocatedTo(BaseModel):
556 allocated_to: str
557 question_instance_id: int
560class AllocatedToList(RootModel):
561 root: List[AllocatedTo]
563 def __iter__(self):
564 return iter(self.root)
567class AnswerStats(BaseModel):
568 model_config = ConfigDict(from_attributes=True)
569 allocated_to: Optional[str] = None
570 status: str
571 question_count: int
574class ImportableAnswers(BaseModel):
575 model_config = ConfigDict(from_attributes=True)
576 title: str
577 issue_date: Optional[datetime]
578 submitted_date: Optional[datetime]
579 question_count: int
580 issue_id: int
583class ImportableAnswersList(RootModel):
584 root: List[ImportableAnswers]
587class AnsweredQuestion(Question):
588 response_state: List[AnswerResponseState]
589 elements: ElGrid
592class SingleRespondentQuestion(Question):
593 elements: ElGrid
594 respondent: Organisation
597class Node(Id):
598 model_config = ConfigDict(from_attributes=True)
599 number: str
600 title: str
603node_pos_docs = """This string gives the position of the node within it's questionnaire
604tree structure. Each 2 character segment is a base 36 encoded representation of a single element
605in a dotted position number. E.g. '0N0108' -> 23.1.18"""
608class NodeTypeEnum(str, Enum):
609 section = "section"
610 question = "question"
613class ProjectNode(Node):
614 type: NodeTypeEnum
615 parent_id: Optional[int] = Field(
616 ..., description="ID of this node's parent section"
617 )
618 node_position: str = Field(..., description=node_pos_docs)
619 depth: int = Field(
620 ..., description="The nested depth of this node within the questionnaire"
621 )
622 description: Optional[str] = None
625class QI(BaseModel):
626 model_config = ConfigDict(from_attributes=True)
627 id: int
628 project_id: int
629 number: str
632class QuestionInstance(QI):
633 section_id: int
634 project_title: Optional[str] = None
637class ScoreGaps(BaseModel):
638 question_id: int
639 number: str
640 title: str
641 node_type: str
642 score_gap: float
643 weight: float
646class EditableSection(BaseModel):
647 title: Annotated[str, StringConstraints(max_length=255)]
648 description: Optional[str] = Field(None, max_length=1024)
651class Section(EditableSection):
652 model_config = ConfigDict(from_attributes=True)
654 id: Optional[PositiveInt] = None
655 parent_id: Optional[PositiveInt] = Field(...)
656 number: Optional[str] = None
658 @field_validator("number")
659 @classmethod
660 def clean_number(cls, number):
661 return getattr(number, "dotted", number)
664class FullSection(Section):
665 id: int
666 subsections: List[TreeNode]
667 questions: List[Question]
670class SummarySection(Section):
671 model_config = ConfigDict(from_attributes=True)
672 questions: List[Node] = Field(default_factory=list)
673 subsections: List[Node] = Field(default_factory=list)
676class ParentId(BaseModel):
677 new_parent_id: CONSTRAINED_ID
680class MoveSection(ParentId):
681 section_id: CONSTRAINED_ID
684class SectionChildNodes(BaseModel):
685 question_ids: Annotated[List[CONSTRAINED_ID], Field(max_length=100)] = Field(
686 default_factory=list
687 )
688 section_ids: Annotated[List[CONSTRAINED_ID], Field(max_length=100)] = Field(
689 default_factory=list
690 )
692 delete_orphans: bool = False
694 @model_validator(mode="before")
695 def either_or(cls, values):
696 sec_ids, qids = (
697 values.get("section_ids", None),
698 values.get("question_ids", None),
699 )
700 if sec_ids is None and qids is None:
701 raise ValueError("Either section_ids or question_ids must be set")
702 if (
703 sec_ids is not None
704 and qids is not None
705 and len(sec_ids) > 0
706 and len(qids) > 0
707 ):
708 raise ValueError(
709 "Cannot assign values to both question_ids and section_ids"
710 )
711 return values
714class WorkflowSection(BaseModel):
715 section: Section
716 questions: List[AnswerResponseState]
717 count: int
720class Nodes(Node):
721 type: str
722 subsections: Optional[list["Nodes"]] = None
723 questions: Optional[list["Nodes"]] = None
726class NodesList(RootModel):
727 root: List[Nodes]
730class QElementStats(BaseModel):
731 TX: int = Field(..., description="Count of Text Input elements")
732 CR: int = Field(..., description="Count of Radio Choices elements")
733 CC: int = Field(..., description="Count of Combon/Select elements")
734 CB: int = Field(..., description="Count of Checkbox elements")
735 LB: int = Field(..., description="Count of Label elements")
736 QA: int = Field(..., description="Count of Question Attachment elements")
737 AT: int = Field(..., description="Count of Upload Field Attachment elements")
738 MD: int = Field(..., description="Count of Media elements")
739 answerable_elements: int = Field(
740 ..., description="Total number of answerable elements"
741 )
744class QuestionnaireStats(BaseModel):
745 questions: int = Field(..., description="Count of Questions in the Project")
746 sections: int = Field(..., description="Count of Sections in the Project")
747 elements: QElementStats = Field(..., description="Statistics for Question Elements")
750class ProjectField(BaseModel):
751 """
752 ProjectFields allow the user to associate custom data fields with their project.
754 By default such fields are invisible to respondent users - behaviour dictated by the 'private'
755 property. ProjectFields are ordered when displayed according to the 'position' property.
756 """
758 model_config = ConfigDict(from_attributes=True)
760 key: Annotated[str, StringConstraints(min_length=1, max_length=64)]
761 value: Annotated[str, StringConstraints(max_length=8192)]
762 private: bool = Field(
763 True, description="Indicates whether this field is visible to Respondents"
764 )
767class NewCategory(BaseModel):
768 """Represents a category which can be assigned to a Project for classification and filtering"""
770 name: Annotated[str, StringConstraints(max_length=50, min_length=3)]
771 description: Annotated[str, StringConstraints(max_length=250)]
774class Category(NewCategory):
775 model_config = ConfigDict(from_attributes=True)
777 id: Optional[int] = None
780class ProjTypeEnum(str, Enum):
781 NORMAL = "NORMAL"
782 REFERENCE = "REFERENCE"
785class UpdateableProject(BaseModel):
786 model_config = ConfigDict(from_attributes=True, populate_by_name=True)
788 title: str = Field("", min_length=5, max_length=256)
789 description: Optional[str] = Field(None, max_length=20000)
790 deadline: Optional[datetime] = None
791 public: bool = Field(
792 False,
793 description="List under public projects and allow uninvited respondents to self-invite",
794 )
795 multiscored: bool = Field(
796 False,
797 description="Indicates whether Multiple Score Sets are enabled for this project",
798 )
799 allow_private_communication: bool = Field(
800 False,
801 description="Allow Notes to be posted between the Project owner an individual Respondents",
802 )
803 expose_response_attachments: bool = False
804 expose_weightings: bool = Field(
805 False,
806 description="Allow Respondents to view weightings assigned to questions and sections",
807 )
808 require_acceptance: bool = Field(
809 True,
810 description="Require that each Respondent must formally accept their invitation to respond",
811 )
812 maximum_score: Annotated[int, Field(gt=0, lt=2147483648)] = 10
813 reminder: Optional[int] = Field(
814 None,
815 alias="deadline_reminder_hours",
816 description=(
817 "Number of hours before the Issue deadline when reminder emails should be "
818 "sent to respondents"
819 ),
820 )
821 email: Optional[EmailStr] = Field(
822 None,
823 alias="from_email_address",
824 description=(
825 "Send notification emails pertaining to the current project from this email "
826 "address"
827 ),
828 )
829 hide_responses: bool = Field(
830 False,
831 alias="hide_responses_until_deadline",
832 description="Hide answers to questions until the Project deadline has passed",
833 )
834 project_type: ProjTypeEnum = ProjTypeEnum.NORMAL
835 project_fields: List[ProjectField] = []
838class NewProject(UpdateableProject):
839 # title duplicates that of UpdateableProject but here it is a required field
840 title: Annotated[
841 str, StringConstraints(strip_whitespace=True, min_length=5, max_length=256)
842 ]
843 questionnaire_title: Annotated[
844 str, StringConstraints(strip_whitespace=True, min_length=5, max_length=64)
845 ] = "Questionnaire"
846 category_ids: List[int] = []
849class FullProject(UpdateableProject):
850 model_config = ConfigDict(from_attributes=True)
852 id: int
853 status: str
854 author_id: str
855 owner_org: Optional[Organisation] = Field(None, alias="owner_organisation")
856 date_published: Optional[datetime] = None
857 date_created: datetime
858 permissions: List[str] = []
859 categories: List[Category]
860 section_id: Optional[int] = None
862 def set_permissions(self, user):
863 self.permissions = user.sorted_permissions
865 @field_validator("permissions", mode="before")
866 def perms(cls, v):
867 return []
870class ListProject(BaseModel):
871 model_config = ConfigDict(from_attributes=True)
872 id: int
873 title: str
874 owner_org_name: str
875 owner_org_id: str
876 status: str
877 deadline: Optional[datetime] = None
878 date_created: datetime
879 watching_since: Optional[datetime] = None
880 is_watched: bool = False
883class ProjectList(BaseModel):
884 data: List[ListProject]
885 pagination: Pagination
888class Supplier(BaseModel):
889 issues: List[Issue]
890 organisation: Organisation
891 users: List[User]
894class AnswerAttachment(Id):
895 model_config = ConfigDict(from_attributes=True)
896 filename: str
897 mimetype: str
898 size: str
899 size_bytes: int
902class Attachment(AnswerAttachment):
903 model_config = ConfigDict(from_attributes=True)
904 private: bool
905 description: Annotated[str, StringConstraints(min_length=1, max_length=255)]
906 date_uploaded: datetime
907 author_id: str
908 org_id: str
911class IssueAttachment(Attachment):
912 model_config = ConfigDict(from_attributes=True)
913 issue_id: int
914 description: Annotated[str, StringConstraints(min_length=1, max_length=1024)]
917class Watcher(BaseModel):
918 user_id: str
919 fullname: str
920 email: Optional[EmailStr] = None
921 watching_since: Optional[datetime] = None
922 is_watching: bool = False
924 @model_validator(mode="after")
925 def set_is_watching(self, values) -> Self:
926 self.is_watching = self.watching_since is not None
927 return self
929 @field_validator("email", mode="before")
930 @classmethod
931 def none_if_zerolength(cls, email):
932 if type(email) is str and len(email.strip()) == 0:
933 return None
934 return email
936 model_config = ConfigDict(from_attributes=True)
939class IssueWatchList(BaseModel):
940 issue_id: int
941 watchlist: List[Watcher]
944class AnswerImportResult(BaseModel):
945 imported: List[str]
946 errors: List[str]
947 unchanged: List[str]
950class ImportAnswers(BaseModel):
951 source_issue_id: int
952 section_number: str
955class SectionImportDoc(BaseModel):
956 project_id: int
957 section_ids: List[int] = []
958 question_ids: List[int] = []
959 clone: bool = True
962class SectionImportResult(BaseModel):
963 section_count: int = 0
964 question_count: int = 0
967class TextReplace(BaseModel):
968 search_term: str
969 replace_term: str
970 dry_run: bool = True
973class HitTypes(str, Enum):
974 questions = "questions"
975 choices = "choices"
976 answers = "answers"
977 notes = "notes"
978 scoreComments = "scoreComments"
981class SearchResult(BaseModel):
982 klass: HitTypes
983 project_title: str
984 project_id: int
985 object_id: Union[str, int]
986 object_ref: Union[str, None] = None
987 snippet: str
990class RelationshipType(BaseModel):
991 model_config = ConfigDict(from_attributes=True)
993 id: Optional[int] = Field(None, gt=0, lt=2147483648)
994 name: Annotated[str, StringConstraints(min_length=2, max_length=128)]
995 description: Optional[str] = Field(None, max_length=256)
998class Relationship(BaseModel):
999 reltype_id: Annotated[int, Field(gt=0, lt=2147483648)]
1000 from_org_id: Annotated[str, StringConstraints(min_length=2, max_length=50)]
1001 to_org_id: Annotated[str, StringConstraints(min_length=2, max_length=50)]
1004class NetworkRelationship(BaseModel):
1005 model_config = ConfigDict(from_attributes=True, populate_by_name=True)
1007 # relationship: str
1008 relationship: RelationshipType = Field(..., alias="relationship_type")
1009 from_org: BaseOrganisation
1010 to_org: BaseOrganisation
1013class NewTag(BaseModel):
1014 """
1015 A Tag is a keyword used to categorize questions
1016 """
1018 name: Annotated[str, StringConstraints(min_length=2, max_length=128)]
1019 description: Optional[Annotated[str, StringConstraints(max_length=256)]] = None
1022class Tag(NewTag):
1023 model_config = ConfigDict(from_attributes=True)
1025 id: Optional[Annotated[int, Field(gt=0, lt=2147483648)]] = None
1028class TagAssigns(BaseModel):
1029 question_instance_ids: CONSTRAINED_ID_LIST
1030 section_ids: CONSTRAINED_ID_LIST
1031 recursive: bool = False
1034class TagGroup(TagAssigns):
1035 tag_id: Annotated[int, Field(gt=0, lt=2147483648)]
1038class MatchedElement(BaseModel):
1039 title: str
1040 number: str
1041 el_type: str
1042 label: str
1043 choices: List[Choice]
1046class QSearchResult(BaseModel):
1047 distinct_question_count: int = Field(
1048 ..., description="Number of questions that match the query"
1049 )
1050 matched_elements: List[MatchedElement]
1053class ReplacedItem(BaseModel):
1054 change_type: str
1055 question_number: str
1056 old: str
1057 new: str
1060class PublishProject(BaseModel):
1061 release_issue_ids: Annotated[
1062 List[Annotated[int, Field(gt=0, lt=2147483648)]],
1063 Field(min_length=0, max_length=50),
1064 ]
1067class PublishResult(BaseModel):
1068 issue_id: int
1069 respondent_id: str
1072class NewWebhook(BaseModel):
1073 org_id: Annotated[str, StringConstraints(max_length=50)]
1074 event_type: Annotated[str, StringConstraints(max_length=50)]
1075 remote_url: AnyHttpUrl = Field(..., alias="url")
1076 http_header: Optional[str] = Field(
1077 None,
1078 description=(
1079 "A custom HTTP Header to add the webhook post request. "
1080 " Format as a Header:Value string"
1081 ),
1082 max_length=1024,
1083 )
1085 @field_validator("event_type")
1086 @classmethod
1087 def validate_event_type(cls, event_type):
1088 from rfpy.model.audit import evt_types
1090 if not hasattr(evt_types, event_type):
1091 raise ValueError(f"{event_type} is not a valid Event Type")
1092 return event_type
1094 @field_validator("http_header")
1095 @classmethod
1096 def validate_http_header(cls, http_header):
1097 if not http_header:
1098 return None
1099 http_header = http_header.strip()
1100 if re.match(r"^[A-Za-z_\-]+\s*:\s*([A-Za-z0-9_\-]+)$", http_header) is None:
1101 raise ValueError("http_header must be of the form XXX:YYY")
1102 return http_header
1105class Webhook(NewWebhook):
1106 model_config = ConfigDict(
1107 from_attributes=True, populate_by_name=True, use_enum_values=True
1108 )
1110 delivery_status: Optional[str] = None
1111 last_delivery: Optional[datetime] = None
1112 error_message: Optional[str] = None
1113 retries: int = 0