list all the users assigned to each application for every app in OKTA

I've been trying to implement this for my project and the code I've written is taking almost 90 minutes. We have total 170 apps in OKTA and total active users in OKTA are close to 1100. I've had so many issues with pagination and rate limiting and I finally came up with a script in Python but it takes forever. If anyone could help me with improvements to my script or suggest me a new approach for implementing the script then that would be a great help.

    import requests
import csv
import os
import time
def get_all_users(domain, okta_api_token):
headers = {
'Authorization': f'SSWS {okta_api_token}',
'Accept': 'application/json',
'Content-Type': 'application/json'
}
okta_url = f'https://{domain}.okta.com'
url = f'{okta_url}/api/v1/users'
all_users = []
while url:
response = requests.get(url, headers=headers)
response.raise_for_status()
users = response.json()
all_users += users
links = response.links
url = links.get('next', {}).get('url')
return all_users
def get_user_apps(domain, okta_api_token, user_id, app_links):
if user_id in app_links:
return app_links[user_id]
headers = {
'Authorization': f'SSWS {okta_api_token}',
'Accept': 'application/json',
'Content-Type': 'application/json'
}
okta_url = f'https://{domain}.okta.com'
url = f'{okta_url}/api/v1/users/{user_id}/appLinks'
apps = []
while url:
response = requests.get(url, headers=headers)
response.raise_for_status()
app_links_data = response.json()
apps += app_links_data
links = response.links
url = links.get('next', {}).get('url')
app_names = [app['label'] for app in apps]
app_links[user_id] = ', '.join(app_names)
return app_links[user_id]
def extract_user_data(domain, okta_api_token, users, app_links):
user_data = []
for user in users:
user_id = user.get('id', 'N/A')
user_name = user.get('profile', {}).get('login', 'N/A')
email = user.get('profile', {}).get('email', 'N/A')
first_name = user.get('profile', {}).get('firstName', 'N/A')
last_name = user.get('profile', {}).get('lastName', 'N/A')
full_name = f"{first_name} {last_name}"
apps = get_user_apps(domain, okta_api_token, user_id, app_links)
user_data.append({'id': user_id, 'username': user_name, 'email': email, 'full_name': full_name, 'apps': apps})
return user_data
def extract_app_user_data(domain, okta_api_token, users):
app_user_data = {}
app_links = {}
for user in users:
user_id = user.get('id', 'N/A')
apps = get_user_apps(domain, okta_api_token, user_id, app_links)
for app in apps.split(', '):
if app not in app_user_data:
app_user_data[app] = []
app_user_data[app].append(user.get('profile', {}).get('login', 'N/A'))
return app_user_data
# Read domain and okta_api_token from environment variables
domain = os.environ['OKTA_DOMAIN']
okta_api_token = os.environ['OKTA_API_TOKEN']
# Start the timer
start_time = time.time()
# Get all users and extract the desired data
users = get_all_users(domain, okta_api_token)
app_user_data = extract_app_user_data(domain, okta_api_token, users)
app_links = {}
extracted_users = extract_user_data(domain, okta_api_token, users, app_links)
# Save extracted app user data to a numbered CSV file
with open('app_user_data.csv', 'w', newline='') as csvfile:
fieldnames = ['app', 'users']
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
for app, users in app_user_data.items():
writer.writerow({'app': app, 'users': ', '.join(users)})
# Save extracted user data to a numbered CSV file
with open('user_data.csv', 'w', newline='') as csvfile:
fieldnames = ['id', 'username', 'email', 'full_name', 'apps']
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
for user in extracted_users:
writer.writerow(user)
# Calculate and print the execution time
end_time = time.time()
execution_time = end_time - start_time
print(f"Execution time: {execution_time} seconds.")

One quick and easy way to make the code faster is to use requests.Session(). A session will keep the connection open, otherwise requests closes it and reopens it every time. It'll make your code shorter and more readable, too. You can set the headers once at the beginning and use the session everywhere.

#!/usr/bin/env python

import requests

# Set these:
url = 'https://COMPANY.okta.com/api/v1/users'
token = '...'

# If you're making multiple API calls, using a session is much faster.
session = requests.Session()
session.headers['authorization'] = 'SSWS ' + token

def get_objects(url):
    while url:
        r = session.get(url)
        for o in r.json():
            yield o
        url = r.links.get('next', {}).get('url')
        
for user in get_objects(url):
    print(user['profile']['login'])