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]