from __future__ import annotations from datetime import datetime from typing import Any, Literal from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, SecretStr, field_validator, model_validator SchedulingStatus = Literal["draft", "collecting", "closed", "decided", "handed_off", "cancelled", "archived"] SchedulingParticipantType = Literal["internal", "external", "resource"] SchedulingParticipantStatus = Literal["draft", "invited", "responded", "declined", "removed"] SchedulingResultVisibility = Literal["organizer", "after_response", "after_close", "public"] SchedulingAvailabilityValue = Literal["available", "maybe", "unavailable"] SchedulingParticipantVisibility = Literal["aggregates_only", "names_and_statuses"] def _known_timezone(value: str | None) -> str | None: if value is None: return None try: ZoneInfo(value) except ZoneInfoNotFoundError as exc: raise ValueError("timezone must be a valid IANA timezone") from exc return value def _participant_email(value: str | None) -> str | None: if value is None: return None normalized = value.strip().casefold() if not normalized: return None local, separator, domain = normalized.partition("@") if ( separator != "@" or not local or not domain or "@" in domain or any(character.isspace() for character in normalized) ): raise ValueError("participant_email must be a valid email address") return normalized class SchedulingCalendarPreferences(BaseModel): model_config = ConfigDict(extra="forbid") enabled: bool = False calendar_id: str | None = Field(default=None, max_length=36) freebusy_enabled: bool = False tentative_holds_enabled: bool = False create_event_on_decision: bool = False class SchedulingCandidateSlotInput(BaseModel): model_config = ConfigDict(extra="forbid") label: str | None = Field(default=None, max_length=500) description: str | None = None start_at: AwareDatetime end_at: AwareDatetime timezone: str | None = Field(default=None, max_length=100) location: str | None = Field(default=None, max_length=500) metadata: dict[str, Any] = Field(default_factory=dict) _validate_timezone = field_validator("timezone")(_known_timezone) @model_validator(mode="after") def validate_range(self) -> "SchedulingCandidateSlotInput": if self.end_at <= self.start_at: raise ValueError("end_at must be after start_at") return self class SchedulingCandidateSlotUpdateRequest(BaseModel): """Partial update of one candidate slot. A semantic option change invalidates existing answers for this slot in the backing Poll. Exact repeats are safe no-ops. """ model_config = ConfigDict(extra="forbid") label: str | None = Field(default=None, min_length=1, max_length=500) description: str | None = None start_at: AwareDatetime | None = None end_at: AwareDatetime | None = None timezone: str | None = Field(default=None, min_length=1, max_length=100) location: str | None = Field(default=None, max_length=500) metadata: dict[str, Any] | None = None _validate_timezone = field_validator("timezone")(_known_timezone) @model_validator(mode="after") def validate_range(self) -> "SchedulingCandidateSlotUpdateRequest": if self.start_at is not None and self.end_at is not None and self.end_at <= self.start_at: raise ValueError("end_at must be after start_at") return self class SchedulingParticipantInput(BaseModel): model_config = ConfigDict(extra="forbid") respondent_id: str | None = Field(default=None, max_length=255) display_name: str | None = Field(default=None, max_length=500) email: str | None = Field(default=None, max_length=320) participant_type: SchedulingParticipantType = "external" required: bool = True metadata: dict[str, Any] = Field(default_factory=dict) _validate_email = field_validator("email")(_participant_email) class SchedulingCandidateSlotReconcileInput(SchedulingCandidateSlotInput): id: str | None = Field(default=None, max_length=36) revision: str | None = Field( default=None, min_length=64, max_length=64, pattern=r"^[0-9a-f]{64}$", ) @model_validator(mode="after") def validate_existing_revision(self) -> "SchedulingCandidateSlotReconcileInput": if self.id is not None and self.revision is None: raise ValueError("revision is required for an existing scheduling slot") if self.id is None and self.revision is not None: raise ValueError("revision can only be supplied for an existing scheduling slot") return self class SchedulingParticipantReconcileInput(SchedulingParticipantInput): id: str | None = Field(default=None, max_length=36) revision: str | None = Field( default=None, min_length=64, max_length=64, pattern=r"^[0-9a-f]{64}$", ) @model_validator(mode="after") def validate_existing_revision(self) -> "SchedulingParticipantReconcileInput": if self.id is not None and self.revision is None: raise ValueError("revision is required for an existing scheduling participant") if self.id is None and self.revision is not None: raise ValueError("revision can only be supplied for an existing scheduling participant") return self class SchedulingRequestCreateRequest(BaseModel): model_config = ConfigDict(extra="forbid") title: str = Field(min_length=1, max_length=500) description: str | None = None location: str | None = Field(default=None, max_length=500) timezone: str = Field(default="UTC", max_length=100) status: Literal["draft", "collecting"] = "draft" deadline_at: datetime | None = None allow_external_participants: bool = True allow_participant_updates: bool = True result_visibility: SchedulingResultVisibility = "after_close" participant_visibility: SchedulingParticipantVisibility = "aggregates_only" notify_on_answers: bool = True single_choice: bool = False max_participants_per_option: int | None = Field(default=None, ge=1) allow_maybe: bool = True allow_comments: bool = False participant_email_required: bool = False anonymous_password_protection_enabled: bool = False anonymous_password: SecretStr | None = Field(default=None, min_length=8, max_length=1024) calendar: SchedulingCalendarPreferences = Field(default_factory=SchedulingCalendarPreferences) slots: list[SchedulingCandidateSlotInput] = Field(default_factory=list, min_length=1) participants: list[SchedulingParticipantInput] = Field(default_factory=list) create_participant_invitations: bool = Field( default=False, deprecated=True, description=( "Compatibility field; participant links are issued only through " "the explicit participant invitation action." ), ) metadata: dict[str, Any] = Field(default_factory=dict) _validate_timezone = field_validator("timezone")(_known_timezone) @model_validator(mode="after") def validate_anonymous_password(self) -> "SchedulingRequestCreateRequest": if self.anonymous_password_protection_enabled and self.anonymous_password is None: raise ValueError("anonymous_password is required when password protection is enabled") if not self.anonymous_password_protection_enabled and self.anonymous_password is not None: raise ValueError("anonymous_password requires password protection to be enabled") return self class SchedulingRequestUpdateRequest(BaseModel): model_config = ConfigDict(extra="forbid") title: str | None = Field(default=None, min_length=1, max_length=500) description: str | None = None location: str | None = Field(default=None, max_length=500) deadline_at: datetime | None = None allow_external_participants: bool | None = None allow_participant_updates: bool | None = None result_visibility: SchedulingResultVisibility | None = None participant_visibility: SchedulingParticipantVisibility | None = None notify_on_answers: bool | None = None single_choice: bool | None = None max_participants_per_option: int | None = Field(default=None, ge=1) allow_maybe: bool | None = None allow_comments: bool | None = None participant_email_required: bool | None = None anonymous_password_protection_enabled: bool | None = None anonymous_password: SecretStr | None = Field(default=None, min_length=8, max_length=1024) calendar: SchedulingCalendarPreferences | None = None slots: list[SchedulingCandidateSlotReconcileInput] | None = Field( default=None, min_length=1, ) participants: list[SchedulingParticipantReconcileInput] | None = None create_participant_invitations: bool = Field( default=False, deprecated=True, description=( "Compatibility field; participant links are issued only through " "the explicit participant invitation action." ), ) metadata: dict[str, Any] | None = None @model_validator(mode="after") def validate_anonymous_password(self) -> "SchedulingRequestUpdateRequest": if self.anonymous_password_protection_enabled is False and self.anonymous_password is not None: raise ValueError("anonymous_password cannot be set while password protection is disabled") return self @model_validator(mode="after") def validate_reconciliation_ids(self) -> "SchedulingRequestUpdateRequest": for field_name in ("slots", "participants"): values = getattr(self, field_name) if values is None: continue ids = [value.id for value in values if value.id is not None] if len(ids) != len(set(ids)): raise ValueError(f"Duplicate ids are not allowed in {field_name}") return self class SchedulingCandidateSlotResponse(BaseModel): id: str poll_option_id: str | None = None label: str description: str | None = None start_at: datetime end_at: datetime timezone: str location: str | None = None position: int revision: str freebusy_checked_at: datetime | None = None freebusy_status: str | None = None freebusy_conflicts: list[dict[str, Any]] = Field(default_factory=list) tentative_hold_event_id: str | None = None metadata: dict[str, Any] = Field(default_factory=dict) class SchedulingParticipantResponse(BaseModel): id: str revision: str | None = None is_current_participant: bool = False respondent_id: str | None = None display_name: str | None = None email: str | None = None participant_type: str | None = None required: bool | None = None status: str poll_invitation_id: str | None = None invitation_token: str | None = None last_invited_at: datetime | None = None responded_at: datetime | None = None metadata: dict[str, Any] = Field(default_factory=dict) class SchedulingParticipantAggregateResponse(BaseModel): total: int = 0 status_counts: dict[str, int] = Field(default_factory=dict) class SchedulingParticipantVisibilityDecisionResponse(BaseModel): requested_visibility: SchedulingParticipantVisibility effective_visibility: SchedulingParticipantVisibility policy_applied: bool = False reason: str | None = None source_path: list[dict[str, Any]] = Field(default_factory=list) details: dict[str, Any] = Field(default_factory=dict) class SchedulingRequestResponse(BaseModel): id: str tenant_id: str | None = None title: str description: str | None = None location: str | None = None timezone: str status: str poll_id: str | None = None selected_slot_id: str | None = None organizer_user_id: str | None = None deadline_at: datetime | None = None allow_external_participants: bool allow_participant_updates: bool result_visibility: str participant_visibility: SchedulingParticipantVisibility notify_on_answers: bool single_choice: bool max_participants_per_option: int | None = None allow_maybe: bool allow_comments: bool participant_email_required: bool anonymous_password_protection_enabled: bool public_participation_policy_enforcement_available: bool | None = None public_participation_policy_enforcement_reason: str | None = None participant_invitation_delivery_available: bool | None = None effective_participant_visibility: SchedulingParticipantVisibility participant_aggregate: SchedulingParticipantAggregateResponse participant_visibility_decision: SchedulingParticipantVisibilityDecisionResponse calendar_integration_enabled: bool | None = None calendar_id: str | None = None calendar_freebusy_enabled: bool | None = None calendar_hold_enabled: bool | None = None create_calendar_event_on_decision: bool | None = None calendar_event_id: str | None = None handed_off_at: datetime | None = None cancelled_at: datetime | None = None cancellation_notice_until: datetime | None = None created_at: datetime updated_at: datetime metadata: dict[str, Any] = Field(default_factory=dict) slots: list[SchedulingCandidateSlotResponse] = Field(default_factory=list) participants: list[SchedulingParticipantResponse] = Field(default_factory=list) class SchedulingRequestListResponse(BaseModel): requests: list[SchedulingRequestResponse] = Field(default_factory=list) class SchedulingStatusResponse(BaseModel): request: SchedulingRequestResponse class SchedulingDecisionRequest(BaseModel): model_config = ConfigDict(extra="forbid") slot_id: str | None = None poll_option_id: str | None = None handoff_to_calendar: bool | None = None class SchedulingAvailabilityAnswerInput(BaseModel): model_config = ConfigDict(extra="forbid") slot_id: str value: SchedulingAvailabilityValue option_revision: str = Field(min_length=64, max_length=64, pattern=r"^[0-9a-f]{64}$") class SchedulingAvailabilityResponseRequest(BaseModel): model_config = ConfigDict(extra="forbid") answers: list[SchedulingAvailabilityAnswerInput] = Field(min_length=1) comment: str | None = Field(default=None, max_length=4000) @model_validator(mode="after") def validate_unique_slots(self) -> "SchedulingAvailabilityResponseRequest": slot_ids = [answer.slot_id for answer in self.answers] if len(slot_ids) != len(set(slot_ids)): raise ValueError("Each scheduling slot can be answered only once") return self class SchedulingAvailabilityAnswerResponse(BaseModel): slot_id: str value: SchedulingAvailabilityValue class SchedulingAvailabilityResponse(BaseModel): request_id: str participant_id: str has_response: bool = False submitted_at: datetime | None = None answers: list[SchedulingAvailabilityAnswerResponse] = Field(default_factory=list) comment: str | None = None class SchedulingPublicParticipationAccessRequest(BaseModel): model_config = ConfigDict(extra="forbid") participant_email: str | None = Field(default=None, max_length=320) password: SecretStr | None = Field(default=None, max_length=1024) _validate_participant_email = field_validator("participant_email")(_participant_email) class SchedulingPublicParticipationSubmitRequest(BaseModel): model_config = ConfigDict(extra="forbid") answers: list[SchedulingAvailabilityAnswerInput] = Field(min_length=1) participant_email: str | None = Field(default=None, max_length=320) password: SecretStr | None = Field(default=None, max_length=1024) comment: str | None = Field(default=None, max_length=4000) idempotency_key: str | None = Field(default=None, min_length=1, max_length=255) _validate_participant_email = field_validator("participant_email")(_participant_email) @model_validator(mode="after") def validate_unique_slots(self) -> "SchedulingPublicParticipationSubmitRequest": slot_ids = [answer.slot_id for answer in self.answers] if len(slot_ids) != len(set(slot_ids)): raise ValueError("Each scheduling slot can be answered only once") return self class SchedulingPublicCandidateSlotResponse(BaseModel): id: str label: str description: str | None = None start_at: datetime end_at: datetime timezone: str location: str | None = None position: int revision: str class SchedulingPublicParticipationResponse(BaseModel): request_id: str title: str description: str | None = None location: str | None = None timezone: str status: str deadline_at: datetime | None = None cancelled_at: datetime | None = None cancellation_notice_until: datetime | None = None cancellation_notice_only: bool = False participant_email_required: bool anonymous_password_required: bool single_choice: bool max_participants_per_option: int | None = None allow_maybe: bool allow_comments: bool allow_participant_updates: bool has_response: bool = False submitted_at: datetime | None = None answers: list[SchedulingAvailabilityAnswerResponse] = Field(default_factory=list) comment: str | None = None replayed: bool = False slots: list[SchedulingPublicCandidateSlotResponse] = Field(default_factory=list) class SchedulingPollOptionResultResponse(BaseModel): option_id: str option_key: str label: str count: int = 0 score: int = 0 values: dict[str, int] = Field(default_factory=dict) ranks: dict[int, int] = Field(default_factory=dict) class SchedulingPollSummaryResponse(BaseModel): poll_id: str kind: str status: str response_count: int option_results: list[SchedulingPollOptionResultResponse] = Field(default_factory=list) leading_option_ids: list[str] = Field(default_factory=list) class SchedulingSummaryResponse(BaseModel): request: SchedulingRequestResponse poll_summary: SchedulingPollSummaryResponse class SchedulingCalendarActionResponse(BaseModel): request: SchedulingRequestResponse created_event_ids: list[str] = Field(default_factory=list) updated_slot_ids: list[str] = Field(default_factory=list) warnings: list[str] = Field(default_factory=list) class SchedulingNotificationResponse(BaseModel): id: str request_id: str participant_id: str | None = None event_kind: str channel: str recipient: str | None = None status: str payload: dict[str, Any] = Field(default_factory=dict) error: str | None = None sent_at: datetime | None = None created_at: datetime updated_at: datetime metadata: dict[str, Any] = Field(default_factory=dict) class SchedulingNotificationListResponse(BaseModel): notifications: list[SchedulingNotificationResponse] = Field(default_factory=list) class SchedulingNotificationCreateRequest(BaseModel): model_config = ConfigDict(extra="forbid") event_kind: Literal["invitation", "reminder", "decision", "cancellation"] channel: str = Field(default="mail", max_length=40) metadata: dict[str, Any] = Field(default_factory=dict) class SchedulingInvitationActionRequest(BaseModel): """Explicitly issue one fresh participant-specific participation link.""" model_config = ConfigDict(extra="forbid") action: Literal["copy", "send"] class SchedulingInvitationActionResponse(BaseModel): participant_id: str action: Literal["copy", "send", "revoke"] status: str action_url: str | None = None issued_at: datetime | None = None replayed: bool = False notification: SchedulingNotificationResponse | None = None class SchedulingPeopleSearchCandidate(BaseModel): """Opaque, task-safe projection of a visible directory candidate.""" selection_key: str kind: str reference_id: str display_name: str email: str | None = None source_module: str | None = None source_label: str | None = None source_revision: str | None = None description: str | None = None class SchedulingPeopleSearchGroup(BaseModel): key: str label: str candidates: list[SchedulingPeopleSearchCandidate] = Field(default_factory=list) class SchedulingPeopleSearchResponse(BaseModel): groups: list[SchedulingPeopleSearchGroup] = Field(default_factory=list)