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]