Resetting utils subtree.
This commit is contained in:
@@ -1,475 +0,0 @@
|
||||
"""
|
||||
|
||||
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.data.api_call import GoogleApiResponse
|
||||
from utils_v2.goog.models.data.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 = "GMail | ",
|
||||
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
|
||||
|
||||
# ┏┓┏┓ ┓ ┏┓ ┏┓
|
||||
# ┃┃┣┫┓┏╋┣┓ ┏┛ ┃┫
|
||||
# ┗┛┛┗┗┻┗┛┗ ┗━•┗┛
|
||||
|
||||
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],
|
||||
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
|
||||
@@ -1,132 +0,0 @@
|
||||
"""
|
||||
|
||||
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
|
||||
@@ -1,239 +0,0 @@
|
||||
"""
|
||||
|
||||
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 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
|
||||
Reference in New Issue
Block a user