""" AUTHOR: Khushal P Soonderji DATE: Friday, 13th Dec., 2024 OBJECTIVE: To handle all SMS related behaviour from one place. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # For Quart: from quart import current_app # My async utils: from utils_v2.string import json from utils_v2.date_time import date_time from utils_v2.database.async_mysql_v2 import AsyncMySQL from utils_v2.database.async_mongo_v2 import AsyncMongo, AsyncMongoStorage from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache # Controllers: from controllers_v2.core.message import CoreMessageController # Models: from models.core.auth_token import CoreAuthTokenModel from models.core.message import CoreMessageModel from models.core.user import CoreUserInfoModel from models.api.sms.send import ( SMSSendRequestData, NimbusSMSIndiaMessage, SavvyBulkSMSKenyaMessage, SMSSendManyResults ) # To work with MongoDB: from bson import ObjectId from pymongo import InsertOne, UpdateOne, ReplaceOne # To work with datatypes: from typing import Literal, List, Dict, Any # To make HTTP requests: import httpx # To make deep-copies: import copy # To work with base-64 encoding: import base64 # To work with date and time: import datetime # For asynchronous activities: import asyncio # To make abstract classes: from abc import ABC, abstractmethod # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class SMSController(CoreMessageController, ABC): # ┏┓ # ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓ # ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛ def __init__( self, cache: AsyncRedisCache = None, http_client: httpx.AsyncClient = None, alert_url: str = None, base_filter: dict = None, debug: bool = True, debug_prefix: str = "SMS (C) | ", debug_only_errors: bool = True ): """ This is the foundational controller for all SMS services. This is built on top of the core message controller, and, in turn, all individual SMS client controllers must be built on top of this. :param cache: The object to use for caching results from database calls. :param http_client: The HTTP client :param base_filter: The basic filter that will be applied to all fetching/updating queries. WARNING: THE BASE FILTER WILL ALWAYS BE APPLIED AUTOMATICALLY. SET THIS UP WISELY. :param debug: Whether, or not, you would like to print debugging messages: :param debug_prefix: The prefix to print with the debugging messages. :param debug_only_errors: Whether you would like to print only error messages or all messages. :return: None. """ # Prepare the combined base filter: sms_filter = {"serviceType": "sms"} if base_filter: for k, v in base_filter.items(): sms_filter[k] = v # Invoke the parent's constructor: super().__init__( cache = cache, alert_url = alert_url, http_client = http_client, base_filter = sms_filter, debug = debug, debug_prefix = debug_prefix, debug_only_errors = debug_only_errors ) # ┏┓┳┳┓┏┓ ┏┓ ┓• # ┗┓┃┃┃┗┓ ┗┓┏┓┏┓┏┫┓┏┓┏┓ # ┗┛┛ ┗┗┛ ┗┛┗ ┛┗┗┻┗┛┗┗┫ # ┛ @abstractmethod async def send( self, mongo_conn: AsyncMongo, auth_token: CoreAuthTokenModel, messages: List[NimbusSMSIndiaMessage | SavvyBulkSMSKenyaMessage], tags: List[Any] ) -> SMSSendManyResults: pass # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass