from __future__ import annotations import datetime from collections.abc import Sequence from typing import Any from pydantic import AwareDatetime, Base64Str, BaseModel, ConfigDict, EmailStr, Field from email_campaigns.audiences.enums import AudienceTarget from email_campaigns.automated.enums import AutomatedEmailType, SourceCampaignStatus from email_campaigns.automated.types import AutomatedEmailOrderBy, SourceCampaignOrderBy from email_campaigns.campaigns.enums import ( EmailCampaignCancelReason, EmailCampaignStatus, ) from email_campaigns.campaigns.types import CampaignSearchContext, EmailCampaignOrderBy from email_campaigns.rosters.types import FanDataListId # Always populate by name BaseModel.model_config["populate_by_name"] = True # Common class LimitOffsetPage[T](BaseModel): total: int items: Sequence[T] limit: int offset: int @classmethod def __concrete_name__(cls: type[Any], params: tuple[type[Any], ...]) -> str: return f"{params[0].__name__.title()}LimitOffsetPage" # Infra class HelloResponse(BaseModel): status: str = "ok" # Fonts class GoogleFont(BaseModel): name: str fallback_font: str | None = Field(default=None, serialization_alias="fallbackFont") generic_font_family: str = Field(serialization_alias="genericFontFamily") css_font_family: str = Field(serialization_alias="cssFontFamily") url: str # Audiences class Audience(BaseModel): id: str name: str fan_count: int = Field(alias="fanCount") is_archived: bool = Field(alias="isArchived") target: AudienceTarget # Emails class EmailDomain(BaseModel): id: str domain: str brand: str class CustomFont(BaseModel): model_config = ConfigDict(populate_by_name=True) name: str css_font_family: str = Field(alias="cssFontFamily") url: str class EmailPreview(BaseModel): id: str name: str subject: str | None preview_text: str | None = Field(alias="previewText") sender_name: str | None = Field(alias="senderName") html: str | None css: str | None custom_fonts: list[CustomFont] | None = Field(default=None, alias="customFonts") favorite_fonts: list[CustomFont] | None = Field(default=None, alias="favoriteFonts") class EmailPrivacyFooter(BaseModel): content: str class GenerateEmailPublicPreviewUrlInput(BaseModel): domain: str class GenerateEmailPublicPreviewUrlResponse(BaseModel): url: str class EmailAnalytics(BaseModel): sends: int = Field(alias="sends") delivered: int = Field(alias="delivered") opens: int = Field(alias="opens") clicks: int = Field(alias="clicks") unique_opens: int = Field(alias="uniqueOpens") unique_clicks: int = Field(alias="uniqueClicks") bounces: int = Field(alias="bounces") spam_complaints: int = Field(alias="spamComplaints") unsubscribes: int = Field(alias="unsubscribes") open_rate: float = Field(alias="openRate") delivered_rate: float = Field(alias="deliveredRate") click_through_rate: float = Field(alias="clickThroughRate") click_to_open_rate: float = Field(alias="clickToOpenRate") bounce_rate: float = Field(alias="bounceRate") spam_complaints_rate: float = Field(alias="spamComplaintsRate") unsubscribe_rate: float = Field(alias="unsubscribeRate") unique_clicks_dropoff: float | None = Field(alias="uniqueClicksDropoff") unique_opens_dropoff: float | None = Field(alias="uniqueOpensDropoff") bounce_rate_change: float | None = Field(alias="bounceRateChange") spam_complaints_rate_change: float | None = Field(alias="spamComplaintsRateChange") unsubscribe_rate_change: float | None = Field(alias="unsubscribeRateChange") open_rate_change: float | None = Field(alias="openRateChange") delivered_rate_change: float | None = Field(alias="deliveredRateChange") unique_clicks_change: float | None = Field(alias="uniqueClicksChange") click_through_rate_change: float | None = Field(alias="clickThroughRateChange") click_to_open_rate_change: float | None = Field(alias="clickToOpenRateChange") class GetEmailAnalyticsResponse(BaseModel): data: EmailAnalytics | None class EmailLinkAnalytics(BaseModel): url: str = Field(alias="url") total_clicks: int = Field(alias="totalClicks") unique_clicks: int = Field(alias="uniqueClicks") percentage_of_total_clicks: float = Field(alias="percentageOfTotalClicks") class GetEmailLinkAnalyticsResponse(BaseModel): data: list[EmailLinkAnalytics] | None class EmailSalesAnalytics(BaseModel): orders_count: int = Field(alias="ordersCount") items_sold: int = Field(alias="itemsSold") net_sales_usd: float = Field(alias="netSalesUsd") gross_sales_usd: float = Field(alias="grossSalesUsd") discounts_usd: float = Field(alias="discountsUsd") taxes_usd: float = Field(alias="taxesUsd") shipping_usd: float = Field(alias="shippingUsd") refunds_usd: float = Field(alias="refundsUsd") average_units_per_order: float = Field(alias="averageUnitsPerOrder") average_order_amount: float = Field(alias="averageOrderAmount") first_time_buyer_rate: float = Field(alias="firstTimeBuyerRate") class GetEmailSalesAnalyticsResponse(BaseModel): data: EmailSalesAnalytics class EmailSalesProductAnalytics(BaseModel): product_title: str | None = Field(alias="productTitle") product_type: str | None = Field(alias="productType") items_sold: int = Field(alias="itemsSold") net_sales_usd: float = Field(alias="netSalesUsd") chart_eligible: bool = Field(alias="chartEligible") class GetEmailSalesProductsAnalyticsResponse(BaseModel): data: list[EmailSalesProductAnalytics] class EmailSalesLocationAnalytics(BaseModel): country_code: str | None = Field(alias="countryCode") items_sold: int = Field(alias="itemsSold") net_sales_usd: float = Field(alias="netSalesUsd") sales_rate: float = Field(alias="salesRate") class GetEmailSalesLocationsAnalyticsResponse(BaseModel): data: list[EmailSalesLocationAnalytics] class EmailSalesShopAnalytics(BaseModel): shop_name: str = Field(alias="shopName") items_sold: int = Field(alias="itemsSold") net_sales_usd: float = Field(alias="netSalesUsd") class GetEmailSalesShopsAnalyticsResponse(BaseModel): data: list[EmailSalesShopAnalytics] class GetEmailStripoAuthTokenResponse(BaseModel): token: str class AutosaveEmailInput(BaseModel): key: str email_id: str = Field(alias="emailId") html: str css: str class AutosaveEmailV2Input(BaseModel): email_id: str = Field(alias="emailId") user_id: str = Field(alias="userId") date_time: int = Field(alias="dateTime") class UpdateEmailContentInput(BaseModel): html: Base64Str | None = Field(default=None, min_length=1) css: Base64Str | None = Field(default=None, min_length=1) class UpdateEmailFontsInput(BaseModel): font: CustomFont class CustomFontsResponse(BaseModel): items: list[CustomFont] # Campaigns class GetEmailDomainsInput(BaseModel): vendor_id: int = Field( alias="vendorId", description="Account vendor id", ) subaccount_id: int = Field( alias="subaccountId", description="Account subaccount id", ) class GetEmailCampaignsInput(BaseModel): search: str | None = Field(None) vendor_id: int | None = Field(None, alias="vendorId") subaccount_id: int | None = Field(None, alias="subaccountId") fandata_list_ids: set[FanDataListId] | None = Field(None, alias="fanDataListIds") statuses: list[EmailCampaignStatus] | None = Field(None) audience_id: str | None = Field(None, alias="audienceId") order_by: list[EmailCampaignOrderBy] | None = Field(None, alias="orderBy") limit: int | None = Field(None, ge=1, le=200) offset: int | None = Field(None, ge=0) class SearchCampaignsInput(BaseModel): search: str | None = Field(None) search_context: CampaignSearchContext | None = Field(None, alias="searchContext") campaign_ids: set[str] | None = Field(None, alias="campaignIds") vendor_id: int | None = Field(None, alias="vendorId") subaccount_id: int | None = Field(None, alias="subaccountId") fandata_list_ids: set[FanDataListId] | None = Field(None, alias="fanDataListIds") statuses: list[EmailCampaignStatus] | None = Field(None) audience_id: str | None = Field(None, alias="audienceId") order_by: list[EmailCampaignOrderBy] | None = Field(None, alias="orderBy") limit: int | None = Field(None, ge=1, le=200) class CreateEmailCampaignInput(BaseModel): name: str = Field(min_length=1, max_length=255) vendor_id: int = Field(alias="vendorId") subaccount_id: int = Field(alias="subaccountId") fandata_list_id: FanDataListId = Field(alias="fanDataListId") html_content: Base64Str | None = Field(alias="htmlContent", default=None) css_content: Base64Str | None = Field(alias="cssContent", default=None) model_config = ConfigDict(str_strip_whitespace=True) class UpdateEmailCampaignInput(BaseModel): name: str = Field(min_length=1, max_length=255) vendor_id: int = Field(alias="vendorId") subaccount_id: int = Field(alias="subaccountId") fandata_list_id: FanDataListId = Field(alias="fanDataListId") sender_name: str | None = Field( alias="senderName", default=None, min_length=1, max_length=255 ) email_domain_id: str | None = Field(alias="emailDomainId", default=None) email_username: str | None = Field( alias="emailUsername", default=None, min_length=1, max_length=255 ) subject: str | None = Field( alias="subject", default=None, min_length=1, max_length=255 ) preview_text: str | None = Field( alias="previewText", default=None, min_length=1, max_length=255 ) audience_id: str | None = Field(alias="audienceId", default=None) model_config = ConfigDict(str_strip_whitespace=True) class EmailCampaign(BaseModel): id: str name: str vendor_id: int = Field(alias="vendorId") subaccount_id: int = Field(alias="subaccountId") fandata_list_id: FanDataListId | None = Field(None, alias="fanDataListId") status: EmailCampaignStatus audience_id: str | None = Field(alias="audienceId") sender_name: str | None = Field(alias="senderName") subject: str | None preview_text: str | None = Field(alias="previewText") email_domain: EmailDomain | None = Field(alias="emailDomain") email_username: str | None = Field(alias="emailUsername") send_at: AwareDatetime | None = Field(alias="sendAt") send_at_timezone: str | None = Field(alias="sendAtTimezone") recipients_count: int = Field(alias="recipientsCount") cancel_reason: EmailCampaignCancelReason | None = Field(alias="cancelReason") cancelled_at: datetime.datetime | None = Field(alias="cancelledAt") has_unapplied_changes: bool = Field(alias="hasUnappliedChanges") created_at: datetime.datetime = Field(alias="createdAt") created_by: str = Field(alias="createdBy") updated_at: datetime.datetime = Field(alias="updatedAt") updated_by: str = Field(alias="updatedBy") analytics: EmailAnalytics | None = None audience: Audience | None = None model_config = ConfigDict(from_attributes=True) class CampaignPreview(BaseModel): id: str name: str model_config = ConfigDict(from_attributes=True) class SendTestEmailCampaignInput(BaseModel): emails: set[EmailStr] = Field(min_length=1, max_length=999) content: Base64Str | None = Field(default=None) class ScheduleEmailCampaignInput(BaseModel): send_at: AwareDatetime = Field(alias="sendAt") timezone: str = Field(min_length=1, max_length=255) class DuplicateEmailCampaignInput(BaseModel): name: str = Field(min_length=1, max_length=255) model_config = ConfigDict(str_strip_whitespace=True) class EmailCampaignInternal(BaseModel): id: str name: str vendor_id: int = Field(alias="vendorId") subaccount_id: int = Field(alias="subaccountId") global_participant_id: str = Field(alias="globalParticipantId") status: EmailCampaignStatus audience_id: str = Field(alias="audienceId") sender_name: str = Field(alias="senderName") html_content: str = Field(alias="html") css_content: str = Field(alias="css") subject: str preview_text: str | None = Field(alias="previewText") email_domain: EmailDomain = Field(alias="emailDomain") email_username: str = Field(alias="emailUsername") send_at: AwareDatetime | None = Field(alias="sendAt") send_at_timezone: str | None = Field(alias="sendAtTimezone") recipients_count: int = Field(alias="recipientsCount") created_at: datetime.datetime = Field(alias="createdAt") created_by: str = Field(alias="createdBy") updated_at: datetime.datetime = Field(alias="updatedAt") updated_by: str = Field(alias="updatedBy") model_config = ConfigDict(from_attributes=True) # Schedules class EmailCampaignSchedule(BaseModel): campaign_id: str = Field(alias="campaignId") campaign_status: EmailCampaignStatus | None = Field(alias="campaignStatus") campaign_send_at: AwareDatetime | None = Field(alias="campaignSendAt") queue_enabled: bool | None = Field(alias="queueEnabled") class GetCampaignLiveStatusOutput(BaseModel): status: EmailCampaignStatus emails_to_send: int = Field(alias="emailsToSend") sent_emails: int = Field(alias="sentEmails") class EstimateSendTimeResponse(BaseModel): estimated_send_duration: int = Field( alias="estimatedSendDuration", description="Estimated duration in seconds from send_at until campaign completes sending.", ) emails_to_send: int = Field(alias="emailsToSend") emails_count: int = Field(alias="emailsCount") class BulkEstimateSendTimeRequest(BaseModel): campaign_ids: list[str] = Field( alias="campaignIds", min_length=1, max_length=100, ) class CampaignEstimate(BaseModel): campaign_id: str = Field(alias="campaignId") estimated_send_duration: int = Field(alias="estimatedSendDuration") emails_to_send: int = Field(alias="emailsToSend") # Source Campaigns class SourceCampaign(BaseModel): id: str campaign_id: str = Field(alias="campaignId") source: str name: str url: str | None status: SourceCampaignStatus has_other_connections: bool | None = Field(None, alias="hasOtherConnections") fandata_list_id: FanDataListId | None = Field(None, alias="fanDataListId") model_config = ConfigDict(from_attributes=True) class SourceCampaignAutomatedEmail(BaseModel): id: str name: str type: AutomatedEmailType subject: str | None preview_text: str | None = Field(None, alias="previewText") sender_name: str | None = Field(None, alias="senderName") email_username: str | None = Field(None, alias="emailUsername") email_domain: EmailDomain | None = Field(None, alias="emailDomain") subscription_only: bool = Field(alias="subscriptionOnly") potential_mailing_list_ids: list[str] = Field( default_factory=list, alias="potentialMailingListIds" ) model_config = ConfigDict(from_attributes=True) class GetSourceCampaignAutomatedEmailsResponse(BaseModel): is_participation_email_automated: bool = Field( alias="isParticipationEmailAutomated" ) is_subscription_email_automated: bool = Field(alias="isSubscriptionEmailAutomated") items: list[SourceCampaignAutomatedEmail] class GetSourceCampaignsInput(BaseModel): search: str | None = Field(None) vendor_id: int | None = Field(None, alias="vendorId") subaccount_id: int | None = Field(None, alias="subaccountId") current_automated_email_id: str | None = Field( None, alias="currentAutomatedEmailId" ) status: str | None = Field(None) email_type: AutomatedEmailType | None = Field(None, alias="emailType") order_by: list[SourceCampaignOrderBy] | None = Field(None, alias="orderBy") limit: int | None = Field(None, ge=1, le=500) offset: int | None = Field(None, ge=0) # Automated Emails class AutomatedEmailTrigger(BaseModel): subscription_only: bool = Field(alias="subscriptionOnly") source_campaign: SourceCampaign = Field(alias="sourceCampaign") model_config = ConfigDict(from_attributes=True) class SourceCampaignConnection(BaseModel): source_campaign_id: str = Field(alias="sourceCampaignId") subscription_only: bool = Field(alias="subscriptionOnly", default=False) class AttachSourceCampaignsInput(BaseModel): source_campaign_connections: list[SourceCampaignConnection] = Field( alias="connectTo" ) class DetachSourceCampaignsInput(BaseModel): source_campaign_ids: list[str] = Field(alias="disconnectFrom") class AutomatedEmail(BaseModel): id: str name: str type: AutomatedEmailType vendor_id: int = Field(alias="vendorId") subaccount_id: int = Field(alias="subaccountId") fandata_list_id: FanDataListId | None = Field(None, alias="fanDataListId") sender_name: str | None = Field(None, alias="senderName") email_username: str | None = Field(None, alias="emailUsername") email_domain: EmailDomain | None = Field(alias="emailDomain") subject: str | None = Field(None) preview_text: str | None = Field(None, alias="previewText") is_automated: bool = Field(alias="isAutomated") source_campaign_connections: list[AutomatedEmailTrigger] = Field( serialization_alias="connectedTo" ) last_sent_at: AwareDatetime | None = Field(alias="lastSentAt") analytics: EmailAnalytics | None = None has_unapplied_changes: bool = Field(alias="hasUnappliedChanges") created_at: datetime.datetime = Field(alias="createdAt") created_by: str = Field(alias="createdBy") updated_at: datetime.datetime = Field(alias="updatedAt") updated_by: str = Field(alias="updatedBy") model_config = ConfigDict(from_attributes=True) class AutomatedEmailStatus(BaseModel): is_automated: bool = Field(alias="isAutomated") has_unapplied_changes: bool = Field(alias="hasUnappliedChanges") model_config = ConfigDict(from_attributes=True) class GetAutomatedEmailsInput(BaseModel): search: str | None = Field(None) vendor_id: int | None = Field(None, alias="vendorId") subaccount_id: int | None = Field(None, alias="subaccountId") fandata_list_ids: set[FanDataListId] | None = Field(None, alias="fanDataListIds") is_automated: bool | None = Field(None, alias="isAutomated") order_by: list[AutomatedEmailOrderBy] | None = Field(None, alias="orderBy") limit: int | None = Field(None, ge=1, le=200) offset: int | None = Field(None, ge=0) class CreateAutomatedEmailInput(BaseModel): name: str = Field(min_length=1, max_length=255) type: AutomatedEmailType vendor_id: int = Field(alias="vendorId") subaccount_id: int = Field(alias="subaccountId") fandata_list_id: FanDataListId = Field(alias="fanDataListId") html_content: Base64Str | None = Field(alias="htmlContent", default=None) css_content: Base64Str | None = Field(alias="cssContent", default=None) model_config = ConfigDict(str_strip_whitespace=True) class UpdateAutomatedEmailInput(BaseModel): name: str = Field(min_length=1, max_length=255) vendor_id: int = Field(alias="vendorId") subaccount_id: int = Field(alias="subaccountId") fandata_list_id: FanDataListId = Field(alias="fanDataListId") type: AutomatedEmailType sender_name: str | None = Field( alias="senderName", default=None, min_length=1, max_length=255 ) email_domain_id: str | None = Field(alias="emailDomainId", default=None) email_username: str | None = Field( alias="emailUsername", default=None, min_length=1, max_length=255 ) subject: str | None = Field( alias="subject", default=None, min_length=1, max_length=255 ) preview_text: str | None = Field( alias="previewText", default=None, min_length=1, max_length=255 ) model_config = ConfigDict(str_strip_whitespace=True) class DuplicateAutomatedEmailInput(BaseModel): name: str = Field(min_length=1, max_length=255) model_config = ConfigDict(str_strip_whitespace=True) class SendAutomatedTestEmailInput(BaseModel): emails: set[EmailStr] = Field(min_length=1, max_length=999) content: Base64Str | None = Field(default=None) with_unapplied_changes: bool = Field(default=False, alias="withUnappliedChanges") # Email Reporting class EmailReportingReportSummaryResponse(BaseModel): id: str name: str emails_count: int = Field(alias="emailsCount") identity_id: str = Field(alias="createdBy") created_at: datetime.datetime = Field(alias="createdAt") updated_at: datetime.datetime = Field(alias="updatedAt") model_config = ConfigDict(from_attributes=True) class EmailReportingReportResponse(BaseModel): id: str name: str email_ids: list[str] = Field(alias="emailIds") emails_count: int = Field(alias="emailsCount") identity_id: str = Field(alias="createdBy") created_at: datetime.datetime = Field(alias="createdAt") updated_at: datetime.datetime = Field(alias="updatedAt") model_config = ConfigDict(from_attributes=True) class GetEmailReportingReportsInput(BaseModel): name: str | None = Field(None) limit: int | None = Field(None, ge=1, le=200) offset: int | None = Field(None, ge=0) class EmailReportingReportInput(BaseModel): name: str = Field(min_length=1, max_length=255) email_ids: list[str] | None = Field(None, alias="emailIds") vendor_id: int | None = Field(None, alias="vendorId") subaccount_id: int | None = Field(None, alias="subaccountId") global_participant_id: str | None = Field(None, alias="globalParticipantId") model_config = ConfigDict(str_strip_whitespace=True) class EmailReportingAnalytics(BaseModel): sends: int = Field(alias="sends") delivered: int = Field(alias="delivered") opens: int = Field(alias="opens") clicks: int = Field(alias="clicks") bounces: int = Field(alias="bounces") spam_complaints: int = Field(alias="spamComplaints") unsubscribes: int = Field(alias="unsubscribes") open_rate: float = Field(alias="openRate") delivered_rate: float = Field(alias="deliveredRate") click_through_rate: float = Field(alias="clickThroughRate") click_to_open_rate: float = Field(alias="clickToOpenRate") bounce_rate: float = Field(alias="bounceRate") spam_complaints_rate: float = Field(alias="spamComplaintsRate") unsubscribe_rate: float = Field(alias="unsubscribeRate") class GetEmailReportingAnalyticsResponse(BaseModel): data: EmailReportingAnalytics class EmailReportingLinkAnalytics(BaseModel): url: str = Field(alias="url") total_clicks: int = Field(alias="totalClicks") percentage_of_total_clicks: float = Field(alias="percentageOfTotalClicks") class GetEmailReportingReportLinkAnalyticsResponse(BaseModel): data: list[EmailReportingLinkAnalytics]