+[docs]
+ asyncdefrefresh_token(self):
+"""
+ Following the specification, the token response for the client credentials grant SHOULD NOT include a refresh token.
+ The client credentials grant should be used to get a new access token when the previous one has expired.
+
+ https://datatracker.ietf.org/doc/html/rfc6749#section-4.4.3
+
+ Some clients may issue a refresh token for the client credentials flow, even though it is not correct according to the specification.
+ In this case, the refresh token will be used to obtain a new access token.
+ """
+ if"refresh_token"inself.token:
+ # use refresh token when it is available
+ awaitsuper().refresh_token()
+
+ # just get a new token using the client credentials
+ awaitself.fetch_token()
+[docs]
+ asyncdefclose(self):
+"""
+ Close the Grant object and its associated resources.
+ """
+ awaitself.session.close()
+
+
+
+[docs]
+ asyncdefensure_active_token(self):
+"""
+ Ensure that the stored access token is still active.
+ If this is not the case, the token will be refreshed.
+ """
+ asyncwithself.lock:
+ ifself.token.is_expired():
+ awaitself.refresh_token()
+[docs]
+ asyncdeffetch_token(self):
+"""
+ Fetch an OAuth 2.0 token from the token endpoint and store it for subsequent use.
+ """
+ self.token=awaitself._fetch_token()
+
+
+ @abstractmethod
+ asyncdef_fetch_token(self)->Token:
+"""
+ Fetch an OAuth 2.0 token from the token endpoint.
+ :return: OAuth 2.0 Token
+ """
+ ...
+
+
+[docs]
+ asyncdefrefresh_token(self):
+"""
+ Obtain a new access token using the refresh token grant and store it for subsequent use.
+ """
+ access_token_request=RefreshTokenAccessTokenRequest(
+ refresh_token=self.token.refresh_token,
+ **self.kwargs,
+ )
+ self.token=awaitself.execute_token_request(access_token_request)
+
+
+
+[docs]
+ asyncdefexecute_token_request(self,data:AccessTokenRequest)->Token:
+"""
+ Execute a token request with the provided data.
+
+ :param data: token request data
+ :return: OAuth 2.0 Token
+ :raises OAuth2Error: if the token request fails
+ :raises aiohttp.ClientResponseError: if the HTTP error cannot be parsed as an OAuth 2.0 error response
+ """
+ asyncwithself.session.post(
+ url=self.token_url,
+ data=data.model_dump(exclude_none=True),
+ )asresponse:
+ ifnotresponse.ok:
+ try:
+ raiseOAuth2Error(
+ ErrorResponse.model_validate(awaitresponse.json())
+ )
+ exceptValidationError:
+ response.raise_for_status()
+ returnToken.model_validate(awaitresponse.json())
+[docs]
+classAccessTokenRequest(BaseModel):
+"""
+ Base model for OAuth 2.0 access token request.
+ """
+
+ grant_type:str
+"""OAuth 2.0 grant type"""
+
+ model_config=ConfigDict(extra="allow")
+
+
+
+
+[docs]
+classAuthorizationCodeAccessTokenRequest(AccessTokenRequest):
+"""
+ Request model for the access token request with the Authorization Code grant.
+
+ https://datatracker.ietf.org/doc/html/rfc6749#section-4.1.3
+ """
+
+ grant_type:str=GrantType.AUTHORIZATION_CODE
+
+ code:str
+"""Authorization code received from the authorization server"""
+
+ redirect_uri:Optional[str]=None
+"""Redirect URI"""
+
+ client_id:str
+"""Client identifier"""
+
+
+
+
+[docs]
+classClientCredentialsAccessTokenRequest(AccessTokenRequest):
+"""
+ Request model for the access token request with the Client Credentials grant.
+
+ https://datatracker.ietf.org/doc/html/rfc6749#section-4.4.2
+ """
+
+ grant_type:str=GrantType.CLIENT_CREDENTIALS
+
+ client_id:str
+"""Client identifier"""
+
+ client_secret:str
+"""Client secret"""
+
+ scope:Optional[str]=None
+"""Scope of the access request"""
+
+
+
+
+[docs]
+classResourceOwnerPasswordCredentialsAccessTokenRequest(AccessTokenRequest):
+"""
+ Request model for the access token request with the Resource Owner Password Credentials grant.
+
+ https://datatracker.ietf.org/doc/html/rfc6749#section-4.3.2
+ """
+
+ grant_type:str=GrantType.RESOURCE_OWNER_PASSWORD_CREDENTIALS
+
+ username:str
+"""Resource owner username"""
+
+ password:str
+"""Resource owner password"""
+
+ scope:Optional[str]=None
+"""Scope of the access request"""
+
+
+
+
+[docs]
+classRefreshTokenAccessTokenRequest(AccessTokenRequest):
+"""
+ Request model for the access token request using a Refresh Token.
+
+ https://datatracker.ietf.org/doc/html/rfc6749#section-6
+ """
+
+ grant_type:str=GrantType.REFRESH_TOKEN
+
+ refresh_token:str
+"""Refresh token"""
+
+ scope:Optional[str]=None
+"""Scope of the access request"""
+
+
+
+
+[docs]
+classDeviceAccessTokenRequest(AccessTokenRequest):
+"""
+ The Device Access Token Request model.
+
+ https://datatracker.ietf.org/doc/html/rfc8628#section-3.4
+
+ :ivar grant_type: The grant type. Value MUST be set to "urn:ietf:params:oauth:grant-type:device_code".
+ :ivar device_code: The device verification code, "device_code" from the device authorization response.
+ :ivar client_id: The client identifier.
+ """
+
+ grant_type:str=GrantType.DEVICE_CODE
+
+ device_code:str
+"""Device verification code"""
+
+ client_id:str
+"""Client identifier"""
+
+
+
+
+[docs]
+classAuthorizationRequest(BaseModel):
+"""
+ The Authorization Request model.
+
+ https://datatracker.ietf.org/doc/html/rfc6749#section-4.1.1
+ """
+
+ response_type:str="code"
+
+ client_id:str
+"""Client identifier"""
+
+ redirect_uri:Optional[str]=None
+"""Redirect URI"""
+
+ scope:Optional[str]=None
+"""Scope of the access request"""
+
+ state:Optional[str]=None
+"""Opaque value used by the client to maintain state between the request and callback"""
+[docs]
+classAuthorizationResponse(BaseModel):
+"""
+ The Authorization Response model.
+
+ https://datatracker.ietf.org/doc/html/rfc6749#section-4.1.2
+
+ :ivar code: The authorization code generated by the authorization server.
+ :ivar state: State
+ """
+
+ code:str
+"""Authorization code generated by the authorization server"""
+
+ state:Optional[str]=None
+"""State parameter from the client authorization request, if it was present"""
+
+
+
+
+[docs]
+classDeviceAuthorizationResponse(BaseModel):
+"""
+ The Device Authorization Response model.
+
+ https://datatracker.ietf.org/doc/html/rfc8628#section-3.2
+ """
+
+ device_code:str
+"""Device verification code"""
+
+ user_code:str
+"""End-user verification code"""
+
+ verification_uri:str
+"""End-user verification URI on the authorization server"""
+
+ verification_uri_complete:Optional[str]=None
+"""Verification URI that includes the "user_code" which is designed for non-textual transmission."""
+
+ expires_in:int
+"""Lifetime in seconds of the "device_code" and "user_code"."""
+
+ interval:int=5
+"""Minimum amount of time in seconds that the client SHOULD wait between polling requests to the token endpoint."""
+
+
+
+
+[docs]
+classErrorResponse(BaseModel):
+"""
+ OAuth2 Error Response model.
+
+ https://datatracker.ietf.org/doc/html/rfc6749#section-5.2
+ """
+
+ error:str
+"""A single ASCII error code"""
+
+ error_description:Optional[str]=None
+"""Human-readable ASCII text providing additional information, used to assist the client developer in understanding the error that occurred."""
+
+ error_uri:Optional[str]=None
+"""A URI identifying a human-readable web page with information about the error, used to provide the client developer with additional information about the error."""
+[docs]
+classToken(BaseModel):
+"""
+ Token Response model.
+
+ https://datatracker.ietf.org/doc/html/rfc6749#section-5.1
+ """
+
+ access_token:str
+"""Access token issued by the authorization server"""
+
+ token_type:str
+"""Type of the token issued"""
+
+ expires_in:Optional[int]=None
+"""Lifetime in seconds of the access token"""
+
+ refresh_token:Optional[str]=None
+"""Refresh token, which can be used to obtain new access tokens"""
+
+ scope:Optional[str]=None
+"""Scope of the access token"""
+
+ expires_at:int
+"""Expiration time of the access token"""
+
+ model_config=ConfigDict(extra="allow")
+
+ @model_validator(mode="before")
+ @classmethod
+ def_validate_expires_at(cls,data):
+"""
+ If the 'expires_at' field is not provided,
+ it will be computed based on the current time and the 'expires_in' field.
+ """
+ ifisinstance(data,dict):
+ if"expires_at"notindata:
+ data["expires_at"]=int(time.time())+data["expires_in"]
+ returndata
+
+
+[docs]
+ defis_expired(self,early_expiry:int=30)->bool:
+"""
+ Indicates whether the access token is expired.
+ The early expiry parameter allows to have some margin on the token expiry time.
+
+ :param early_expiry: early expiry in seconds
+ :return: True if the access token is valid given the early expiry, False otherwise
+ """
+ ifnotself.expires_at:
+ raiseValueError("No token expiration information")
+ returnself.expires_at-early_expiry<time.time()