khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(20250116) Started upgrading the mail module (to eventually work on cron).

Browse Source
This commit is contained in:
khushal
2025-01-16 16:39:23 +05:30
parent efd7dda9f0
commit ed1e86c5e9
71 changed files with 2423 additions and 464 deletions
Download Patch File Download Diff File Expand all files Collapse all files
models/api/finstitutions/trading/auth/oauth.py
+7 -98
View File
@@ -39,6 +39,13 @@ sys.path.append("..")
from pydantic import BaseModel, Field, field_validator, PastDatetime, model_validator
from typing import Optional, Literal, Union
# Models:
from models.finstitutions.trading.oauth import (
PaperTradingAuth,
ZerodhaKiteAuth,
ICICIBreezeAuth
)
# My utils:
from utils_v2.string import regex
from utils_v2.date_time import date_time
@@ -75,104 +82,6 @@ REGEX_SESSION_TOKEN = r"^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]
# *****************************************************************************************************************
class PaperTradingAuth(BaseModel):
username: str = Field(
description = "??",
frozen = True
)
password: str = Field(
description = "??",
frozen = True
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# ---------------------------------------------------------------------------------------------------------------------
class ZerodhaKiteAuth(BaseModel):
userId: str = Field(
description = "How Zerodha's Kite platform identifies this user.",
frozen = True,
alias = "clientId"
)
apiKey: str = Field(
description = (
"The API key of your Kite app. "
"This remains constant throughout the life of the app."
),
frozen = True
)
apiSecret: str = Field(
description = (
"The API secret of your Kite app. "
"This can be changed if you think the security of your app has been compromised."
),
frozen = True
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
populate_by_name = True
# ---------------------------------------------------------------------------------------------------------------------
class ICICIBreezeAuth(BaseModel):
userId: str = Field(
description = "How ICICI's Breeze platform identifies this user.",
frozen = True,
alias = "clientId"
)
apiKey: str = Field(
description = (
"The API key of your Breeze app. "
"This remains constant throughout the life of the app."
),
frozen = True
)
apiSecret: str = Field(
description = (
"The API secret of your Breeze app. "
"This can be changed if you think the security of your app has been compromised."
),
frozen = True
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
populate_by_name = True
# ---------------------------------------------------------------------------------------------------------------------
class TradingAuthRequestHeaders(BaseModel):
sessionToken: str = Field(
models/api/finstitutions/trading/symbols/list.py
-39
View File
@@ -79,45 +79,6 @@ REGEX_SESSION_TOKEN = r"^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]
# *****************************************************************************************************************
class TradingSymbolListBrokerResponse(BaseModel):
success: bool = Field(
description = "whether, or not, the symbol list request was successful",
default = False,
frozen = False
)
message: str = Field(
description = "a brief message to help debug in failed cases",
default = "",
frozen = False
)
data: List[TradingSymbol] | None = Field(
description = "the actual response from the broker with his list of tradeable symbols",
default = None,
frozen = False
)
exception: Exception | None = Field(
description = "if something goes wrong, the exception will be held here",
default = None,
frozen = False
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
arbitrary_types_allowed = True
# ---------------------------------------------------------------------------------------------------------------------
class TradingSymbolListRequestHeaders(BaseModel):
sessionToken: str = Field(
models/api/chat/__init__.py → models/api/message/__init__.py
View File
models/api/mail/__init__.py → models/api/message/chat/__init__.py
View File
models/api/chat/auth.py → models/api/message/chat/auth.py
+3 -46
View File
@@ -39,6 +39,9 @@ sys.path.append("..")
from pydantic import BaseModel, Field, field_validator, PastDatetime, model_validator
from typing import Optional, Literal, Union
# Models:
from models.message.chat.auth import TelegramAuth, WhatsAppNimbusAuth
# My utils:
from utils_v2.string import regex
from utils_v2.date_time import date_time
@@ -75,52 +78,6 @@ REGEX_SESSION_TOKEN = r"^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]
# *****************************************************************************************************************
class TelegramAuth(BaseModel):
botToken: str = Field(
description = "the token granted by BotFather",
min_length = 1,
frozen = True
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# ---------------------------------------------------------------------------------------------------------------------
class WhatsAppNimbusAuth(BaseModel):
apiKey: str = Field(
description = "???",
min_length = 1,
frozen = True
)
senderId: str = Field(
description = "???",
min_length = 1,
frozen = True
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# ---------------------------------------------------------------------------------------------------------------------
class ChatAuthRequestHeaders(BaseModel):
sessionToken: str = Field(
models/api/sms/__init__.py → models/api/message/mail/__init__.py
View File
models/api/mail/get.py → models/api/message/mail/get.py
View File
models/api/mail/list.py → models/api/message/mail/list.py
View File
models/api/mail/oauth.py → models/api/message/mail/oauth.py
View File
models/api/mail/send.py → models/api/message/mail/send.py
View File
models/api/mail/sync.py → models/api/message/mail/sync.py
-92
View File
@@ -156,98 +156,6 @@ class MailSyncRequestData(BaseModel):
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"
# ---------------------------------------------------------------------------------------------------------------------
class MailSendOneResult(BaseModel):
success: bool = Field(
description = "whether, or not, the mail was successfully sent",
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 sent",
default = None
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
models/api/mail/tags.py → models/api/message/mail/tags.py
View File
models/api/message/sms/__init__.py
View File
models/api/message/sms/auth.py
+144
View File
@@ -0,0 +1,144 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Thursday, 5th Dec., 2024.
OBJECTIVE:
To provide a structure to receive auth details of various SMS providers.
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, model_validator
from typing import Optional, Literal, Union
# Models:
from models.message.sms.auth import NimbusSMSIndiaAuth, SavvyBulkSMSKenyaAuth
# My utils:
from utils_v2.string import regex
from utils_v2.date_time import date_time
# 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 SMSAuthRequestHeaders(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 SMSAuthRequestData(BaseModel):
smsClient: Literal["nimbusSmsIndia", "savvyBulkSmsKenya"] = Field(alias = "client")
auth: Union[NimbusSMSIndiaAuth, SavvyBulkSMSKenyaAuth]
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
@model_validator(mode = "after")
def ensure_harmony(cls, values):
client = values.smsClient
auth = values.auth
harmony_map = {
"nimbusSmsIndia": NimbusSMSIndiaAuth,
"savvyBulkSmsKenya": SavvyBulkSMSKenyaAuth
}
if not isinstance(auth, harmony_map[client]):
raise ValueError(f"incorrect 'auth' for selected client '{client}'")
return values
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
models/api/sms/list.py → models/api/message/sms/list.py
View File
models/api/message/sms/send.py
+160
View File
@@ -0,0 +1,160 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Monday, 9th Dec., 2024.
OBJECTIVE:
To provide a structure to receive API calls to send SMS messages from various third-party clients.
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
from typing import Optional, Literal, Union, List, Any
# My utils:
from utils_v2.string import regex
from utils_v2.date_time import date_time
# Models:
from models.core.message import CoreMessageModel
from utils_v2.sms.models.sms_message import SentSMSMessageModel
from models.message.sms.send import (
NimbusSMSIndiaMessage,
SavvyBulkSMSKenyaMessage,
SMSSendOneResult,
SMSSendManyResults
)
# To work with date and time:
import datetime
# To work with MongoDB:
from bson.objectid import ObjectId
# *****************************************************************************************************************
# ***** ****
# *** 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 SMSSendRequestHeaders(BaseModel):
sessionToken: str | None = Field(
description = "the session token of the user who is requesting the service",
pattern = REGEX_SESSION_TOKEN,
frozen = True,
default = None,
alias = "X-Session-Token"
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "allow"
def model_dump(self, *args, **kwargs):
return super().model_dump(*args, by_alias = True, **kwargs)
# ---------------------------------------------------------------------------------------------------------------------
class SMSSendRequestData(BaseModel):
tokenKey: ObjectId
message: List[NimbusSMSIndiaMessage] | List[SavvyBulkSMSKenyaMessage]
tags: List[Any] | None = Field(default = None, validate_default = True)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
arbitrary_types_allowed = True
# ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
@field_validator("tokenKey", mode = "before")
def parse_oid(cls, value):
try: value = ObjectId(value)
except: pass
return value
@field_validator("message", mode = "before")
def ensure_list(cls, value):
if not isinstance(value, list): value = [value]
return value
@field_validator("tags", mode = "after")
def null_to_list(cls, value):
return [] if value is None else value
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
models/api/sms/tags.py → models/api/message/sms/tags.py
View File
models/finstitutions/trading/oauth.py
+98
View File
@@ -74,6 +74,104 @@ import datetime
# *****************************************************************************************************************
class PaperTradingAuth(BaseModel):
username: str = Field(
description = "??",
frozen = True
)
password: str = Field(
description = "??",
frozen = True
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# ---------------------------------------------------------------------------------------------------------------------
class ZerodhaKiteAuth(BaseModel):
userId: str = Field(
description = "How Zerodha's Kite platform identifies this user.",
frozen = True,
alias = "clientId"
)
apiKey: str = Field(
description = (
"The API key of your Kite app. "
"This remains constant throughout the life of the app."
),
frozen = True
)
apiSecret: str = Field(
description = (
"The API secret of your Kite app. "
"This can be changed if you think the security of your app has been compromised."
),
frozen = True
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
populate_by_name = True
# ---------------------------------------------------------------------------------------------------------------------
class ICICIBreezeAuth(BaseModel):
userId: str = Field(
description = "How ICICI's Breeze platform identifies this user.",
frozen = True,
alias = "clientId"
)
apiKey: str = Field(
description = (
"The API key of your Breeze app. "
"This remains constant throughout the life of the app."
),
frozen = True
)
apiSecret: str = Field(
description = (
"The API secret of your Breeze app. "
"This can be changed if you think the security of your app has been compromised."
),
frozen = True
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
populate_by_name = True
# ---------------------------------------------------------------------------------------------------------------------
class TradingOAuthCallbackResponse(BaseModel):
success: bool = Field(
models/finstitutions/trading/symbols.py
+39
View File
@@ -220,6 +220,45 @@ class TradingSymbol(BaseModel):
return value
# ---------------------------------------------------------------------------------------------------------------------
class TradingSymbolListBrokerResponse(BaseModel):
success: bool = Field(
description = "whether, or not, the symbol list request was successful",
default = False,
frozen = False
)
message: str = Field(
description = "a brief message to help debug in failed cases",
default = "",
frozen = False
)
data: List[TradingSymbol] | None = Field(
description = "the actual response from the broker with his list of tradeable symbols",
default = None,
frozen = False
)
exception: Exception | None = Field(
description = "if something goes wrong, the exception will be held here",
default = None,
frozen = False
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
arbitrary_types_allowed = True
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
models/message/__init__.py
View File
models/message/chat/__init__.py
View File
models/message/chat/auth.py
+130
View File
@@ -0,0 +1,130 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Friday, 6th Dec., 2024.
OBJECTIVE:
To provide a structure to receive auth details of various chat apps (like Telegram and WhatsApp).
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, model_validator
from typing import Optional, Literal, Union
# My utils:
from utils_v2.string import regex
from utils_v2.date_time import date_time
# 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 TelegramAuth(BaseModel):
botToken: str = Field(
description = "the token granted by BotFather",
min_length = 1,
frozen = True
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# ---------------------------------------------------------------------------------------------------------------------
class WhatsAppNimbusAuth(BaseModel):
apiKey: str = Field(
description = "???",
min_length = 1,
frozen = True
)
senderId: str = Field(
description = "???",
min_length = 1,
frozen = True
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
models/message/mail/__init__.py
View File
models/message/mail/oauth.py
+185
View File
@@ -0,0 +1,185 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Thursday, 16th Jan., 2025.
OBJECTIVE:
To provide the structure for the request and response of the APIs that will be used to request OAuth2.0
authorization for mail services.
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
from typing import Optional, Literal, Any
# My utils:
from utils_v2.string import regex
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
class OAuthMailGetAuthorizationURLResponse(BaseModel):
success: bool = Field(
description = "To indicate whether or not, the action was a success",
frozen = False,
default = False
)
url: str | None = Field(
description = "The URL to use to integrate the mail client.",
frozen = False,
default = None
)
message: str = Field(
description = "To explain what happened in the process generating a URL.",
frozen = False,
default = "ERR: Message not captured."
)
exception: Any = Field(
description = "To pass on any exception that occurred in the process.",
frozen = False,
default = None
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# ┏┓ ┏┓
# ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏
# ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛
pass
# ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
pass
# ---------------------------------------------------------------------------------------------------------------------
class OAuthMailHandleCallbackResponse(BaseModel):
success: bool = Field(
description = "To indicate whether or not, the action was a success",
frozen = False,
default = False
)
action: Literal[
"unknown", # ..... Initial value when the user's intent is not known.
"denied", # ...... The user consciously denied permission.
"cancelled", # ... The user cancelled the process midway.
"authorized" # ... The user gave authorization.
] = Field(
description = "To describe what the user did with the authorization URL.",
frozen = True,
default = "unknown"
)
message: str = Field(
description = "To explain what happened in the process of handling the OAuth callback.",
frozen = False,
default = "ERR: Message not captured."
)
exception: Any = Field(
description = "To pass on any exception that occurred in the process.",
frozen = False,
default = None
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# ┏┓ ┏┓
# ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏
# ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛
pass
# ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
pass
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
models/message/mail/send.py
+120
View File
@@ -0,0 +1,120 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Thursday, 19th Dec., 2024.
OBJECTIVE:
To provide a structure to send mails.
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, EmailStr
from typing import Optional, Literal, List
# Models:
from models.core.message import CoreMessageModel
# My utils:
from utils_v2.string import json
from utils_v2.string import regex
from utils_v2.date_time import date_time
# To work with date and time:
import datetime
# To work with MongoDB:
from bson.objectid import ObjectId
# *****************************************************************************************************************
# ***** ****
# *** 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 MailSendOneResult(BaseModel):
success: bool = Field(
description = "whether, or not, the mail was successfully sent",
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 sent",
default = None
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
models/message/mail/sync.py
+150
View File
@@ -0,0 +1,150 @@
"""
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 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
models/message/sms/__init__.py
View File
models/api/sms/auth.py → models/message/sms/auth.py
-57
View File
@@ -143,63 +143,6 @@ class SavvyBulkSMSKenyaAuth(BaseModel):
extra = "forbid"
# ---------------------------------------------------------------------------------------------------------------------
class SMSAuthRequestHeaders(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 SMSAuthRequestData(BaseModel):
smsClient: Literal["nimbusSmsIndia", "savvyBulkSmsKenya"] = Field(alias = "client")
auth: Union[NimbusSMSIndiaAuth, SavvyBulkSMSKenyaAuth]
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
@model_validator(mode = "after")
def ensure_harmony(cls, values):
client = values.smsClient
auth = values.auth
harmony_map = {
"nimbusSmsIndia": NimbusSMSIndiaAuth,
"savvyBulkSmsKenya": SavvyBulkSMSKenyaAuth
}
if not isinstance(auth, harmony_map[client]):
raise ValueError(f"incorrect 'auth' for selected client '{client}'")
return values
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
models/api/sms/send.py → models/message/sms/send.py
-63
View File
@@ -151,69 +151,6 @@ class SavvyBulkSMSKenyaMessage(BaseModel):
# ---------------------------------------------------------------------------------------------------------------------
class SMSSendRequestHeaders(BaseModel):
sessionToken: str | None = Field(
description = "the session token of the user who is requesting the service",
pattern = REGEX_SESSION_TOKEN,
frozen = True,
default = None,
alias = "X-Session-Token"
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "allow"
def model_dump(self, *args, **kwargs):
return super().model_dump(*args, by_alias = True, **kwargs)
# ---------------------------------------------------------------------------------------------------------------------
class SMSSendRequestData(BaseModel):
tokenKey: ObjectId
message: List[NimbusSMSIndiaMessage] | List[SavvyBulkSMSKenyaMessage]
tags: List[Any] | None = Field(default = None, validate_default = True)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
arbitrary_types_allowed = True
# ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
@field_validator("tokenKey", mode = "before")
def parse_oid(cls, value):
try: value = ObjectId(value)
except: pass
return value
@field_validator("message", mode = "before")
def ensure_list(cls, value):
if not isinstance(value, list): value = [value]
return value
@field_validator("tags", mode = "after")
def null_to_list(cls, value):
return [] if value is None else value
# ---------------------------------------------------------------------------------------------------------------------
class SMSSendOneResult(BaseModel):
success: bool = Field(