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

Merge commit 'bfbc270b655b1db87dfd9e3afaf79d9968042da5' as 'utils_v2'

Browse Source
This commit is contained in:
yatmesh
2025-06-12 13:46:57 +05:30
parent c03e75d30e bfbc270b65
commit 83a9c7cf89
217 changed files with 149785 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils_v2/goog/controllers/base.py
+479
View File
@@ -0,0 +1,479 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Tuesday, 26th Nov., 2024
OBJECTIVE:
To provide a base class for common behaviour of Google's APIs.
REFERENCES:
1. Quickstart: https://developers.google.com/gmail/api/quickstart/python
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# System-level activities:
import io
# My utils:
from utils_v2.date_time import date_time
from utils_v2.goog.models.api_call import GoogleApiResponse
from utils_v2.goog.models.auth_tokens import GoogleAuthTokens
# Related to Google:
from google_auth_oauthlib.flow import InstalledAppFlow
# To make API calls:
import httpx
# To work with date and time:
import datetime
# For working with datatypes:
from typing import Literal, List
# For debugging:
from icecream import IceCreamDebugger
import inspect
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class AsyncGoogleBase:
def __init__(
self,
service_name: str,
oauth_json: dict,
http_client: httpx.AsyncClient,
redirect_url: str = None,
debug = True,
debug_prefix = "Google | ",
debug_only_errors = True
):
"""
To initialize any Google API from one base class. The client's id and secret are available in the file
downloaded form https://console.cloud.google.com/apis/credentials (do not forget to select your app).
:param service_name: A string to identify this service.
:param oauth_json: The OAuth credentials downloaded from https://console.cloud.google.com/apis/credentials
:param http_client: An asynchronous HTTP client to make API calls.
:param redirect_url: Where you would like to receive the confirmation of the user authorization.
:param debug: Whether, or not, you would like to show debugging messages on the terminal.
:param debug_prefix: The prefix string to identify the debugging messages.
:param debug_only_errors: Whether you would like to show all debugging messages or just error messages.
"""
# Prepare the debugging utility:
self._debug_prefix = debug_prefix
self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
if not debug: self._printer.disable()
self._debug_only_errors = debug_only_errors
# Accept the input configuration:
self._service_name = service_name
self._http_client = http_client
self._oauth_json = oauth_json
self._client_id = self._oauth_json["web"]["client_id"]
self._client_secret = self._oauth_json["web"]["client_secret"]
self._redirect_url = redirect_url
def enable_debug(self):
self._printer.enable()
def disable_debug(self):
self._printer.disable()
def debug_only_errors(self):
self._debug_only_errors = True
def debug_everything(self):
self._debug_only_errors = False
# ┏┓ •
# ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏
# ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛
# ┛
@property
def client_id(self):
return self._client_id
@property
def client_secret(self):
return self._client_secret
@property
def http_client(self):
return self._http_client
# ┏┓┏┓ ┓ ┏┓ ┏┓
# ┃┃┣┫┓┏╋┣┓ ┏┛ ┃┫
# ┗┛┛┗┗┻┗┛┗ ┗━•┗┛
async def get_authorization_url(
self,
scopes: List[str],
state: str = None,
access_type: Literal["online", "offline"] = "offline",
approval_prompt: Literal["auto", "force", "consent"] = "auto",
include_granted_scopes: Literal["true", "false"] = "true",
user_email: str = None
) -> str:
"""
TO get the OAuth2.0 authorization URL for one user.
DOCUMENTATION:
1. https://developers.google.com/identity/protocols/oauth2/web-server
:param scopes: The set of permission you want the user to give.
:param state: A unique identifier for your user. If not supplied, a random string will be generated.
:param access_type: Set the value to offline if your application needs to refresh access tokens when the user is
not present at the browser.
:param approval_prompt: "force" ensures that the consent screen is always shown to the user, regardless of
whether the user has previously granted consent for the requested scopes. It forces the user to re-approve
the app's access, which can be useful if the app is requesting new permissions or if the consent needs to be
explicitly confirmed. "consent" ensures the user's consent is required if they haven't approved the app's
requested permissions yet. "auto" allows Google to automatically determine whether the consent screen should
be shown.
:param include_granted_scopes: Enables applications to use incremental authorization to request access to
additional scopes in context. If you set this parameter's value to true and the authorization request is
granted, then the new access token will also cover any scopes to which the user previously granted the
application access.
:param user_email:
:return:
"""
# Create a flow:
flow = InstalledAppFlow.from_client_config(
self._oauth_json,
scopes = scopes,
redirect_uri = self._redirect_url
)
# Get an authorization URL:
auth_url, state = flow.authorization_url(
access_type = access_type,
approval_prompt = approval_prompt,
include_granted_scopes = include_granted_scopes,
login_hint = user_email,
state = state
)
# Done here:
return auth_url
async def get_authorization_tokens(
self,
scopes: List[str] | None,
redirect_url: str
) -> GoogleAuthTokens:
"""
When the user accepts or declines an authorization request, Google sends you an alert on your redirect URL. Pass
the URL as it is to this method to generate the authorization tokens that you can store in the database and
reuse for this user's activities.
:param scopes: The set of permissions the user granted.
:param redirect_url: The exact URL that was hit (with the query params) that Google hit when the user did
something on your authorization URL. Fortunately, this URL is readily available in Quart and Flask by
calling 'request.url'.
:return: The authorization tokens.
"""
# Create a flow:
flow = InstalledAppFlow.from_client_config(
self._oauth_json,
scopes = scopes,
redirect_uri = self._redirect_url
)
# Get the credentials:
credentials = flow.fetch_token(authorization_response = redirect_url)
ttl = credentials["expires_in"] - 60
return GoogleAuthTokens(
accessToken = credentials["access_token"],
refreshToken = credentials["refresh_token"],
expiresAt = date_time.get_current_utc_date_time() + datetime.timedelta(seconds = ttl),
scopes = credentials["scope"]
)
# ┏┓ ┳┓ ┓•
# ┣ ┏┓┏┓┏┓┏┓ ┃┃┏┓┏┏┓┏┫┓┏┓┏┓
# ┗┛┛ ┛ ┗┛┛ ┻┛┗ ┗┗┛┗┻┗┛┗┗┫
# ┛
@staticmethod
async def __get_error_message(api_response: GoogleApiResponse) -> str:
"""
To extract various kinds of error messages from Google's responses.
:param api_response: The formatted response from the API call.
:return: The message string.
"""
try: return (await api_response.get_json())["error"]["message"]
except: return api_response.response.reason_phrase
# ┏┓┏┓┳ ┏┓ ┓┓•
# ┣┫┃┃┃ ┃ ┏┓┃┃┓┏┓┏┓
# ┛┗┣┛┻ ┗┛┗┻┗┗┗┛┗┗┫
# ┛
async def get(
self,
url: str,
headers: dict = None,
params: dict = None
) -> GoogleApiResponse:
"""
To call an API using the GET method.
:param url: The URL to call.
:param headers: The headers to pass.
:param params: The params to send in the query string itself.
:return: A structured response that includes the raw response, the exception (if any), and so on.
"""
# Prepare the structure of the response:
api_response = GoogleApiResponse(
serviceName = self._service_name,
action = inspect.stack()[1].function,
url = url,
method = "GET"
)
try:
# Make the API call:
response = await self._http_client.get(
url = url,
headers = headers,
params = params
)
# Note down the results:
api_response.response = response
api_response.httpCode = response.status_code
api_response.message = await self.__get_error_message(api_response)
# If something goes wrong:
except Exception as exception:
api_response.exception = exception
api_response.message = str(exception)
self._printer(exception, api_response.url, api_response.method, headers, params)
# Done here:
return api_response
async def post(
self,
url: str,
headers: dict = None,
json: dict = None,
data: dict = None,
content: str | bytes = None
) -> GoogleApiResponse:
"""
To call an API using the POST method.
:param url: The URL to call.
:param headers: The headers to pass.
:param json: The params to send in the JSON body.
:param data: The params to send in the form-data in the body.
:param content: The raw content to be sent in the body (typically as an octet-stream).
:return: A structured response that includes the raw response, the exception (if any), and so on.
"""
# Prepare the structure of the response:
api_response = GoogleApiResponse(
serviceName = self._service_name,
action = inspect.stack()[1].function,
url = url,
method = "POST"
)
try:
# Make the API call:
response = await self._http_client.post(
url = url,
headers = headers,
json = json,
data = data,
content = content
)
# Note down the results:
api_response.response = response
api_response.httpCode = response.status_code
api_response.message = await self.__get_error_message(api_response)
# If something goes wrong:
except Exception as exception:
api_response.exception = exception
api_response.message = str(exception)
self._printer(exception, api_response.url, api_response.method, headers, json, data)
# Done here:
return api_response
async def put(
self,
url: str,
headers: dict = None,
json: dict = None,
data: dict = None
) -> GoogleApiResponse:
"""
To call an API using the PUT method.
:param url: The URL to call.
:param headers: The headers to pass.
:param json: The params to send in the JSON body.
:param data: The params to send in the form-data in the body.
:return: A structured response that includes the raw response, the exception (if any), and so on.
"""
# Prepare the structure of the response:
api_response = GoogleApiResponse(
serviceName = self._service_name,
action = inspect.stack()[1].function,
url = url,
method = "PUT"
)
try:
# Make the API call:
response = await self._http_client.put(
url = url,
headers = headers,
json = json,
data = data
)
# Note down the results:
api_response.response = response
api_response.httpCode = response.status_code
api_response.message = await self.__get_error_message(api_response)
# If something goes wrong:
except Exception as exception:
api_response.exception = exception
api_response.message = str(exception)
self._printer(exception, api_response.url, api_response.method, headers, json, data)
# Done here:
return api_response
async def delete(
self,
url: str,
headers: dict = None
) -> GoogleApiResponse:
"""
To call an API using the DELETE method.
:param url: The URL to call.
:param headers: The headers to pass.
:return: A structured response that includes the raw response, the exception (if any), and so on.
"""
# Prepare the structure of the response:
api_response = GoogleApiResponse(
serviceName = self._service_name,
action = inspect.stack()[1].function,
url = url,
method = "DELETE"
)
try:
# Make the API call:
response = await self._http_client.delete(
url = url,
headers = headers
)
# Note down the results:
api_response.response = response
api_response.httpCode = response.status_code
api_response.message = await self.__get_error_message(api_response)
# If something goes wrong:
except Exception as exception:
api_response.exception = exception
api_response.message = str(exception)
self._printer(exception, api_response.url, api_response.method, headers)
# Done here:
return api_response
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass