Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/azure/core/credentials_async.py: 70%

1# ------------------------------------ 

2# Copyright (c) Microsoft Corporation. 

3# Licensed under the MIT License. 

4# ------------------------------------ 

5from __future__ import annotations 

6from types import TracebackType 

7from typing import Any, Optional, AsyncContextManager, Type, Union, TYPE_CHECKING 

8from typing_extensions import Protocol, runtime_checkable 

9 

10if TYPE_CHECKING: 

11 from .credentials import AccessToken, AccessTokenInfo, TokenRequestOptions 

12 

13 

14@runtime_checkable 

15class AsyncTokenCredential(Protocol, AsyncContextManager["AsyncTokenCredential"]): 

16 """Protocol for classes able to provide OAuth tokens.""" 

17 

18 async def get_token( 

19 self, 

20 *scopes: str, 

21 claims: Optional[str] = None, 

22 tenant_id: Optional[str] = None, 

23 enable_cae: bool = False, 

24 **kwargs: Any, 

25 ) -> AccessToken: 

26 """Request an access token for `scopes`. 

27 

28 :param str scopes: The type of access needed. 

29 

30 :keyword str claims: Additional claims required in the token, such as those returned in a resource 

31 provider's claims challenge following an authorization failure. 

32 :keyword str tenant_id: Optional tenant to include in the token request. 

33 :keyword bool enable_cae: Indicates whether to enable Continuous Access Evaluation (CAE) for the requested 

34 token. Defaults to False. 

35 

36 :rtype: AccessToken 

37 :return: An AccessToken instance containing the token string and its expiration time in Unix time. 

38 """ 

39 ... 

40 

41 async def close(self) -> None: 

42 pass 

43 

44 async def __aexit__( 

45 self, 

46 exc_type: Optional[Type[BaseException]] = None, 

47 exc_value: Optional[BaseException] = None, 

48 traceback: Optional[TracebackType] = None, 

49 ) -> None: 

50 pass 

51 

52 

53@runtime_checkable 

54class AsyncSupportsTokenInfo(Protocol, AsyncContextManager["AsyncSupportsTokenInfo"]): 

55 """Protocol for classes able to provide OAuth access tokens with additional properties.""" 

56 

57 async def get_token_info(self, *scopes: str, options: Optional[TokenRequestOptions] = None) -> AccessTokenInfo: 

58 """Request an access token for `scopes`. 

59 

60 This is an alternative to `get_token` to enable certain scenarios that require additional properties 

61 on the token. 

62 

63 :param str scopes: The type of access needed. 

64 :keyword options: A dictionary of options for the token request. Unknown options will be ignored. Optional. 

65 :paramtype options: TokenRequestOptions 

66 

67 :rtype: AccessTokenInfo 

68 :return: An AccessTokenInfo instance containing the token string and its expiration time in Unix time. 

69 """ 

70 ... 

71 

72 async def close(self) -> None: 

73 pass 

74 

75 async def __aexit__( 

76 self, 

77 exc_type: Optional[Type[BaseException]] = None, 

78 exc_value: Optional[BaseException] = None, 

79 traceback: Optional[TracebackType] = None, 

80 ) -> None: 

81 pass 

82 

83 

84AsyncTokenProvider = Union[AsyncTokenCredential, AsyncSupportsTokenInfo]