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

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

22 statements  

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 """Close the credential, releasing any resources. 

43 

44 :return: None 

45 :rtype: None 

46 """ 

47 

48 async def __aexit__( 

49 self, 

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

51 exc_value: Optional[BaseException] = None, 

52 traceback: Optional[TracebackType] = None, 

53 ) -> None: 

54 pass 

55 

56 

57@runtime_checkable 

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

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

60 

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

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

63 

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

65 on the token. 

66 

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

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

69 :paramtype options: TokenRequestOptions 

70 

71 :rtype: AccessTokenInfo 

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

73 """ 

74 ... 

75 

76 async def close(self) -> None: 

77 """Close the credential, releasing any resources. 

78 

79 :return: None 

80 :rtype: None 

81 """ 

82 

83 async def __aexit__( 

84 self, 

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

86 exc_value: Optional[BaseException] = None, 

87 traceback: Optional[TracebackType] = None, 

88 ) -> None: 

89 pass 

90 

91 

92AsyncTokenProvider = Union[AsyncTokenCredential, AsyncSupportsTokenInfo]