(20250220) Bug fix in reminders for work.

api/blueprints/common/otp/timed_otp.py

@@ -0,0 +1,292 @@
+"""
+
+    AUTHOR:
+
+        Khushal P Soonderji
+
+    DATE:
+
+        Tuesday, 10th Sept., 2024
+
+    OBJECTIVE:
+
+        To be able to send SMS messages and get reports and balance through Nimbus IT's SMS sending service.
+        The core logic is implemented elsewhere, this module aims to build an API layer around it.
+
+    REFERENCES:
+
+        N/A
+
+    DOWNLOADS:
+
+        N/A
+
+    NOTES:
+
+        N/A
+
+"""
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                   IMPORT                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# To make sibling directories accessible for imports:
+import sys
+sys.path.append(".")
+sys.path.append("..")
+
+# For using Quart:
+from quart import Blueprint, current_app
+
+# Common:
+from shared import constants
+
+# My utils:
+from utils_v2.string import json
+from utils_v2.date_time import date_time
+from utils_v2.security.otp import HashedOTP
+from utils_v2.api.response import ResponseModel
+from utils_v2.api.codes import StatusCodes, HttpCodes
+from utils_v2.api.async_quart import (
+    log_request_to_mongo,
+    only_whitelisted_ips,
+    should_not_be_under_maintenance,
+    read_input,
+    limit_rate, validate_input
+)
+
+# Data models:
+from models.common.otp.timed_otp import GenerateTimedOTP, VerifyTimedOTP
+
+# For asynchronous activities:
+import asyncio
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                           MACROS / ONE-TIME INIT                                              ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# Related to Quart:
+timed_otp_bp = Blueprint("timed_otp", __name__)
+api_version = "2.0.0"
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                VARIABLES                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                  FUNCTIONS                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+@timed_otp_bp.record_once
+def init(blueprint_setup_state):
+
+    # This gets called when the blueprint is registered.
+    # Consider this to be a one-time setup for the whole blueprint:
+    pass
+
+
+# ---------------------------------------------------------------------------------------------------------------------
+
+
+@timed_otp_bp.route("/generate", methods = ["POST", "GET"])
+@read_input(sanitize_headers = True, sanitize_data = True)
+@log_request_to_mongo(
+    attr_name = "mongo",
+    log_type = "otp",
+    operation = "generateOtp",
+    api_version = api_version,
+    log_input = True,
+    log_output = True
+)
+@should_not_be_under_maintenance(attr_name = "is_under_maintenance")
+# @only_whitelisted_ips(attr_name = "whitelisted_ips")
+@validate_input(data_validator = lambda x: GenerateTimedOTP(**x))
+# @limit_rate(
+#     attr_name = "redis_cache",
+#     rate_limit = 1,
+#     seconds = 2,
+#     data_keys = ["id"],
+#     allow_if_exception = False
+# )
+async def generate_otp(
+        inbound_headers: dict = None,
+        inbound_data: dict | GenerateTimedOTP = None,
+        inbound_files: dict = None,
+        log_id: str = None
+):
+
+    """
+    Generates one OTP for the amount of time specified in the request.
+    :param inbound_headers: auto-extracted by the decorators from 'async_quart.py'.
+    :param inbound_data: auto-extracted by the decorators from 'async_quart.py'.
+    :param inbound_files: auto-extracted by the decorators from 'async_quart.py'.
+    :param log_id: An identifier for the logs (if logging is enabled).
+    :return: A standard response structure from the function in 'async_quart.py'.
+    """
+
+    try:
+
+        # Generate the OTP:
+        otp_key = current_app.redis_cache.make_key(str(inbound_data.id))
+        otp_client = HashedOTP(secret = HashedOTP.generate_secret())
+        otp = otp_client.generate_otp(count = 0)
+
+        # Delete any existing OTP with the same identifiers:
+        await current_app.redis_cache.delete(key = otp_key)
+
+        # Store the OTP in Redis:
+        otp_stored = await current_app.redis_cache.set(
+            key = otp_key,
+            value = {
+                "otp": otp,
+                "att": inbound_data.attempts,
+                "iat": date_time.get_current_utc_date_time().timestamp(),
+                "sec": inbound_data.seconds
+            },
+            expiry = inbound_data.seconds
+        )
+
+        # Done here:
+        return ResponseModel(
+            api_version = api_version,
+            status_code = StatusCodes.OK if otp_stored else StatusCodes.FAILED,
+            data = {"otp": otp} if otp_stored else None
+        )
+
+    # In case the client terminates the connection prematurely:
+    except asyncio.CancelledError as exception:
+        current_app.printer(exception)
+        return ResponseModel(api_version = api_version, status_code = StatusCodes.CLIENT_CLOSED_REQUEST)
+
+
+# ---------------------------------------------------------------------------------------------------------------------
+
+
+@timed_otp_bp.route("/verify", methods = ["POST", "GET"])
+@read_input(sanitize_headers = True, sanitize_data = True)
+@log_request_to_mongo(
+    attr_name = "mongo",
+    log_type = "otp",
+    operation = "verifyOtp",
+    api_version = api_version,
+    log_input = True,
+    log_output = True
+)
+@should_not_be_under_maintenance(attr_name = "is_under_maintenance")
+# @only_whitelisted_ips(attr_name = "whitelisted_ips")
+@validate_input(data_validator = lambda x: VerifyTimedOTP(**x))
+# @limit_rate(
+#     attr_name = "redis_cache",
+#     rate_limit = 1,
+#     seconds = 2,
+#     data_keys = ["id"],
+#     allow_if_exception = False,
+#     count_for_http_codes = [200]
+# )
+async def verify_otp(
+        inbound_headers: dict = None,
+        inbound_data: dict | VerifyTimedOTP = None,
+        inbound_files: dict = None,
+        log_id: str = None
+):
+
+    """
+    Verifies the claimed OTP against the stored OTP.
+    :param inbound_headers: auto-extracted by the decorators from 'async_quart.py'.
+    :param inbound_data: auto-extracted by the decorators from 'async_quart.py'.
+    :param inbound_files: auto-extracted by the decorators from 'async_quart.py'.
+    :param log_id: An identifier for the logs (if logging is enabled).
+    :return: A standard response structure from the function in 'async_quart.py'.
+    """
+
+    try:
+
+        # Start by assuming failure:
+        is_valid = False
+        attempts_left = None
+
+        # Fetch the OTP from Redis:
+        otp_key = current_app.redis_cache.make_key(str(inbound_data.id))
+        stored_otp = await current_app.redis_cache.get(key = otp_key)
+
+        # Test the validity of the OTP:
+        if stored_otp is not None:
+
+            # Delete the record from the cache.:
+            await current_app.redis_cache.delete(key = otp_key)
+
+            # Make note of the attempts left.
+            # We reduce the count by one straightaway:
+            attempts_left = stored_otp["att"] - 1
+
+            # If the stored OTP matches the claimed OTP,
+            # we note down the acceptance and delete the record from the cache:
+            if inbound_data.otp == stored_otp["otp"]:
+                is_valid = True
+                attempts_left = 0
+
+            # If the stored OTP and the claimed OTP don't match,
+            # we reduce the attempt count and :
+            elif attempts_left > 0:
+
+                # Update the OTP info in the cache:
+                otp_updated = await current_app.redis_cache.set(
+                    key = otp_key,
+                    value = {
+                        "otp": stored_otp["otp"],
+                        "att": attempts_left,
+                        "iat": stored_otp["iat"],
+                        "sec": stored_otp["sec"]
+                    },
+                    expiry = (stored_otp["iat"] + stored_otp["sec"]) - date_time.get_current_utc_date_time().timestamp()
+                )
+
+                # If the update failed:
+                if not otp_updated: attempts_left = None
+
+        # Done here:
+        return ResponseModel(
+            api_version = api_version,
+            status_code = StatusCodes.OK if is_valid else StatusCodes.FAILED,
+            data = {
+                "isValid": is_valid,
+                "attemptsLeft": attempts_left
+            },
+            http_code = HttpCodes.UNAUTHORIZED
+        )
+
+    # In case the client terminates the connection prematurely:
+    except asyncio.CancelledError as exception:
+        current_app.printer(exception)
+        return ResponseModel(api_version = api_version, status_code = StatusCodes.CLIENT_CLOSED_REQUEST)
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                               MAIN PROGRAM                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+if __name__ == "__main__":
+
+    pass

