m2_g2_fs/usr/local/lib/python3.10/dist-packages/redis/credentials.py

import logging
from abc import ABC, abstractmethod
from typing import Any, Callable, Optional, Tuple, Union

logger = logging.getLogger(__name__)


class CredentialProvider:
    """
    Credentials Provider.
    """

    def get_credentials(self) -> Union[Tuple[str], Tuple[str, str]]:
        raise NotImplementedError("get_credentials must be implemented")

    async def get_credentials_async(self) -> Union[Tuple[str], Tuple[str, str]]:
        logger.warning(
            "This method is added for backward compatability. "
            "Please override it in your implementation."
        )
        return self.get_credentials()


class StreamingCredentialProvider(CredentialProvider, ABC):
    """
    Credential provider that streams credentials in the background.
    """

    @abstractmethod
    def on_next(self, callback: Callable[[Any], None]):
        """
        Specifies the callback that should be invoked
        when the next credentials will be retrieved.

        :param callback: Callback with
        :return:
        """
        pass

    @abstractmethod
    def on_error(self, callback: Callable[[Exception], None]):
        pass

    @abstractmethod
    def is_streaming(self) -> bool:
        pass


class UsernamePasswordCredentialProvider(CredentialProvider):
    """
    Simple implementation of CredentialProvider that just wraps static
    username and password.
    """

    def __init__(self, username: Optional[str] = None, password: Optional[str] = None):
        self.username = username or ""
        self.password = password or ""

    def get_credentials(self):
        if self.username:
            return self.username, self.password
        return (self.password,)

    async def get_credentials_async(self) -> Union[Tuple[str], Tuple[str, str]]:
        return self.get_credentials()