使用Asyncio run_until_complete调用Okta API时出现user对象为None错误

问题详情 #

我调用Okta API读取CSV文件批量更新用户主邮箱，但执行时获取的user对象为None类型，最终导致run_until_complete执行失败，报错指向访问user.profile.email的代码行。

原代码 #

import csv
import asyncio
from okta.client import Client as OktaClient
from okta import models

config = {
    'orgUrl': '',
    'token': ''
}

okta_client = OktaClient(config)

async def main(): 
    with open(r'C:\Users\user\Downloads\Export Users 2023-07-26 20-20-57.csv') as csv_file:
        csv_reader = csv.DictReader(csv_file)

        for line in csv_reader:
            user, resp, err = await okta_client.get_user(line['Primary email'])
            print(user.profile.email)
            new_profile = user.profile
            new_profile.email = line['New primary email']
            updated_user_obj = models.User({'profile': new_profile})
            updated_user, _, err = await okta_client.update_user(user.id, updated_user_obj)
            
        print('Users upddated')
loop = asyncio.get_event_loop()
loop.run_until_complete(main())

报错信息 #

Traceback (most recent call last):
  File "C:\Users\hami_\Documents\Stagwell\Okta\Okta Automations\bulk_user_update_primaryemail.py", line 28, in <module>
    loop.run_until_complete(main())
  File "C:\Users\hami_\AppData\Local\Programs\Python\Python311\Lib\asyncio\base_events.py", line 653, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "C:\Users\hami_\Documents\Stagwell\Okta\Okta Automations\bulk_user_update_primaryemail.py", line 20, in main
    print(user.profile.email)

解决方法 #

1. 添加错误处理逻辑 #

原代码未处理okta_client.get_user的返回错误，当API调用失败或用户不存在时，user会返回None。需先判断错误并跳过异常用户：

async def main(): 
    with open(r'C:\Users\user\Downloads\Export Users 2023-07-26 20-20-57.csv') as csv_file:
        csv_reader = csv.DictReader(csv_file)
        for line_idx, line in enumerate(csv_reader, 1):
            original_email = line['Primary email']
            user, resp, err = await okta_client.get_user(original_email)
            
            # 处理获取用户失败的情况
            if err:
                print(f"第{line_idx}行用户获取失败，邮箱：{original_email}，错误：{err}")
                continue
            if not user:
                print(f"第{line_idx}行未找到用户，邮箱：{original_email}")
                continue
            
            # 执行更新逻辑
            try:
                new_profile = user.profile
                new_profile.email = line['New primary email']
                updated_user_obj = models.User({'profile': new_profile})
                updated_user, _, update_err = await okta_client.update_user(user.id, updated_user_obj)
                if update_err:
                    print(f"第{line_idx}行用户更新失败，原邮箱：{original_email}，错误：{update_err}")
                else:
                    print(f"第{line_idx}行用户更新成功，原邮箱：{original_email}，新邮箱：{new_profile.email}")
            except Exception as e:
                print(f"第{line_idx}行处理异常，原邮箱：{original_email}，异常：{str(e)}")
                
        print('批量更新任务结束')

2. 验证CSV数据准确性 #

• 确认CSV中Primary email列的邮箱地址与Okta系统存储的完全一致（注意大小写、空格、拼写错误）
• 检查New primary email列格式符合Okta邮箱规范

3. 确认API权限 #

确保配置的Okta API Token拥有以下权限：

• okta.users.read（读取用户权限）
• okta.users.update（更新用户权限）

4. 优化异步执行逻辑 #

Python 3.7+推荐使用asyncio.run()替代get_event_loop()，代码更简洁兼容性更好：

asyncio.run(main())

内容的提问来源于stack exchange，提问作者Axel