""" AUTHOR: Khushal P Soonderji DATE: Monday, 2nd Dec., 2024. OBJECTIVE: To provide the structure for the request that will come in to sync the mails of a particular user. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # For making data behaviour_models: from pydantic import BaseModel, Field, field_validator, PastDatetime, AwareDatetime from typing import Optional, Literal # My utils: from utils_v2.string import regex from utils_v2.date_time import date_time # Data models: from models.core.message import CoreMessageModel # To work with date and time: import datetime # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # RegEx Patterns: REGEX_SESSION_TOKEN = r"^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$" # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** class MailSyncRequestHeaders(BaseModel): sessionToken: str = Field( description = "the session token of the user who is requesting the service", pattern = REGEX_SESSION_TOKEN, frozen = True, alias = "X-Session-Token" ) # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ # ┗┛┗┛┛┗┛┗┗┫ # ┛ class Config: extra = "allow" def model_dump(self, *args, **kwargs): return super().model_dump(*args, by_alias = True, **kwargs) # --------------------------------------------------------------------------------------------------------------------- class MailSyncRequestData(BaseModel): tokenKey: str = Field( description = "the account identifier (Mongo ObjectId) granted by 'MailOAuthModel.get_account_identifier'", frozen = True ) maxCount: int = Field( description = "the max. no. of e-mails to sync at a given time", default = 100, ge = 1, le = 100, frozen = True ) startDate: AwareDatetime = Field( description = "the starting date from which the user wants to sync their mail", default_factory = lambda: date_time.get_current_utc_date_time() - datetime.timedelta(days = 1), frozen = True ) endDate: AwareDatetime = Field( description = "the ending date till which the user wants to sync their mail", default_factory = lambda: date_time.get_current_utc_date_time() - datetime.timedelta(seconds = 1), frozen = True ) forceSync: bool = Field( description = "use this to forcefully re-sync mails when you need to overwrite existing data in mongodb", default = False ) # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ # ┗┛┗┛┛┗┛┗┗┫ # ┛ class Config: extra = "forbid" # ┓┏ ┓• ┓ • # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓ # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗ @field_validator("startDate", "endDate", mode = "before") def to_datetime(cls, value): if not isinstance(value, datetime.datetime): value = date_time.parse_date_time( input_value = value, timezone = date_time.TIMEZONE_UTC ) return value # --------------------------------------------------------------------------------------------------------------------- class MailSyncOneResult(BaseModel): success: bool = Field( description = "whether, or not, the mail was successfully sync'd", default = False ) message: str | None = Field( description = "a brief message to summarize the result of the process", default = None ) mailMessage: CoreMessageModel | None = Field( description = "the actual data of the mail; can be null in a successful process if the mail is already sync'd", default = None ) # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ # ┗┛┗┛┛┗┛┗┗┫ # ┛ class Config: extra = "forbid" # --------------------------------------------------------------------------------------------------------------------- class MailSyncManyResults(BaseModel): totalCount: int = Field( description = "the total no. of mails that were to be sync'd", default = 0 ) successCount: int = Field( description = "the no. of mails that were successfully sync'd", default = 0 ) failureCount: int = Field( description = "the no. of mails that were successfully sync'd", default = 0 ) message: str = Field( description = "a brief message to summarize the results of the process", default = None ) # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ # ┗┛┗┛┛┗┛┗┗┫ # ┛ class Config: extra = "forbid" # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass