Explain how to use global token cache and app

rayluo · rayluo · commit 8bab1dd6fea2 · 2023-10-24T22:19:41.000-07:00
diff --git a/sample/confidential_client_certificate_sample.py b/sample/confidential_client_certificate_sample.py
@@ -31,6 +31,7 @@
 import sys  # For simplicity, we'll read config file from 1st CLI param sys.argv[1]
 import json
 import logging
+import time
 
 import requests
 import msal
@@ -42,27 +43,37 @@
 
 config = json.load(open(sys.argv[1]))
 
-# Create a preferably long-lived app instance which maintains a token cache.
-app = msal.ConfidentialClientApplication(
+# If for whatever reason you plan to recreate same ClientApplication periodically,
+# you shall create one global token cache and reuse it by each ClientApplication
+global_token_cache = msal.TokenCache()  # The TokenCache() is in-memory.
+    # See more options in https://msal-python.readthedocs.io/en/latest/#tokencache
+
+# Create a preferably long-lived app instance, to avoid the overhead of app creation
+global_app = msal.ConfidentialClientApplication(
     config["client_id"], authority=config["authority"],
     client_credential={"thumbprint": config["thumbprint"], "private_key": open(config['private_key_file']).read()},
-    # token_cache=...  # Default cache is in memory only.
-                       # You can learn how to use SerializableTokenCache from
-                       # https://msal-python.readthedocs.io/en/latest/#msal.SerializableTokenCache
+    token_cache=global_token_cache,  # Let this app (re)use an existing token cache.
+        # If absent, ClientApplication will create its own empty token cache
     )
 
-# Since MSAL 1.23, acquire_token_for_client(...) will automatically look up
-# a token from cache, and fall back to acquire a fresh token when needed.
-result = app.acquire_token_for_client(scopes=config["scope"])
-
-if "access_token" in result:
-    # Calling graph using the access token
-    graph_data = requests.get(  # Use token to call downstream service
-        config["endpoint"],
-        headers={'Authorization': 'Bearer ' + result['access_token']},).json()
-    print("Graph API call result: %s" % json.dumps(graph_data, indent=2))
-else:
-    print(result.get("error"))
-    print(result.get("error_description"))
-    print(result.get("correlation_id"))  # You may need this when reporting a bug
+
+def acquire_and_use_token():
+    # Since MSAL 1.23, acquire_token_for_client(...) will automatically look up
+    # a token from cache, and fall back to acquire a fresh token when needed.
+    result = global_app.acquire_token_for_client(scopes=config["scope"])
+
+    if "access_token" in result:
+        # Calling graph using the access token
+        graph_data = requests.get(  # Use token to call downstream service
+            config["endpoint"],
+            headers={'Authorization': 'Bearer ' + result['access_token']},).json()
+        print("Graph API call result: %s" % json.dumps(graph_data, indent=2))
+    else:
+        print("Token acquisition failed")  # Examine result["error_description"] etc. to diagnose error
+
+
+while True:  # Here we mimic a long-lived daemon
+    acquire_and_use_token()
+    print("Press Ctrl-C to stop.")
+    time.sleep(5)  # Let's say your app would run a workload every X minutes
 
diff --git a/sample/confidential_client_secret_sample.py b/sample/confidential_client_secret_sample.py
@@ -30,6 +30,7 @@
 import sys  # For simplicity, we'll read config file from 1st CLI param sys.argv[1]
 import json
 import logging
+import time
 
 import requests
 import msal
@@ -41,28 +42,37 @@
 
 config = json.load(open(sys.argv[1]))
 
-# Create a preferably long-lived app instance which maintains a token cache.
-app = msal.ConfidentialClientApplication(
+# If for whatever reason you plan to recreate same ClientApplication periodically,
+# you shall create one global token cache and reuse it by each ClientApplication
+global_token_cache = msal.TokenCache()  # The TokenCache() is in-memory.
+    # See more options in https://msal-python.readthedocs.io/en/latest/#tokencache
+
+# Create a preferably long-lived app instance, to avoid the overhead of app creation
+global_app = msal.ConfidentialClientApplication(
     config["client_id"], authority=config["authority"],
     client_credential=config["secret"],
-    # token_cache=...  # Default cache is in memory only.
-                       # You can learn how to use SerializableTokenCache from
-                       # https://msal-python.readthedocs.io/en/latest/#msal.SerializableTokenCache
+    token_cache=global_token_cache,  # Let this app (re)use an existing token cache.
+        # If absent, ClientApplication will create its own empty token cache
     )
 
-# Since MSAL 1.23, acquire_token_for_client(...) will automatically look up
-# a token from cache, and fall back to acquire a fresh token when needed.
-result = app.acquire_token_for_client(scopes=config["scope"])
-
-if "access_token" in result:
-    # Calling graph using the access token
-    graph_data = requests.get(  # Use token to call downstream service
-        config["endpoint"],
-        headers={'Authorization': 'Bearer ' + result['access_token']},).json()
-    print("Graph API call result: %s" % json.dumps(graph_data, indent=2))
-
-else:
-    print(result.get("error"))
-    print(result.get("error_description"))
-    print(result.get("correlation_id"))  # You may need this when reporting a bug
+
+def acquire_and_use_token():
+    # Since MSAL 1.23, acquire_token_for_client(...) will automatically look up
+    # a token from cache, and fall back to acquire a fresh token when needed.
+    result = global_app.acquire_token_for_client(scopes=config["scope"])
+
+    if "access_token" in result:
+        # Calling graph using the access token
+        graph_data = requests.get(  # Use token to call downstream service
+            config["endpoint"],
+            headers={'Authorization': 'Bearer ' + result['access_token']},).json()
+        print("Graph API call result: %s" % json.dumps(graph_data, indent=2))
+    else:
+        print("Token acquisition failed")  # Examine result["error_description"] etc. to diagnose error
+
+
+while True:  # Here we mimic a long-lived daemon
+    acquire_and_use_token()
+    print("Press Ctrl-C to stop.")
+    time.sleep(5)  # Let's say your app would run a workload every X minutes
 
diff --git a/sample/device_flow_sample.py b/sample/device_flow_sample.py
@@ -20,6 +20,7 @@
 import sys  # For simplicity, we'll read config file from 1st CLI param sys.argv[1]
 import json
 import logging
+import time
 
 import requests
 import msal
@@ -31,58 +32,70 @@
 
 config = json.load(open(sys.argv[1]))
 
-# Create a preferably long-lived app instance which maintains a token cache.
-app = msal.PublicClientApplication(
+# If for whatever reason you plan to recreate same ClientApplication periodically,
+# you shall create one global token cache and reuse it by each ClientApplication
+global_token_cache = msal.TokenCache()  # The TokenCache() is in-memory.
+    # See more options in https://msal-python.readthedocs.io/en/latest/#tokencache
+
+# Create a preferably long-lived app instance, to avoid the overhead of app creation
+global_app = msal.PublicClientApplication(
     config["client_id"], authority=config["authority"],
-    # token_cache=...  # Default cache is in memory only.
-                       # You can learn how to use SerializableTokenCache from
-                       # https://msal-python.readthedocs.io/en/latest/#msal.SerializableTokenCache
+    token_cache=global_token_cache,  # Let this app (re)use an existing token cache.
+        # If absent, ClientApplication will create its own empty token cache
     )
 
-# The pattern to acquire a token looks like this.
-result = None
-
-# Note: If your device-flow app does not have any interactive ability, you can
-#   completely skip the following cache part. But here we demonstrate it anyway.
-# We now check the cache to see if we have some end users signed in before.
-accounts = app.get_accounts()
-if accounts:
-    logging.info("Account(s) exists in cache, probably with token too. Let's try.")
-    print("Pick the account you want to use to proceed:")
-    for a in accounts:
-        print(a["username"])
-    # Assuming the end user chose this one
-    chosen = accounts[0]
-    # Now let's try to find a token in cache for this account
-    result = app.acquire_token_silent(config["scope"], account=chosen)
-
-if not result:
-    logging.info("No suitable token exists in cache. Let's get a new one from AAD.")
-
-    flow = app.initiate_device_flow(scopes=config["scope"])
-    if "user_code" not in flow:
-        raise ValueError(
-            "Fail to create device flow. Err: %s" % json.dumps(flow, indent=4))
-
-    print(flow["message"])
-    sys.stdout.flush()  # Some terminal needs this to ensure the message is shown
-
-    # Ideally you should wait here, in order to save some unnecessary polling
-    # input("Press Enter after signing in from another device to proceed, CTRL+C to abort.")
-
-    result = app.acquire_token_by_device_flow(flow)  # By default it will block
-        # You can follow this instruction to shorten the block time
-        #    https://msal-python.readthedocs.io/en/latest/#msal.PublicClientApplication.acquire_token_by_device_flow
-        # or you may even turn off the blocking behavior,
-        # and then keep calling acquire_token_by_device_flow(flow) in your own customized loop.
-
-if "access_token" in result:
-    # Calling graph using the access token
-    graph_data = requests.get(  # Use token to call downstream service
-        config["endpoint"],
-        headers={'Authorization': 'Bearer ' + result['access_token']},).json()
-    print("Graph API call result: %s" % json.dumps(graph_data, indent=2))
-else:
-    print(result.get("error"))
-    print(result.get("error_description"))
-    print(result.get("correlation_id"))  # You may need this when reporting a bug
+
+def acquire_and_use_token():
+    # The pattern to acquire a token looks like this.
+    result = None
+
+    # Note: If your device-flow app does not have any interactive ability, you can
+    #   completely skip the following cache part. But here we demonstrate it anyway.
+    # We now check the cache to see if we have some end users signed in before.
+    accounts = global_app.get_accounts()
+    if accounts:
+        logging.info("Account(s) exists in cache, probably with token too. Let's try.")
+        print("Pick the account you want to use to proceed:")
+        for a in accounts:
+            print(a["username"])
+        # Assuming the end user chose this one
+        chosen = accounts[0]
+        # Now let's try to find a token in cache for this account
+        result = global_app.acquire_token_silent(config["scope"], account=chosen)
+
+    if not result:
+        logging.info("No suitable token exists in cache. Let's get a new one from AAD.")
+
+        flow = global_app.initiate_device_flow(scopes=config["scope"])
+        if "user_code" not in flow:
+            raise ValueError(
+                "Fail to create device flow. Err: %s" % json.dumps(flow, indent=4))
+
+        print(flow["message"])
+        sys.stdout.flush()  # Some terminal needs this to ensure the message is shown
+
+        # Ideally you should wait here, in order to save some unnecessary polling
+        # input("Press Enter after signing in from another device to proceed, CTRL+C to abort.")
+
+        result = global_app.acquire_token_by_device_flow(flow)  # By default it will block
+            # You can follow this instruction to shorten the block time
+            #    https://msal-python.readthedocs.io/en/latest/#msal.PublicClientApplication.acquire_token_by_device_flow
+            # or you may even turn off the blocking behavior,
+            # and then keep calling acquire_token_by_device_flow(flow) in your own customized loop.
+
+    if "access_token" in result:
+        # Calling graph using the access token
+        graph_data = requests.get(  # Use token to call downstream service
+            config["endpoint"],
+            headers={'Authorization': 'Bearer ' + result['access_token']},).json()
+        print("Graph API call result: %s" % json.dumps(graph_data, indent=2))
+    else:
+        print("Token acquisition failed")  # Examine result["error_description"] etc. to diagnose error
+
+
+while True:  # Here we mimic a long-lived daemon
+    acquire_and_use_token()
+    print("Press Ctrl-C to stop.")
+    time.sleep(5)  # Let's say your app would run a workload every X minutes.
+        # The first acquire_and_use_token() call will prompt. Others hit the cache.
+
diff --git a/sample/interactive_sample.py b/sample/interactive_sample.py
@@ -20,66 +20,77 @@
 
     python sample.py parameters.json
 """
-
 import sys  # For simplicity, we'll read config file from 1st CLI param sys.argv[1]
-import json, logging, msal, requests
+import json, logging, time, msal, requests
 
 # Optional logging
 # logging.basicConfig(level=logging.DEBUG)  # Enable DEBUG log for entire script
 # logging.getLogger("msal").setLevel(logging.INFO)  # Optionally disable MSAL DEBUG logs
 
 config = json.load(open(sys.argv[1]))
 
-# Create a preferably long-lived app instance which maintains a token cache.
-app = msal.PublicClientApplication(
+# If for whatever reason you plan to recreate same ClientApplication periodically,
+# you shall create one global token cache and reuse it by each ClientApplication
+global_token_cache = msal.TokenCache()  # The TokenCache() is in-memory.
+    # See more options in https://msal-python.readthedocs.io/en/latest/#tokencache
+
+# Create a preferably long-lived app instance, to avoid the overhead of app creation
+global_app = msal.PublicClientApplication(
     config["client_id"], authority=config["authority"],
     #allow_broker=True,  # If opted in, you will be guided to meet the prerequisites, when applicable
                          # See also: https://docs.microsoft.com/en-us/azure/active-directory/develop/scenario-desktop-acquire-token-wam#wam-value-proposition
-    # token_cache=...  # Default cache is in memory only.
-                       # You can learn how to use SerializableTokenCache from
-                       # https://msal-python.readthedocs.io/en/latest/#msal.SerializableTokenCache
+    token_cache=global_token_cache,  # Let this app (re)use an existing token cache.
+        # If absent, ClientApplication will create its own empty token cache
     )
 
-# The pattern to acquire a token looks like this.
-result = None
-
-# Firstly, check the cache to see if this end user has signed in before
-accounts = app.get_accounts(username=config.get("username"))
-if accounts:
-    logging.info("Account(s) exists in cache, probably with token too. Let's try.")
-    print("Account(s) already signed in:")
-    for a in accounts:
-        print(a["username"])
-    chosen = accounts[0]  # Assuming the end user chose this one to proceed
-    print("Proceed with account: %s" % chosen["username"])
-    # Now let's try to find a token in cache for this account
-    result = app.acquire_token_silent(config["scope"], account=chosen)
-
-if not result:
-    logging.info("No suitable token exists in cache. Let's get a new one from AAD.")
-    print("A local browser window will be open for you to sign in. CTRL+C to cancel.")
-    result = app.acquire_token_interactive(  # Only works if your app is registered with redirect_uri as http://localhost
-        config["scope"],
-        #parent_window_handle=...,  # If broker is enabled, you will be guided to provide a window handle
-        login_hint=config.get("username"),  # Optional.
-            # If you know the username ahead of time, this parameter can pre-fill
-            # the username (or email address) field of the sign-in page for the user,
-            # Often, apps use this parameter during reauthentication,
-            # after already extracting the username from an earlier sign-in
-            # by using the preferred_username claim from returned id_token_claims.
-
-        #prompt=msal.Prompt.SELECT_ACCOUNT,  # Or simply "select_account". Optional. It forces to show account selector page
-        #prompt=msal.Prompt.CREATE,  # Or simply "create". Optional. It brings user to a self-service sign-up flow.
-            # Prerequisite: https://docs.microsoft.com/en-us/azure/active-directory/external-identities/self-service-sign-up-user-flow
-        )
-
-if "access_token" in result:
-    # Calling graph using the access token
-    graph_response = requests.get(  # Use token to call downstream service
-        config["endpoint"],
-        headers={'Authorization': 'Bearer ' + result['access_token']},)
-    print("Graph API call result: %s ..." % graph_response.text[:100])
-else:
-    print(result.get("error"))
-    print(result.get("error_description"))
-    print(result.get("correlation_id"))  # You may need this when reporting a bug
+
+def acquire_and_use_token():
+    # The pattern to acquire a token looks like this.
+    result = None
+
+    # Firstly, check the cache to see if this end user has signed in before
+    accounts = global_app.get_accounts(username=config.get("username"))
+    if accounts:
+        logging.info("Account(s) exists in cache, probably with token too. Let's try.")
+        print("Account(s) already signed in:")
+        for a in accounts:
+            print(a["username"])
+        chosen = accounts[0]  # Assuming the end user chose this one to proceed
+        print("Proceed with account: %s" % chosen["username"])
+        # Now let's try to find a token in cache for this account
+        result = global_app.acquire_token_silent(config["scope"], account=chosen)
+
+    if not result:
+        logging.info("No suitable token exists in cache. Let's get a new one from AAD.")
+        print("A local browser window will be open for you to sign in. CTRL+C to cancel.")
+        result = global_app.acquire_token_interactive(  # Only works if your app is registered with redirect_uri as http://localhost
+            config["scope"],
+            #parent_window_handle=...,  # If broker is enabled, you will be guided to provide a window handle
+            login_hint=config.get("username"),  # Optional.
+                # If you know the username ahead of time, this parameter can pre-fill
+                # the username (or email address) field of the sign-in page for the user,
+                # Often, apps use this parameter during reauthentication,
+                # after already extracting the username from an earlier sign-in
+                # by using the preferred_username claim from returned id_token_claims.
+
+            #prompt=msal.Prompt.SELECT_ACCOUNT,  # Or simply "select_account". Optional. It forces to show account selector page
+            #prompt=msal.Prompt.CREATE,  # Or simply "create". Optional. It brings user to a self-service sign-up flow.
+                # Prerequisite: https://docs.microsoft.com/en-us/azure/active-directory/external-identities/self-service-sign-up-user-flow
+            )
+
+    if "access_token" in result:
+        # Calling graph using the access token
+        graph_response = requests.get(  # Use token to call downstream service
+            config["endpoint"],
+            headers={'Authorization': 'Bearer ' + result['access_token']},)
+        print("Graph API call result: %s ..." % graph_response.text[:100])
+    else:
+        print("Token acquisition failed")  # Examine result["error_description"] etc. to diagnose error
+
+
+while True:  # Here we mimic a long-lived daemon
+    acquire_and_use_token()
+    print("Press Ctrl-C to stop.")
+    time.sleep(5)  # Let's say your app would run a workload every X minutes
+        # The first acquire_and_use_token() call will prompt. Others hit the cache.
+
diff --git a/sample/username_password_sample.py b/sample/username_password_sample.py
diff --git a/sample/vault_jwt_sample.py b/sample/vault_jwt_sample.py
