lectom/descope-create_bulk_users
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

테스트 케이스 추가, update_or_create 추가

Browse Source
This commit is contained in:
Lectom C Han
2025-10-13 15:12:52 +09:00
parent 48c95b2c24
commit d4df3ee861
6 changed files with 309 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
main.py
View File

@@ -66,9 +66,17 @@ def csv_from_google_sheet_url(url: str) -> list[dict]:
def _prepare_user_data(user, expiry_timestamp=None):
"""
사용자 데이터로부터 Descope API에 필요한 테넌트, 커스텀 속성 등을 준비합니다.
미리 정의된 필드 외의 모든 키-값 쌍은 custom_attributes에 포함됩니다.
"""
login_id = user.get('login_id')
# Descope 사용자 객체의 최상위 레벨 필드 또는 스크립트에서 특별히 처리되는 필드 목록
known_fields = {
'login_id', 'email', 'display_name', 'tenants', 'role_name',
'status', 'new_password', 'custom_attributes_key', 'custom_attributes_value',
'company', 'egBimLExpiryDate'
}
user_tenants = []
tenants_str = user.get('tenants')
if tenants_str:
@@ -85,6 +93,11 @@ def _prepare_user_data(user, expiry_timestamp=None):
print(f"경고: {login_id}의 'tenants' 필드를 파싱할 수 없습니다: {tenants_str}")
custom_attributes = {}
# user 딕셔너리의 모든 항목을 순회하며 custom_attributes 구성
for key, value in user.items():
if key not in known_fields and value:
custom_attributes[key] = value
if user.get('custom_attributes_key') and user.get('custom_attributes_value'):
custom_attributes[user['custom_attributes_key']] = user['custom_attributes_value']
@@ -94,9 +107,8 @@ def _prepare_user_data(user, expiry_timestamp=None):
custom_attributes['completeForm'] = True
# egBimLExpiryDate 처리
final_expiry_date = expiry_timestamp # CLI 인자가 우선순위가 가장 높음
final_expiry_date = expiry_timestamp
if final_expiry_date is None and user.get('egBimLExpiryDate'):
# CLI 인자가 없고 CSV에 값이 있을 경우, 해당 값을 변환하여 사용
final_expiry_date = convert_to_timestamp(user.get('egBimLExpiryDate'))
if final_expiry_date is not None:
@@ -112,16 +124,13 @@ def convert_to_timestamp(date_value):
if not date_value:
return None
# Already an integer timestamp
if isinstance(date_value, int):
return date_value
if isinstance(date_value, str):
# Check if it's a string representation of an integer
try:
return int(date_value)
except ValueError:
# If not, try to parse it as a YYYY-MM-DD date string
try:
dt = datetime.strptime(date_value, '%Y-%m-%d')
dt_utc = dt.replace(tzinfo=timezone.utc)
@@ -134,11 +143,13 @@ def convert_to_timestamp(date_value):
return None
def create_users(users_data, expiry_timestamp=None):
def create_users(users_data, expiry_timestamp=None, dry_run=False):
"""
CSV에서 읽어온 사용자 데이터를 기반으로 Descope를 통해 계정을 일괄 생성합니다.
"""
print("--- 사용자 계정 일괄 생성을 시작합니다 ---")
if dry_run:
print("*** DRY RUN 모드로 실행됩니다. 실제 데이터는 변경되지 않습니다. ***")
try:
descope_client = get_descope_client()
for user in users_data:
@@ -149,9 +160,14 @@ def create_users(users_data, expiry_timestamp=None):
continue
user_tenants, custom_attributes = _prepare_user_data(user, expiry_timestamp)
status = user.get('status', 'activated')
if dry_run:
print(f"[DRY RUN] 사용자 생성 예정: {login_id}, 속성: {custom_attributes}")
if status != 'activated':
print(f"[DRY RUN] 사용자 상태 업데이트 예정: {login_id} -> {status}")
continue
print(f"사용자 생성 시도: {login_id}")
descope_client.mgmt.user.create(
login_id=login_id,
@@ -182,11 +198,13 @@ def create_users(users_data, expiry_timestamp=None):
print("--- 사용자 계정 일괄 생성이 완료되었습니다 ---")
def update_or_create_users(users_data, expiry_timestamp=None):
def update_or_create_users(users_data, expiry_timestamp=None, dry_run=False):
"""
사용자 이메일(login_id)을 확인하여 존재하면 정보를 업데이트하고, 존재하지 않으면 새로 생성합니다.
"""
print("--- 사용자 정보 업데이트 또는 생성을 시작합니다 ---")
if dry_run:
print("*** DRY RUN 모드로 실행됩니다. 실제 데이터는 변경되지 않습니다. ***")
try:
descope_client = get_descope_client()
for user in users_data:
@@ -199,20 +217,20 @@ def update_or_create_users(users_data, expiry_timestamp=None):
user_tenants, custom_attributes = _prepare_user_data(user, expiry_timestamp)
display_name = user.get('display_name')
# 사용자 존재 여부 확인
if dry_run:
print(f"[DRY RUN] 사용자 업데이트 또는 생성 예정: {login_id}, 이름: {display_name}, 속성: {custom_attributes}")
continue
try:
descope_client.mgmt.user.load(login_id=login_id)
user_exists = True
except AuthException as e:
# Descope SDK는 사용자를 찾지 못하면 AuthException을 발생시킵니다.
# 오류 메시지에 "not found"가 포함되어 있는지 확인하여 사용자가 없는 경우를 특정합니다.
if "user not found" in str(e).lower() or "user not found" in str(e.args).lower():
user_exists = False
else:
raise e # 다른 종류의 AuthException은 다시 발생시킴
raise e
if user_exists:
# 사용자 업데이트
print(f"사용자 업데이트 시도: {login_id}")
descope_client.mgmt.user.update(
login_id=login_id,
@@ -223,7 +241,6 @@ def update_or_create_users(users_data, expiry_timestamp=None):
)
print(f"성공: {login_id} 사용자 정보가 업데이트되었습니다.")
else:
# 사용자 생성
print(f"사용자 생성 시도: {login_id}")
descope_client.mgmt.user.create(
login_id=login_id,
@@ -234,10 +251,7 @@ def update_or_create_users(users_data, expiry_timestamp=None):
)
print(f"성공: {login_id} 사용자가 생성되었습니다.")
# 상태 업데이트 (생성/업데이트 공통)
status = user.get('status', 'activated')
# 'activated' 상태는 기본값이므로 별도로 업데이트할 필요가 없습니다.
# 다른 상태일 경우에만 상태 업데이트를 시도합니다.
if status != 'activated':
try:
print(f"사용자 상태 업데이트 시도: {login_id} -> {status}")
@@ -258,11 +272,13 @@ def update_or_create_users(users_data, expiry_timestamp=None):
print("--- 사용자 정보 업데이트 또는 생성이 완료되었습니다 ---")
def change_passwords(users_data):
def change_passwords(users_data, dry_run=False):
"""
CSV에서 읽어온 사용자 데이터를 기반으로 Descope를 통해 비밀번호를 일괄 변경합니다.
"""
print("--- 사용자 비밀번호 일괄 변경을 시작합니다 ---")
if dry_run:
print("*** DRY RUN 모드로 실행됩니다. 실제 데이터는 변경되지 않습니다. ***")
try:
descope_client = get_descope_client()
for user in users_data:
@@ -274,6 +290,10 @@ def change_passwords(users_data):
print(f"경고: 'login_id' 또는 'new_password'가 없는 행을 건너뜁니다: {user}")
continue
if dry_run:
print(f"[DRY RUN] 비밀번호 변경 예정: {login_id}")
continue
print(f"비밀번호 변경 시도: {login_id}")
descope_client.mgmt.user.set_active_password(
login_id=login_id,
@@ -297,7 +317,7 @@ def change_passwords(users_data):
print("--- 사용자 비밀번호 일괄 변경이 완료되었습니다 ---")
def test_create_single_user():
def test_create_single_user(dry_run=False):
"""
단일 사용자 생성을 테스트하는 함수입니다.
"""
@@ -306,14 +326,13 @@ def test_create_single_user():
'login_id': 'testuser.gemini@example.com',
'email': 'testuser.gemini@example.com',
'display_name': 'Gemini Test User',
'tenants': '["T31ZmUcwOZbwk0y3YmMxrPCpzpQR"]', # 실제 테스트용 Tenant ID로 변경 필요
'tenants': '["T31ZmUcwOZbwk0y3YmMxrPCpzpQR"]',
'company': 'Gemini Test Inc.',
'egBimLExpiryDate': 1798729200,
'status': 'activated',
'verifiedEmail': True,
'is_test_user': True
}]
create_users(test_user_data)
create_users(test_user_data, dry_run=dry_run)
print("--- 단일 사용자 생성 테스트가 완료되었습니다 ---")
@@ -330,11 +349,12 @@ def main():
help="수행할 작업 (create: 생성, change_password: 비밀번호 변경, update_or_create: 업데이트 또는 생성, test: 테스트)"
)
parser.add_argument("--expiry-date", help="사용자 라이선스 만료일을 'YYYY-MM-DD' 형식으로 지정합니다. 이 값을 지정하면 CSV 파일의 모든 'egBimLExpiryDate' 값을 덮어씁니다.")
parser.add_argument("--dry-run", action='store_true', help="실제 API를 호출하지 않고 실행할 작업만 출력합니다.")
args = parser.parse_args()
if args.action == 'test':
test_create_single_user()
test_create_single_user(dry_run=args.dry_run)
return
if not args.source:
@@ -353,16 +373,15 @@ def main():
if args.expiry_date:
expiry_timestamp = convert_to_timestamp(args.expiry_date)
if expiry_timestamp is None:
# convert_to_timestamp 함수 내부에서 이미 경고 메시지를 출력하므로 여기서는 종료만 합니다.
return
print(f"전체 만료 날짜가 {args.expiry_date}로 설정되었습니다 (UTC 타임스탬프: {expiry_timestamp}).")
if args.action == 'create':
create_users(users_data, expiry_timestamp)
create_users(users_data, expiry_timestamp, dry_run=args.dry_run)
elif args.action == 'change_password':
change_passwords(users_data)
change_passwords(users_data, dry_run=args.dry_run)
elif args.action == 'update_or_create':
update_or_create_users(users_data, expiry_timestamp)
update_or_create_users(users_data, expiry_timestamp, dry_run=args.dry_run)
except FileNotFoundError:
print(f"오류: 파일을 찾을 수 없습니다 - {args.source}")