Cognito Snippet

CognitoService.py

import boto3


class CognitoService():

    def __init__(self, cognitoPoolId, client_id):
        self.cognitoPoolId = cognitoPoolId
        self.client = boto3.client('cognito-idp')
        self.client_id = client_id

    def createUser(self, params):
        attributes = []
        attribute = {}
        if 'attributes' in params:
            print(params)
            for param in params['attributes']:
                attribute['Name'] = param
                attribute['Value'] = params['attributes'][param]
                attributes.append(attribute.copy())
        attributes.append({
            'Name': 'email',
            'Value': params['email']
        })
        response = self.client.admin_create_user(
            UserPoolId = self.cognitoPoolId,
            Username = params['username'],
            UserAttributes = attributes,
            TemporaryPassword = params['password'],
            DesiredDeliveryMediums = ['EMAIL']
        )
        return response

    def updateUser(self, username, params):
        if not params:
            print('Nothing to update')
            return
        attributes = []
        attribute = {}
        for param in params:
            attribute['Name'] = param
            attribute['Value'] = params[param]
            attributes.append(attribute.copy())

        response = self.client.admin_update_user_attributes(
            UserPoolId = self.cognitoPoolId,
            Username = username,
            UserAttributes = attributes
        )
        return response

    def login_user(self, username, password):
        response = self.client.admin_initiate_auth(
            UserPoolId=self.cognitoPoolId,
            ClientId=self.client_id,
            AuthFlow="ADMIN_NO_SRP_AUTH",
            AuthParameters = {
                "USERNAME":username,
                "PASSWORD":password
            }
        )
        if response:
            if "AuthenticationResult" in response:
                return response["AuthenticationResult"]["AccessToken"]
        return

    def confirm_user(self, username):
        response = self.client.admin_confirm_sign_up(
            UserPoolId=self.cognitoPoolId,
            Username=username
        )
        if response:
            return "User authenticated!"
        return "User not confirmed"
    
    def post_confirmation(self, username, group):
        # add user to 'demo' group
        response = self.client.admin_add_user_to_group(
            UserPoolId=self.cognitoPoolId,
            Username=username,
            GroupName=group
        )
        return response