import logging from typing import Dict import boto3 logger = logging.getLogger(__name__) USER_POOL_MAP: Dict[str, str] = {} def get_user_pool_id(user_pool_name): global USER_POOL_MAP if not USER_POOL_MAP: cognito = boto3.client("cognito-idp", region_name="eu-west-1") for pool in cognito.list_user_pools(MaxResults=60)["UserPools"]: USER_POOL_MAP[pool["Name"]] = pool["Id"] return USER_POOL_MAP[user_pool_name] def create_user( new_email, tmp_passwd, invi_default_schema, user_pool_name, first_name=None, last_name=None, company_name=None, ): cognito = boto3.client("cognito-idp", region_name="eu-west-1") attributes = [ {"Name": "email", "Value": new_email}, {"Name": "email_verified", "Value": "true"}, ] if company_name: attributes.append({"Name": "custom:company", "Value": company_name}) if first_name: attributes.append({"Name": "given_name", "Value": first_name}) if last_name: attributes.append({"Name": "family_name", "Value": last_name}) result = cognito.admin_create_user( UserPoolId=get_user_pool_id(user_pool_name), Username=new_email, UserAttributes=attributes, TemporaryPassword=tmp_passwd, MessageAction="SUPPRESS", DesiredDeliveryMediums=[], ) sub = None for attr in result["User"]["Attributes"]: if attr["Name"] == "sub": sub = attr["Value"] break if sub: return sub else: logger.error("NEW USER CREATION FAILED") raise RuntimeError("Could not create new user.") def get_cognito_user_attributes(user_id, user_pool_name): cognito = boto3.client("cognito-idp", region_name="eu-west-1") user = cognito.admin_get_user( UserPoolId=get_user_pool_id(user_pool_name), Username=user_id ) if "UserAttributes" in user: return {attr["Name"]: attr["Value"] for attr in user["UserAttributes"]} else: return {} def delete_user(user_id, user_pool_name): cognito = boto3.client("cognito-idp", region_name="eu-west-1") cognito.admin_delete_user( UserPoolId=get_user_pool_id(user_pool_name), Username=user_id )