Merge commit '0ab1241a4e25b3fa49ca48926bc1d0b288fdf7fd' as 'utils_v2'

2024-12-21 10:08:39 +05:30
parent 32aedc8a25 0ab1241a4e
commit eb29878df6
164 changed files with 136594 additions and 0 deletions
@@ -0,0 +1,132 @@
+"""
+
+    AUTHOR:
+
+        Khushal P Soonderji
+
+    DATE:
+
+        Wednesday, 30th Oct., 2024.
+
+    OBJECTIVE:
+
+        To provide a data model for describing the API response from Google's APIs.
+
+    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, model_validator
+from typing import Optional, Literal, Union, Dict, List, Any
+
+# My utils:
+from utils_v2.string import json
+from utils_v2.string import regex
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                           MACROS / ONE-TIME INIT                                              ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                VARIABLES                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                  FUNCTIONS                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+class GoogleApiResponse(BaseModel):
+
+    serviceName: str = Field(frozen = True, default = None)
+    action: str = Field(frozen = True, default = None)
+
+    url: str = Field(frozen = True)
+    method: str = Field(frozen = True)
+    response: Any = None
+    httpCode: int = None
+
+    success: bool = False
+    message: str = None
+    data: Any = None
+
+    exception: Any = None
+
+    # ┏┓    ┏•
+    # ┃ ┏┓┏┓╋┓┏┓
+    # ┗┛┗┛┛┗┛┗┗┫
+    #          ┛
+
+    class Config:
+        extra = "forbid"
+
+    # ┏┓           ┏┓
+    # ┃ ┓┏┏╋┏┓┏┳┓  ┣ ┓┏┏┓┏┏
+    # ┗┛┗┻┛┗┗┛┛┗┗  ┻ ┗┻┛┗┗┛
+
+    def to_markdown(self):
+        if self.exception: message = "❌ *GOOGLE API EXCEPTION:* ❌\n\n"
+        else: message = "*GOOGLE API RESPONSE:*\n\n"
+        message += f"*SERVICE:*\n`{self.serviceName}`\n\n"
+        message += f"*ACTION:*\n`{self.action}`\n\n"
+        message += f"*URL:*\n`{self.url}`\n\n"
+        message += f"*METHOD:*\n`{self.method}`\n\n"
+        message += f"*RESPONSE:*\n`{self.response}`\n\n"
+        message += f"*MESSAGE:*\n`{self.message}`\n\n"
+        message += f"*EXCEPTION:*\n`{self.exception.__class__.__name__}: {str(self.exception)}`\n\n"
+        return message
+
+    async def get_json(self):
+        try: return self.response.json()
+        except: return {}
+
+    async def get_content(self):
+        try: return self.response.content
+        except: return b""
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                               MAIN PROGRAM                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+if __name__ == "__main__":
+
+    pass
@@ -0,0 +1,243 @@
+"""
+
+    AUTHOR:
+
+        Khushal P Soonderji
+
+    DATE:
+
+        Wednesday, 30th Oct., 2024.
+
+    OBJECTIVE:
+
+        To provide a data model for describing the tokens to be used for Google's APIs.
+
+    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, model_validator, AwareDatetime
+from typing import Optional, Literal, Union, Dict, List, Any
+
+# Related to Google:
+from google.auth.transport.requests import Request
+from google.oauth2.credentials import Credentials
+
+# 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
+import dateparser
+
+# To make API calls:
+import httpx
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                           MACROS / ONE-TIME INIT                                              ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                VARIABLES                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                  FUNCTIONS                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+class GoogleAuthTokens(BaseModel):
+
+    accessToken: str = Field(description = "the main 'bearer' token")
+    refreshToken: str = Field(description = "token to be used to refresh the access token")
+    expiresAt: AwareDatetime = Field(description = "the time (utc) at which the token will expire")
+    scopes: List[str] = Field(description = "the list of permissions", default = [])
+    email: str | None = Field(description = "the email id of the user", default = None)
+    displayName: str | None = Field(description = "the display name of the user", default = None)
+    displayPictureUrl: str | None = Field(description = "the url to the display picture of the user", default = None)
+
+    # ┏┓    ┏•
+    # ┃ ┏┓┏┓╋┓┏┓
+    # ┗┛┗┛┛┗┛┗┗┫
+    #          ┛
+
+    class Config:
+        extra = "allow"
+
+    # ┓┏  ┓• ┓   •
+    # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
+    # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
+
+    @field_validator("expiresAt", mode = "before")
+    def parse_dates(cls, value):
+        if not isinstance(value, datetime.datetime):
+            parsed = date_time.parse_date_time(value, date_formats = ["%Y%m%d", "%Y-%m-%d"])
+            value = parsed if isinstance(parsed, datetime.datetime) else dateparser.parse(value)
+        if isinstance(value, datetime.datetime): value = date_time.to_timezone(value, date_time.TIMEZONE_UTC)
+        return value
+
+    # ┏┓           •
+    # ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏
+    # ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛
+    #       ┛
+
+    @property
+    def client_user_id(self):
+        return {"email": self.email}
+
+    @property
+    def expired(self):
+        return True if date_time.get_current_utc_date_time() >= self.expiresAt else False
+
+    @property
+    def ttl(self):
+        return (self.expiresAt - date_time.get_current_utc_date_time()).total_seconds()
+
+    # ┏┓           ┏┓
+    # ┃ ┓┏┏╋┏┓┏┳┓  ┣ ┓┏┏┓┏┏
+    # ┗┛┗┻┛┗┗┛┛┗┗  ┻ ┗┻┛┗┗┛
+
+    def refresh(
+            self,
+            client_id: str,
+            client_secret: str,
+            force_refresh: bool = False
+    ) -> bool:
+
+        """
+        Synchronously (blocking) refreshes the existing access tokens in place.
+        :param client_id: The id of the client app (OAuth JSON) for which these tokens were granted.
+        :param client_secret: The secret of the client app (OAuth JSON) for which these tokens were granted.
+        :param force_refresh: Whether you want to refresh the token even if it hasn't expired.
+        :return: True if refreshed, else False.
+        """
+
+        # Start by assuming failure:
+        success = False
+
+        try:
+
+            # Go ahead only if either the token has expired,
+            # or the user has asked to forcefully refresh the tokens:
+            if self.expired or force_refresh:
+
+                # Create the credentials:
+                credentials = Credentials.from_authorized_user_info(
+                    info = {
+                        "client_id": client_id,
+                        "client_secret": client_secret,
+                        "refresh_token": self.refreshToken,
+                        "expires_at": self.expiresAt
+                    }
+                )
+
+                # Request a refresh:
+                credentials.refresh(Request())
+
+                # Note down the new credentials:
+                if credentials.token != self.accessToken:
+                    success = True
+                    self.accessToken = credentials.token
+                    self.refreshToken = credentials.refresh_token
+                    self.expiresAt = date_time.as_if_timezone(credentials.expiry, timezone = date_time.TIMEZONE_UTC)
+
+        # In case something goes wrong:
+        except Exception as exception:
+            success = False
+
+        # Done here:
+        return success
+
+    async def arefresh(
+            self,
+            http_client: httpx.AsyncClient,
+            client_id: str,
+            client_secret: str,
+            force_refresh: bool = False
+    ) -> bool:
+
+        """
+        Asynchronously refreshes the existing access tokens in place.
+        :param http_client: The HTTP client to use to make the refresh request.
+        :param client_id: The id of the client (OAuth JSON) for which these tokens were granted.
+        :param client_secret: The secret of the client (OAuth JSON) for which these tokens were granted.
+        :param force_refresh: Whether you want to refresh the token even if it hasn't expired.
+        :return: True if refreshed, else False.
+        """
+
+        # Currently we don't really know how to refresh tokens through low-level API calls,
+        # so we will pass on the intent to the regular, synchronous function.
+        return self.refresh(
+            client_id = client_id,
+            client_secret = client_secret,
+            force_refresh = force_refresh
+        )
+
+    def has_scopes(self, scopes: List[str]) -> bool:
+
+        """
+        Checks if all the specified scopes were granted.
+        :param scopes: The list of scopes to check. These are the permissions you need.
+        :return: True if all specified scoped are present, else False.
+        """
+
+        # Start by assuming success:
+        has_scopes = True
+
+        # Now loop through the needed scopes and check:
+        for scope in scopes:
+            if scope not in self.scopes:
+                has_scopes = False
+                break
+
+        # Done here:
+        return has_scopes
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                               MAIN PROGRAM                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+if __name__ == "__main__":
+
+    pass