from typing import Dict, List from src.clients import service from src.logger import logger from src.utils.api import add_auth0_prefix, remove_auth0_prefix def deactivate_accounts() -> Dict[str, List[int]]: """Deactivate inactive or missing accounts. Returns: Table name to deactivated ID list. """ service.atlas_data.set_token(service.atlas_auth.get_access_token()) user_id_list = service.user_data.get_accounts() if not user_id_list: logger.info("Accounts not found") return {} logger.info(f"{len(user_id_list)} accounts found") result = service.atlas_data.get_users(user_id_list=add_auth0_prefix(user_id_list)) logger.info(f"{len(result)} users found") inactive_user_id_list = remove_auth0_prefix([i["id"] for i in result if not i["is_active"]], user_id_list) logger.info(f"Inactive {inactive_user_id_list}") missing_user_id_list = list(set(user_id_list) - set(remove_auth0_prefix([i["id"] for i in result], user_id_list))) logger.info(f"Missing {missing_user_id_list}") deactivate_user_id_list = inactive_user_id_list + missing_user_id_list if not deactivate_user_id_list: logger.info("Nothing to deactivate") return {} inactive_dup_user_id_list = service.user_data.get_accounts(is_active=False, user_id_list=deactivate_user_id_list) if inactive_dup_user_id_list: logger.warning(f"Inactive duplicates {inactive_dup_user_id_list}") deactivate_user_id_list = [i for i in deactivate_user_id_list if i not in inactive_dup_user_id_list] if not deactivate_user_id_list: logger.info("Nothing to deactivate") return {} result = service.user_data.deactivate_accounts(user_id_list=deactivate_user_id_list) logger.info(f"Deactivated {result}") return result def delete_inactive_accounts() -> Dict[str, List[int]]: """Delete inactive accounts and dependencies. """ result = service.user_data.delete_inactive_accounts() logger.info(f"Deleted {result}") return result