Add token caching

This commit is contained in:
2024-06-24 21:40:36 +02:00
parent cc49757545
commit be1aa99ada

View File

@@ -13,7 +13,8 @@ except ImportError:
from backports.zoneinfo import ZoneInfo
WEEKDAYS= {0:"Mo", 1:"Di", 2:"Mi", 3:"Do", 4: "Fr", 5:"Sa", 6: "So"}
result = None
token = None
cache = msal.SerializableTokenCache()
# Optional logging
@@ -21,36 +22,38 @@ result = None
# Create a preferably long-lived app instance which maintains a token cache.
app = msal.ConfidentialClientApplication(
MsalConfig.CLIENT_ID, authority=MsalConfig.AUTHORITY,
MsalConfig.CLIENT_ID,
authority=MsalConfig.AUTHORITY,
client_credential=MsalConfig.SECRET,
# token_cache=... # Default cache is in memory only.
token_cache=cache
# Default cache is in memory only.
# You can learn how to use SerializableTokenCache from
# https:#msal-python.readthedocs.io/en/latest/#msal.SerializableTokenCache
)
def get_access_token():
global app, result
global app, token
# The pattern to acquire a token looks like this.
# Firstly, looks up a token from cache
# Since we are looking for token for the current app, NOT for an end user,
# notice we give account parameter as None.
result = app.acquire_token_silent(MsalConfig.SCOPE, account=None)
token = app.acquire_token_silent(MsalConfig.SCOPE, account=None)
if result is None:
if token is None:
logging.info("No suitable token exists in cache. Let's get a new one from AAD.")
result= app.acquire_token_for_client(scopes=MsalConfig.SCOPE)
token= app.acquire_token_for_client(scopes=MsalConfig.SCOPE)
else:
logging.info("Token was found in cache.")
if not "access_token" in result: # a final check
logging.error(result.get("error"))
logging.error(result.get("error_description"))
logging.error(result.get("correlation_id")) # You may need this when reporting a bug
if not "access_token" in token: # a final check
logging.error(token.get("error"))
logging.error(token.get("error_description"))
logging.error(token.get("correlation_id")) # You may need this when reporting a bug
raise AssertionError("Was not able to get an access token. Check msal auth.")
return result
return token
def execute_get_request(token: dict, endpoint:str):
return requests.get( # Use token to call downstream service