Upload file to S3 with an authenticated Cognito User

cognitoupload.py

import boto3
import uuid

class CognitoUserFileUploader(object):

    def __init__(self, *args, **kwargs):
        self.__dict__.update(kwargs)
        self.id_token = self.get_cognito_id_token(
            self.username, self.refresh_token,
            self.device_key, self.client_id
        )
        self.identity_id = self.get_identity_id(
            self.account_id, self.identity_pool_id,
            self.provider_name, self.id_token
        )
        self.aws_credentials = self.get_credentials(
            self.identity_id, self.provider_name, self.id_token
        )

    def get_cognito_id_token(self, username, refresh_token,
                             device_key, client_id):
        client = boto3.client('cognito-idp', region_name=self.region_name)
        response = client.initiate_auth(
            AuthFlow='REFRESH_TOKEN',
            AuthParameters={
                'USERNAME': username,
                'REFRESH_TOKEN': refresh_token,
                'DEVICE_KEY': device_key
            },
            ClientId=client_id
        )
        return response['AuthenticationResult']['IdToken']

    def get_identity_id(self, account_id, identity_pool_id,
                        provider_name, id_token):
        client = boto3.client('cognito-identity', region_name=self.region_name)
        creds = client.get_id(
            AccountId=account_id, IdentityPoolId=identity_pool_id,
            Logins={provider_name: id_token}
        )
        return creds['IdentityId']

    def get_credentials(self, identity_id, provider_name, id_token):
        client = boto3.client('cognito-identity', region_name=self.region_name)
        creds = client.get_credentials_for_identity(
            IdentityId=identity_id,
            Logins={provider_name: id_token},
        )
        return creds['Credentials']

    def upload_file(self, file_path, bucket_name):
        prefix = self.s3_key_prefix or ''
        key = str(uuid.uuid4())
        s3_client = boto3.client(
            's3',
            aws_access_key_id=self.aws_credentials['AccessKeyId'],
            aws_secret_access_key=self.aws_credentials['SecretKey'],
            aws_session_token=self.aws_credentials['SessionToken'],
        )
        key = '/'.join([prefix, key])
        return s3_client.upload_file(file_path, bucket_name, key)


file_uploader = CognitoUserFileUploader(
    region_name='ap-southeast-2',
    refresh_token=REFRESH_TOKEN,
    username=USERNAME,
    device_key=DEVICE_KEY,
    client_id=CLIENT_ID,
    account_id=ACCOUNT_ID,
    identity_pool_id=IDENTITY_POOL_ID,
    provider_name=PROVIDER_NAME,
    s3_key_prefix='myfolder'
)

file_path = '~/Desktop/testing.png'
file_uploader.upload_file(file_path, bucket_name=BUCKET_NAME)