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

(20241125) Documentation added.

Browse Source
This commit is contained in:
khushal
2024-11-25 15:31:30 +05:30
parent 6f5c3f11a1
commit b5a4f0cd52
3 changed files with 104 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils_v2/oauth/services/goog.py
+52 -16
View File
@@ -61,6 +61,9 @@ import copy
# To work with date and time:
import datetime
# For working with datatypes:
from typing import List
# *****************************************************************************************************************
# ***** ****
@@ -115,7 +118,7 @@ class GoogleOAuth(OAuthBase):
async def initialize(
self,
scopes,
scopes: List,
raise_exception = False
) -> bool:
@@ -158,7 +161,7 @@ class GoogleOAuth(OAuthBase):
"""
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
: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:
@@ -195,7 +198,18 @@ class GoogleOAuth(OAuthBase):
self,
raise_exception = False,
**kwargs
) -> Credentials | dict | None:
) -> dict | None:
"""
To get the tokens of a user. Plural 'tokens' because OAuth typically has one access token that expires
every-so-often, and one refresh token that grants you a new access token. In Google's case, their servers hit
your callback URL with some query params. You must furnish this whole URL right here under the kwarg
'redirect_url'. Fortunately, this URL is readily available in Quart and Flask by calling 'request.url'.
:param kwargs: The identifiers of the user who wants to use your service and any other service-specific options.
: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: The tokens for the service if successful, or None if failed.
"""
# Start by assuming failure:
tokens = None
@@ -224,9 +238,21 @@ class GoogleOAuth(OAuthBase):
async def refresh_tokens(
self,
old_tokens: dict,
force_refresh = False,
raise_exception = False
) -> dict | None:
"""
To refresh the tokens of a user. Plural 'tokens' because OAuth typically has one access token that expires
every-so-often, and one refresh token that grants you a new access token.
:param old_tokens: The current set of tokens as obtained by 'get_tokens'.
:param force_refresh: To force a refresh request even if the tokens haven't yet expired.
: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: The same tokens if they haven't expired, refreshed tokens if the tokens have expired and were
successfully refreshed, None if the tokens have expired but could not be refreshed.
"""
# Start by assuming failure:
tokens = None
@@ -234,11 +260,11 @@ class GoogleOAuth(OAuthBase):
# 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
if not tokens_expired and not force_refresh: 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:
if credentials.refresh_token:
credentials.refresh(Request())
tokens = {
"access_token": credentials.token,
@@ -271,6 +297,16 @@ class GoogleOAuth(OAuthBase):
raise_exception = False
) -> Credentials | None:
"""
This is a very service-specific thing. When using Google's APIs, you will need to use their native 'Credentials'
object. This method allows you to quickly convert the output of 'get_tokens' and 'refresh_tokens' to that object
for easy use.
:param tokens: The output of either 'get_tokens' or 'refresh_tokens'.
: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: Google's API's native 'Credentials' object if successful, else None.
"""
# Start by assuming failure:
credentials = None
@@ -311,7 +347,7 @@ if __name__ == "__main__":
my_goog = GoogleOAuth(
config = secrets_dict,
redirect_url = r"https://thecaoffice.com/nexcom/converse/mail/callback/gmail",
redirect_url = r"https://api.thecaoffice.com/converse/mail/callback/gmail",
debug = True,
debug_prefix = "OAuth (Goog) | "
)
@@ -324,24 +360,24 @@ if __name__ == "__main__":
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))
tokens = await my_goog.get_tokens(redirect_url = redirect_url)
print("TOKENS:", json.to_string(tokens, 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",
# # To test refreshing:
# tokens = {
# "access_token": "ya29.a0AeDClZAYoo85BXRId_n-hwo_amKshzi46c33GaJcsZZvGB7A7OGU2RFYcWBM_BleNBfAFUSJP2NHAvmd7Nsp_U5Kg68hXSy0iO99PNTm3pvKrJSzbkA-rXsVLsCnBIfPUMyNt2nOOVJmGwm17DNN0jAELkm1fPNTju7SZzmuaCgYKAZwSARMSFQHGX2Mis8TZui2rZT1gKySVds-N0w0175",
# "refresh_token": "1//0gnqzjMf9YT19CgYIARAAGBASNgF-L9Ir3rcY37nGrV45XyOUBRllEH7Txui7T1JbwevlmDoNw7PuMu149cCWQSwsScuKaZusUQ",
# "expires_in": 3539,
# "expires_at": dateparser.parse("2024-11-25 09:24:56.690876+00:00"),
# "token_type": "Bearer",
# "expires_at": dateparser.parse("2024-11-25 10:40:40.833699+00:00"),
# "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))
new_tok = await my_goog.refresh_tokens(old_tokens = tokens, force_refresh = False)
print("REFRESHED TOKENS:", json.to_string(new_tok, default = str))
asyncio.run(main())