controllers_v2/common/otp/timed_otp.py

@@ -0,0 +1,296 @@
+"""
+
+    AUTHOR:
+
+        Khushal P Soonderji
+
+    DATE:
+
+        Thursday, 19th 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("..")
+
+# My async utils:
+from utils_v2.date_time import date_time
+from utils_v2.security.otp import HashedOTP
+from utils_v2.database.async_mongo_v2 import AsyncMongo
+from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache
+
+# Controllers:
+from controllers_v2.message.sms.base import SMSController
+
+# Models:
+from models.core.auth_token import CoreAuthTokenModel
+from models.common.function_response import FunctionCallResponse
+from models.common.otp.timed_otp import GenerateTimedOTP, VerifyTimedOTP
+from models.message.sms.send import NimbusSMSIndiaMessage, SavvyBulkSMSKenyaMessage
+
+# To work with datatypes:
+from typing import List, Any
+
+# For debugging:
+from icecream import IceCreamDebugger
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                           MACROS / ONE-TIME INIT                                              ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                VARIABLES                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                  FUNCTIONS                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                   CLASSES                                                     ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+class TimedOTPController:
+
+    # ┏┓┓      ┓┏
+    # ┃ ┃┏┓┏┏  ┃┃┏┓┏┓┏
+    # ┗┛┗┗┻┛┛  ┗┛┗┻┛ ┛
+
+    KEY_PREFIX = "otp_"
+
+    # ┏┓
+    # ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓
+    # ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛
+
+    def __init__(
+            self,
+            debug: bool = True,
+            debug_prefix: str = "TOTP (C) | ",
+            debug_only_errors: bool = True
+    ):
+
+        """
+        This is a simple controller that will allow us to generate and verify timed OTPs.
+        :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.
+        """
+
+        # For debugging:
+        self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
+        if not debug: self._printer.disable()
+        self._debug_only_errors = debug_only_errors
+
+    def enable_terminal_print(self):
+        self._printer.enable()
+
+    def disable_terminal_print(self):
+        self._printer.disable()
+
+    def debug_only_errors(self):
+        self._debug_only_errors = True
+
+    def debug_everything(self):
+        self._debug_only_errors = False
+
+    # ┏┓┏┳┓┏┓  ┏┓           •
+    # ┃┃ ┃ ┃┃  ┃┓┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏┓
+    # ┗┛ ┻ ┣┛  ┗┛┗ ┛┗┗ ┛ ┗┻┗┗┗┛┛┗
+
+    async def generate(
+            self,
+            redis_cache: AsyncRedisCache,
+            inbound_data: GenerateTimedOTP
+    ) -> str | None:
+
+        """
+        To generate one OTP and store it in cache.
+        :param redis_cache: The instance of the cache client to use to hold the OTP.
+        :param inbound_data: The
+        :return:
+        """
+
+        # Generate the OTP:
+        otp_key = self.KEY_PREFIX + redis_cache.make_key(str(inbound_data.id))
+        otp_client = HashedOTP(secret = HashedOTP.generate_secret())
+        otp = otp_client.generate_otp(count = 0)
+
+        # Store the OTP in Redis:
+        otp_stored = await redis_cache.set(
+            key = otp_key,
+            value = {
+                "otp": otp,
+                "att": inbound_data.attempts,
+                "iat": date_time.get_current_utc_date_time().timestamp(),
+                "sec": inbound_data.seconds
+            },
+            expiry = inbound_data.seconds
+        )
+
+        # Done here:
+        return otp if otp_stored else None
+
+    async def generate_and_send_sms(
+            self,
+            mongo_data_conn: AsyncMongo,
+            redis_cache: AsyncRedisCache,
+            auth_token: CoreAuthTokenModel,
+            sms_client: SMSController,
+            sms_message: NimbusSMSIndiaMessage | SavvyBulkSMSKenyaMessage,
+            inbound_data: GenerateTimedOTP,
+            tags: List[str] = None
+    ) -> FunctionCallResponse:
+
+        """
+        Generate an OTP and send it via SMS.
+        :param mongo_data_conn: The client to use to connect to the database.
+        :param redis_cache: The client to use to connect to the cache.
+        :param auth_token: The credentials
+        :param sms_client:
+        :param sms_message:
+        :param inbound_data:
+        :param tags:
+        :return:
+        """
+
+        # Start by assuming failure:
+        response = FunctionCallResponse()
+
+        # Generate the OTP:
+        otp = await self.generate(redis_cache, inbound_data)
+        if not otp:
+            response.message = "Failed to generate OTP."
+            return response
+
+        # Send the SMS:
+        sms_response = await sms_client.send_many_sms(
+            mongo_data_conn = mongo_data_conn,
+            auth_token = auth_token,
+            messages = [sms_message],
+            tags = tags
+        )
+
+        # Analyze the actions:
+        success = True if sms_response.successCount == 1 else False
+        response.success = success
+        response.message = "OTP sent successfully." if success else "OTP could NOT be sent."
+
+        # Done here:
+        return response
+
+    # ┏┓┏┳┓┏┓  ┓┏    •┏•    •
+    # ┃┃ ┃ ┃┃  ┃┃┏┓┏┓┓╋┓┏┏┓╋┓┏┓┏┓
+    # ┗┛ ┻ ┣┛  ┗┛┗ ┛ ┗┛┗┗┗┻┗┗┗┛┛┗
+
+    async def verify(
+            self,
+            redis_cache: AsyncRedisCache,
+            inbound_data: VerifyTimedOTP
+    ) -> FunctionCallResponse:
+
+        """
+        To verify if an OTP is valid, or not.
+        :param redis_cache:
+        :param inbound_data:
+        :return:
+        """
+
+        # Start by assuming failure:
+        response = FunctionCallResponse()
+
+        # Fetch the OTP from Redis:
+        otp_key = self.KEY_PREFIX + redis_cache.make_key(str(inbound_data.id))
+        stored_otp = await redis_cache.get(key = otp_key)
+        if not stored_otp:
+            response.message = "No OTP found for this request. It may have expired."
+            return response
+
+        # Delete the record from the cache.:
+        await redis_cache.delete(key = otp_key)
+
+        # Make note of the attempts left.
+        # We reduce the count by one straightaway:
+        attempts_left = stored_otp["att"] - 1
+
+        # If the stored OTP matches the claimed OTP,
+        # we note down the acceptance and delete the record from the cache:
+        if inbound_data.otp == stored_otp["otp"]:
+            response.success = True
+            response.message = "OTP verified successfully."
+            return response
+
+        # If the stored OTP and the claimed OTP don't match,
+        # we reduce the attempt count and store the updated payload in cache:
+        elif attempts_left > 0:
+            otp_updated = await redis_cache.set(
+                key = otp_key,
+                value = {
+                    "otp": stored_otp["otp"],
+                    "att": attempts_left,
+                    "iat": stored_otp["iat"],
+                    "sec": stored_otp["sec"]
+                },
+                expiry = (stored_otp["iat"] + stored_otp["sec"]) - date_time.get_current_utc_date_time().timestamp()
+            )
+            if not otp_updated: attempts_left = 0
+
+        # Done here:
+        response.message = f"Invalid OTP. {attempts_left} attempts left."
+        return response
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                               MAIN PROGRAM                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+if __name__ == "__main__":
+
+    pass

cron/misc/reminders_for_work.py

@@ -307,9 +307,10 @@ async def get_reminders_to_send() -> pd.DataFrame | None:
 
         # Create the response DataFrame:
         if db_status == 1 and not exception:
-            reminders_json = db_json["data"].get("rs0", [])
+            reminders_json = db_json["data"].get("rs0")
             if reminders_json:
-                reminders_df = pd.DataFrame()
+                reminders_df = pd.DataFrame(reminders_json)
+                if reminders_df.empty: print("EMPTY!!")
                 reminders_df.rename(
                     columns = {
                         "token_id": "tokenKey",
@@ -514,7 +515,7 @@ async def send_reminders_once() -> None:
     """
 
     reminders_df = await get_reminders_to_send()
