app/core/cognito.py

from botocore.client import BaseClient
from botocore.exceptions import BotoCoreError, ClientError
from dotenv import dotenv_values
from fastapi import UploadFile

from app.core.aws import get_session

_PATH_ENV = "cognito.env"


class Cognito:
    @staticmethod
    def client(file: UploadFile) -> BaseClient:
        session = get_session(_PATH_ENV, file)
        return session.client("cognito-idp")

    @staticmethod
    def auth(client: BaseClient, username: str, password: str):
        config = dotenv_values(dotenv_path=_PATH_ENV)
        try:
            return client.initiate_auth(
                ClientId=config["AWS_CLIENT_ID_COGNITO"],
                AuthFlow='USER_PASSWORD_AUTH',
                AuthParameters={
                    'USERNAME': username,
                    'PASSWORD': password
                },
            )
        except (BotoCoreError, ClientError) as error:
            if isinstance(error, ClientError):
                return error.response
            else:
                return {"Error": error}