Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/redis/credentials.py: 60%

1import logging 

2from abc import ABC, abstractmethod 

3from typing import Any, Callable, Optional, Tuple, Union 

4 

5logger = logging.getLogger(__name__) 

6 

7 

8class CredentialProvider: 

9 """ 

10 Credentials Provider. 

11 """ 

12 

13 def get_credentials(self) -> Union[Tuple[str], Tuple[str, str]]: 

14 raise NotImplementedError("get_credentials must be implemented") 

15 

16 async def get_credentials_async(self) -> Union[Tuple[str], Tuple[str, str]]: 

17 logger.warning( 

18 "This method is added for backward compatability. " 

19 "Please override it in your implementation." 

20 ) 

21 return self.get_credentials() 

22 

23 

24class StreamingCredentialProvider(CredentialProvider, ABC): 

25 """ 

26 Credential provider that streams credentials in the background. 

27 """ 

28 

29 @abstractmethod 

30 def on_next(self, callback: Callable[[Any], None]): 

31 """ 

32 Specifies the callback that should be invoked 

33 when the next credentials will be retrieved. 

34 

35 :param callback: Callback with 

36 :return: 

37 """ 

38 pass 

39 

40 @abstractmethod 

41 def on_error(self, callback: Callable[[Exception], None]): 

42 pass 

43 

44 @abstractmethod 

45 def is_streaming(self) -> bool: 

46 pass 

47 

48 

49class UsernamePasswordCredentialProvider(CredentialProvider): 

50 """ 

51 Simple implementation of CredentialProvider that just wraps static 

52 username and password. 

53 """ 

54 

55 def __init__(self, username: Optional[str] = None, password: Optional[str] = None): 

56 self.username = username or "" 

57 self.password = password or "" 

58 

59 def get_credentials(self): 

60 if self.username: 

61 return self.username, self.password 

62 return (self.password,) 

63 

64 async def get_credentials_async(self) -> Union[Tuple[str], Tuple[str, str]]: 

65 return self.get_credentials()