-    if reminders_df is not None:
+    if reminders_df is not None and not reminders_df.empty:
         tasks = [
             send_reminders_by_chat(reminders_df[reminders_df["serviceType"] == "chat"]),
             send_reminders_by_mail(reminders_df[reminders_df["serviceType"] == "mail"]),

models/common/function_response.py

@@ -0,0 +1,133 @@
+"""
+
+    AUTHOR:
+
+        Khushal P Soonderji
+
+    DATE:
+
+        Thursday, 20th Feb., 2025.
+
+    OBJECTIVE:
+
+        To give a general response structure to anything happening inside a function.
+
+    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
+from typing import Any
+
+# To work with MongoDB:
+from bson.objectid import ObjectId
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                           MACROS / ONE-TIME INIT                                              ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                VARIABLES                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                  FUNCTIONS                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+class FunctionCallResponse(BaseModel):
+
+    success: bool = Field(
+        description = "Whether, or not, the objective of the function was achieved.",
+        frozen = False,
+        default = False
+    )
+
+    message: str = Field(
+        description = "A brief message to indicate what happened in the process of executing the function.",
+        frozen = False,
+        default = "ERR: Message NOT captured."
+    )
+
+    exception: Exception = Field(
+        description = "Any exception that occurred in the process of execution.",
+        frozen = False,
+        default = None
+    )
+
+    data: Any = Field(
+        description = "Any data to be returned from the function that would be useful outside.",
+        frozen = False,
+        default = None
+    )
+
+    key: ObjectId = Field(
+        description = "the expendable reference to this auth; expose this to the ui",
+        frozen = True,
+        default_factory = lambda: ObjectId()
+    )
+
+    # ┏┓    ┏•
+    # ┃ ┏┓┏┓╋┓┏┓
+    # ┗┛┗┛┛┗┛┗┗┫
+    #          ┛
+
+    class Config:
+        extra = "forbid"
+        arbitrary_types_allowed = True
+
+    def model_dump(self, *args, **kwargs):
+        return super().model_dump(*args, by_alias = True, **kwargs)
+
+    # ┓┏  ┓• ┓   •
+    # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
+    # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
+
+    pass
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                               MAIN PROGRAM                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+if __name__ == "__main__":
+
+    pass

models/common/otp/timed_otp.py

@@ -0,0 +1,121 @@
+"""
+
+    AUTHOR:
+
+        Khushal P Soonderji
+
+    DATE:
+
+        Tuesday, 10th Sept., 2024.
+
+    OBJECTIVE:
+
+        To provide a data structure for the JSON received in the API calls to generate and verify timed OTPs.
+
+    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 models:
+from pydantic import BaseModel, Field, field_validator, Extra
+from typing import Optional, Any, Dict, List
+from typing_extensions import Annotated
+
+# My utils:
+from utils_v2.string import regex
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                           MACROS / ONE-TIME INIT                                              ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                VARIABLES                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                  FUNCTIONS                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+class GenerateTimedOTP(BaseModel):
+
+    id: str | Dict | List
+    attempts: Optional[int] = 3
+    seconds: Optional[int | float] = 30.0
+
+    class Config:
+        extra = "forbid"
+
+    def get(self, key: str, default = None):
+        return getattr(self, key, default)
+
+    @field_validator("attempts")
+    def validate_attempts(cls, value):
+        if not 1 <= value <= 10: raise ValueError("Attempts must be at least 1 and at most 10.")
+        return value
+
+    @field_validator("seconds")
+    def validate_expiry(cls, value):
+        if not 30 <= value <= 300: raise ValueError("Expiry must be at least 30s and at most 300s.")
+        return value
+
+
+# ---------------------------------------------------------------------------------------------------------------------
+
+
+class VerifyTimedOTP(BaseModel):
+
+    id: str | Dict | List
+    otp: str
+
+    class Config:
+        extra = "forbid"
+
+    def get(self, key: str, default = None):
+        return getattr(self, key, default)
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                               MAIN PROGRAM                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+if __name__ == "__main__":
+
+    pass