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

(20241125) GMail Client Started. Label-Management Ready.

Browse Source
This commit is contained in:
khushal
2024-11-25 19:21:33 +05:30
parent bccf61459e
commit ed415c6b3a
3 changed files with 622 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils_v2/oauth/services/goog.py
+40 -19
View File
@@ -110,6 +110,7 @@ SCOPES_GMAIL_FULL = [r"https://mail.google.com/"]
class GoogleOAuth(OAuthBase):
# Class variables:
service_name = "google"
__flow = None
# ┏┓┓ ┳┳┓ ┓ ┓
@@ -291,6 +292,25 @@ class GoogleOAuth(OAuthBase):
# ┗┛┗ ┛ ┗┛┗┗┗ ┗┛┣┛┗ ┗┗┛┗┗
# ┛
@staticmethod
async def credentials_have_expired(credentials: Credentials) -> bool:
"""
Checks is a 'Credentials' object has expired or not. The native mechanism has some complication with timezone
considerations. This method was created to overcome that bug.
:param credentials: The 'Credentials' object of the user.
:return: True if the credentials have expired, False if they're yet valid.
"""
# Take the timezone into account:
expires_at = date_time.as_if_timezone(
credentials.expiry,
timezone = date_time.TIMEZONE_UTC
)
# Return the assessment based on the evaluation with the timezone considered:
return True if date_time.get_current_utc_date_time() >= expires_at else False
async def credentials_from_tokens(
self,
tokens: dict,
@@ -355,26 +375,27 @@ if __name__ == "__main__":
async def main():
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: ")
tokens = await my_goog.get_tokens(redirect_url = redirect_url)
print("TOKENS:", json.to_string(tokens, default = str))
# await asyncio.sleep(1.0)
# print("Service Name:", my_goog.service_name)
# 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: ")
# tokens = await my_goog.get_tokens(redirect_url = redirect_url)
# print("TOKENS:", json.to_string(tokens, default = str))
# # 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 10:40:40.833699+00:00"),
# "scopes": [
# "https://www.googleapis.com/auth/gmail.labels",
# "https://www.googleapis.com/auth/gmail.modify"
# ]
# }
# 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 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 = tokens, force_refresh = False)
print("REFRESHED TOKENS:", json.to_string(new_tok, default = str))