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

(20241125) Testing OAuth2.0 with Google.

Browse Source
This commit is contained in:
khushal
2024-11-25 14:43:26 +05:30
parent 1519ee00c1
commit 285bee70f3
7 changed files with 441 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils_v2/oauth/base.py
+38
View File
@@ -41,6 +41,9 @@ import io
# For defining the class's structure:
from abc import ABC, abstractmethod
# For working with datatypes:
from typing import List
# For debugging:
from icecream import IceCreamDebugger
@@ -115,6 +118,41 @@ class OAuthBase(ABC):
def debug_everything(self):
self._debug_only_errors = False
# ┏┓┓ ┳┳┓ ┓ ┓
# ┣┫┣┓┏╋┏┓┏┓┏╋ ┃┃┃┏┓╋┣┓┏┓┏┫┏
# ┛┗┗┛┛┗┛ ┗┻┗┗ ┛ ┗┗ ┗┛┗┗┛┗┻┛
@abstractmethod
async def initialize(
self,
scopes: List
) -> bool:
"""
To initialize the service-specific OAuth2.0 class. For example, in Google's case, we need to initialize an app
flow that was created for an app through its Cloud Console panel.
:param scopes: The list of permissions being requested. The word 'scopes' has been borrowed from Google's OAuth
documentation (which was implemented first).
:return: True if the initialization succeeded, False if it failed.
"""
pass
async def get_authorization_url(
self,
**kwargs
) -> str | None:
"""
To create an authorization URL which will be then sent to the front-end for the user to click and grant/decline
various permissions.
:param kwargs: The identifiers of the user who wants to use your service (where your service needs access to
their second-party account).
:return: The authorization URL if successful, or None if failed.
"""
pass
# *****************************************************************************************************************
# ***** ****
utils_v2/oauth/services/goog.py
+214 -19
View File
@@ -15,6 +15,8 @@
REFERENCES:
1. https://developers.google.com/calendar/api/quickstart/python
2. https://developers.google.com/identity/protocols/oauth2/web-server#python
3. https://www.youtube.com/watch?v=vQQEaSnQ_bs&t=940s&pp=ygUVb2F1dGgyIHB5dGhvbiB5b3V0dWJl
DOWNLOADS:
@@ -40,6 +42,7 @@ import io
# My utils:
from utils_v2.string import json
from utils_v2.date_time import date_time
# The base model:
from utils_v2.oauth.base import OAuthBase
@@ -52,6 +55,12 @@ from google_auth_oauthlib.flow import InstalledAppFlow
# For asynchronous activities:
import asyncio
# To make deep-copies:
import copy
# To work with date and time:
import datetime
# *****************************************************************************************************************
# ***** ****
@@ -100,28 +109,191 @@ class GoogleOAuth(OAuthBase):
# Class variables:
__flow = None
async def init(
# ┏┓┓ ┳┳┓ ┓ ┓
# ┣┫┣┓┏╋┏┓┏┓┏╋ ┃┃┃┏┓╋┣┓┏┓┏┫┏
# ┛┗┗┛┛┗┛ ┗┻┗┗ ┛ ┗┗ ┗┛┗┗┛┗┻┛
async def initialize(
self,
scopes = None
):
scopes,
raise_exception = False
) -> bool:
# Initialize your Google App:
self._printer("Initializing flow.")
self.__flow = InstalledAppFlow.from_client_config(
self._config,
scopes = scopes or SCOPES_GMAIL_MAIL_MANAGEMENT,
redirect_uri = self._redirect_url
)
"""
Initialize the Google OAuth mechanism by creating an app-flow. This defines the app that you are trying to
deploy. You must create this app in Google's Cloud Platform's console.
:param scopes: The scopes (permissions) needed by this app.
:param raise_exception: If set to True, any exception that occurs will be propagated. If set to false, any
exception that occurs will be suppressed.
:return: True if the initialization succeeded, False if it failed.
"""
async def get_authorization_url(self):
# Start by assuming success:
success = True
authorization_url, state = self.__flow.authorization_url(
access_type = "offline",
include_granted_scopes = "true"
)
try:
# Initialize your Google App:
self._printer("Initializing flow.")
self.__flow = InstalledAppFlow.from_client_config(
self._config,
scopes = scopes,
redirect_uri = self._redirect_url
)
# If something goes wrong:
except Exception as exception:
self._printer(exception)
if raise_exception: raise
success = False
# Done here:
return success
async def get_authorization_url(
self,
raise_exception = False,
**kwargs,
) -> str | None:
"""
To generate an authorization URL that can be sent to the front end. When the
:param raise_exception: If set to True, any exception that occurs will be propagated. If set to false, any
exception that occurs will be suppressed.
:param kwargs: Any no. of keyword args that you might want to give to this specific service.
:return:
"""
# Start by assuming failure:
authorization_url = None
try:
# Request a URL that will be sent to the user to request
# permissions to access their account:
authorization_url, _ = self.__flow.authorization_url(
access_type = kwargs.get("access_type", "offline"),
approval_prompt = kwargs.get("approval_prompt", "force"),
include_granted_scopes = kwargs.get("include_granted_scopes", "true"),
login_hint = kwargs.get("email"),
state = kwargs.get("user_id")
)
# Done here:
return authorization_url
# If something goes wrong:
except Exception as exception:
self._printer(exception)
if raise_exception: raise
authorization_url = None
# Done here:
return authorization_url
async def get_tokens(
self,
raise_exception = False,
**kwargs
) -> Credentials | dict | None:
# Start by assuming failure:
tokens = None
try:
# Fetch the tokens:
credentials = self.__flow.fetch_token(authorization_response = kwargs["redirect_url"])
tokens = {
"access_token": credentials.get("access_token"),
"refresh_token": credentials.get("refresh_token"),
"expires_in": (ttl := credentials["expires_in"] - 60),
"expires_at": date_time.get_current_utc_date_time() + datetime.timedelta(seconds = ttl),
"scopes": credentials.get("scope"),
}
# If something goes wrong:
except Exception as exception:
self._printer(exception)
if raise_exception: raise
tokens = None
# Done here:
return tokens
async def refresh_tokens(
self,
old_tokens: dict,
raise_exception = False
) -> dict | None:
# Start by assuming failure:
tokens = None
try:
# If the tokens haven't expired, just return the existing tokens back:
tokens_expired = True if date_time.get_current_utc_date_time() >= old_tokens["expires_at"] else False
if not tokens_expired: return old_tokens
# Construct the credentials and request a refresh:
credentials = await self.credentials_from_tokens(old_tokens)
if tokens_expired and credentials.refresh_token:
credentials.refresh(Request())
tokens = {
"access_token": credentials.token,
"refresh_token": credentials.refresh_token,
"expires_at": (exp_at := date_time.as_if_timezone(
credentials.expiry,
timezone = date_time.TIMEZONE_UTC
)),
"expires_in": (exp_at - date_time.get_current_utc_date_time()).total_seconds(),
"scopes": credentials.scopes,
}
# If something goes wrong:
except Exception as exception:
self._printer(exception)
if raise_exception: raise
tokens = None
# Done here:
return tokens
# ┏┓ • ┏┓ •┏•
# ┗┓┏┓┏┓┓┏┓┏┏┓ ┗┓┏┓┏┓┏┓╋┓┏
# ┗┛┗ ┛ ┗┛┗┗┗ ┗┛┣┛┗ ┗┗┛┗┗
# ┛
async def credentials_from_tokens(
self,
tokens: dict,
raise_exception = False
) -> Credentials | None:
# Start by assuming failure:
credentials = None
try:
# Make a deep-copy and add some fields from the config:
tokens_copy = copy.deepcopy(tokens)
first_key = list(self._config.keys())[0]
tokens_copy["client_id"] = self._config.get(first_key, {}).get("client_id")
tokens_copy["client_secret"] = self._config.get(first_key, {}).get("client_secret")
# Create the credentials:
credentials = Credentials.from_authorized_user_info(info = tokens_copy)
# If something goes wrong:
except Exception as exception:
self._printer(exception)
if raise_exception: raise
credentials = None
# Done here:
return credentials
# *****************************************************************************************************************
# ***** ****
@@ -132,21 +304,44 @@ class GoogleOAuth(OAuthBase):
if __name__ == "__main__":
import dateparser
secrets_file = r"../../../creds/google_converse_test_oauth.json"
secrets_dict = json.from_file(secrets_file)
my_goog = GoogleOAuth(
config = secrets_dict,
# redirect_url = r"https://v2.api.bicree.com/user/callback/test",
redirect_url = r"127.0.0.1",
redirect_url = r"https://nexcom.ditscentre.in/converse/test/callback",
debug = True,
debug_prefix = "OAuth (Goog) | "
)
async def main():
await my_goog.init()
print("AUTH URL:", await my_goog.get_authorization_url())
await my_goog.initialize(scopes = SCOPES_GMAIL_MAIL_MANAGEMENT)
# await asyncio.sleep(1.0)
# print("AUTH URL 0:", await my_goog.get_authorization_url(
# user_id = "BHOPLI",
# # email = "pskhushal@gmail.com"
# ))
#
# redirect_url = input("Paste the redirect URL here: ")
# print("TOKENS:", json.to_string(await my_goog.get_tokens(redirect_url = redirect_url), default = str))
old_tok = {
"access_token": "ya29.a0AeDClZAjmLUZ1hh0aTbddz4ThjRzwQNMwdi_H4AYO-C4ETvWpK8aHYz5eV9PTUlFDJKQoHYtFgu4u2XfoiOdIEMVNwAKFbb8sakIWef7Yk5HdiDYC0A-MUp5XZnoNGLuP_GW_O3IxfCLH7cC0fb3AfHx4OsBpa_Qu1X-IpvfaCgYKAaoSARMSFQHGX2MiX81qoRex393eWE229OMCUg0175",
"refresh_token": "1//0g4mVQrydnc1ECgYIARAAGBASNgF-L9Irk_UxcRjgkz_YyK5Ujs1qCaj8nKL7bqQ0jHnHYlpVRh_Pwm77X4Angp8R-o-fgmKfzg",
"expires_in": 3539,
"expires_at": dateparser.parse("2024-11-25 09:24:56.690876+00:00"),
"token_type": "Bearer",
"scopes": [
"https://www.googleapis.com/auth/gmail.labels",
"https://www.googleapis.com/auth/gmail.modify"
]
}
new_tok = await my_goog.refresh_tokens(old_tokens = old_tok)
print("NEW TOKENS:", json.to_string(new_tok, default = str))
asyncio.run(main())