列出 OKTA 中每个应用程序分配给每个应用程序的所有用户

Question

我一直在尝试在我的项目中实现这一点，我编写的代码花了将近 90 分钟。我们在 OKTA 中总共有 170 个应用程序，OKTA 中的总活跃用户接近 1100 个。我在分页和速率限制方面遇到了很多问题，我最终想出了一个 Python 脚本，但这需要很长时间。如果有人可以帮助我改进我的脚本或建议我一种实现脚本的新方法，那么这将是一个很大的帮助。

    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.")

Answer 1

使代码更快的一种快速简单的方法是使用

requests.Session()

。 A

session

将保持连接打开，否则

requests

将关闭它并每次重新打开它。它也会使您的代码更短、更具可读性。您可以在开始时设置一次

headers

，然后在任何地方使用

session

。

#!/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'])