From f838e438b6fa4eda19d7bab061a9f4a3ef1e45e3 Mon Sep 17 00:00:00 2001 From: box-sdk-build Date: Fri, 1 Sep 2023 01:42:53 -0700 Subject: [PATCH] generated with codegen at box/box-codegen@5aebe76 and spec at box/box-openapi@f01a34e --- box_sdk_gen/auth.py | 9 +++++++-- box_sdk_gen/auth_schemas.py | 20 -------------------- box_sdk_gen/ccg_auth.py | 18 +++++++++++------- box_sdk_gen/developer_token_auth.py | 11 ++++++++--- box_sdk_gen/fetch.py | 5 +++-- box_sdk_gen/jwt_auth.py | 16 ++++++++++------ box_sdk_gen/oauth.py | 19 +++++++++++++------ test/auth.py | 6 ++++-- 8 files changed, 56 insertions(+), 48 deletions(-) diff --git a/box_sdk_gen/auth.py b/box_sdk_gen/auth.py index 89f1348..60926b1 100644 --- a/box_sdk_gen/auth.py +++ b/box_sdk_gen/auth.py @@ -2,13 +2,18 @@ from typing import Optional from .network import NetworkSession +from .schemas import AccessToken class Authentication: @abstractmethod - def retrieve_token(self, network_session: Optional[NetworkSession] = None) -> str: + def retrieve_token( + self, network_session: Optional[NetworkSession] = None + ) -> AccessToken: pass @abstractmethod - def refresh(self, network_session: Optional[NetworkSession] = None) -> str: + def refresh_token( + self, network_session: Optional[NetworkSession] = None + ) -> AccessToken: pass diff --git a/box_sdk_gen/auth_schemas.py b/box_sdk_gen/auth_schemas.py index 87c3f55..eed0996 100644 --- a/box_sdk_gen/auth_schemas.py +++ b/box_sdk_gen/auth_schemas.py @@ -72,23 +72,3 @@ class FileScope(str, Enum): ITEM_PREVIEW = 'item_preview' ITEM_RENAME = 'item_rename' ITEM_SHARE = 'item_share' - - -class AccessToken(BaseObject): - def __init__( - self, - access_token: Union[None, str] = None, - expires_in: Union[None, int] = None, - token_type: Union[None, str] = None, - restricted_to: Union[None, List[FileScope]] = None, - refresh_token: Union[None, str] = None, - issued_token_type: Union[None, str] = None, - **kwargs - ): - super().__init__(**kwargs) - self.access_token = access_token - self.expires_in = expires_in - self.token_type = token_type - self.restricted_to = restricted_to - self.refresh_token = refresh_token - self.issued_token_type = issued_token_type diff --git a/box_sdk_gen/ccg_auth.py b/box_sdk_gen/ccg_auth.py index a1c3715..03f5e75 100644 --- a/box_sdk_gen/ccg_auth.py +++ b/box_sdk_gen/ccg_auth.py @@ -7,10 +7,10 @@ TokenRequestBoxSubjectType, TokenRequest, TokenRequestGrantType, - AccessToken, ) from .fetch import fetch, FetchResponse, FetchOptions from .network import NetworkSession +from .schemas import AccessToken class CCGConfig: @@ -62,7 +62,7 @@ def __init__(self, config: CCGConfig): Configuration object of Client Credentials Grant auth. """ self.config = config - self.token: Union[None, str] = None + self.token: Union[None, AccessToken] = None if config.user_id: self.subject_id = self.config.user_id @@ -71,17 +71,21 @@ def __init__(self, config: CCGConfig): self.subject_type = TokenRequestBoxSubjectType.ENTERPRISE self.subject_id = self.config.enterprise_id - def retrieve_token(self, network_session: Optional[NetworkSession] = None) -> str: + def retrieve_token( + self, network_session: Optional[NetworkSession] = None + ) -> AccessToken: """ Return a current token or get a new one when not available. :return: Access token """ if self.token is None: - return self.refresh(network_session=network_session) + self.refresh_token(network_session=network_session) return self.token - def refresh(self, network_session: Optional[NetworkSession] = None) -> str: + def refresh_token( + self, network_session: Optional[NetworkSession] = None + ) -> AccessToken: """ Fetch a new access token :return: @@ -106,8 +110,8 @@ def refresh(self, network_session: Optional[NetworkSession] = None) -> str: ) token_response = AccessToken.from_dict(json.loads(response.text)) - self.token = token_response.access_token - return self.token + self.token = token_response + return token_response def as_user(self, user_id: str): """ diff --git a/box_sdk_gen/developer_token_auth.py b/box_sdk_gen/developer_token_auth.py index b18e2a0..9ed8506 100644 --- a/box_sdk_gen/developer_token_auth.py +++ b/box_sdk_gen/developer_token_auth.py @@ -2,14 +2,19 @@ from .auth import Authentication from .network import NetworkSession +from .schemas import AccessToken class DeveloperTokenAuth(Authentication): def __init__(self, token: str): - self.token: str = token + self.token: AccessToken = AccessToken(access_token=token) - def retrieve_token(self, network_session: Optional[NetworkSession] = None) -> str: + def retrieve_token( + self, network_session: Optional[NetworkSession] = None + ) -> AccessToken: return self.token - def refresh(self, network_session: Optional[NetworkSession] = None) -> str: + def refresh_token( + self, network_session: Optional[NetworkSession] = None + ) -> AccessToken: raise Exception("Developer token has expired. Please provide a new one.") diff --git a/box_sdk_gen/fetch.py b/box_sdk_gen/fetch.py index 82245a1..ea6176f 100644 --- a/box_sdk_gen/fetch.py +++ b/box_sdk_gen/fetch.py @@ -124,7 +124,7 @@ def fetch(url: str, options: FetchOptions) -> FetchResponse: ) if response.reauthentication_needed: - options.auth.refresh(options.network_session) + options.auth.refresh_token(options.network_session) elif response.status_code != 429 and response.status_code < 500: __raise_on_unsuccessful_request( network_response=response.network_response, @@ -161,7 +161,8 @@ def __compose_headers_for_request(options: FetchOptions) -> Dict[str, str]: headers = options.headers or {} if options.auth: headers['Authorization'] = ( - f'Bearer {options.auth.retrieve_token(options.network_session)}' + 'Bearer' + f' {options.auth.retrieve_token(options.network_session).access_token}' ) headers['User-Agent'] = USER_AGENT_HEADER diff --git a/box_sdk_gen/jwt_auth.py b/box_sdk_gen/jwt_auth.py index ff91999..1cc0462 100644 --- a/box_sdk_gen/jwt_auth.py +++ b/box_sdk_gen/jwt_auth.py @@ -18,10 +18,10 @@ TokenRequestBoxSubjectType, TokenRequest, TokenRequestGrantType, - AccessToken, ) from .fetch import fetch, FetchResponse, FetchOptions from .network import NetworkSession +from .schemas import AccessToken class JWTConfig: @@ -142,7 +142,7 @@ def __init__(self, config: JWTConfig): ) self.config = config - self.token = None + self.token: Union[None, AccessToken] = None if config.enterprise_id: self.subject_type = TokenRequestBoxSubjectType.ENTERPRISE @@ -155,17 +155,21 @@ def __init__(self, config: JWTConfig): config.private_key, config.private_key_passphrase ) - def retrieve_token(self, network_session: Optional[NetworkSession] = None) -> str: + def retrieve_token( + self, network_session: Optional[NetworkSession] = None + ) -> AccessToken: """ Return a current token or get a new one when not available. :return: Access token """ if self.token is None: - return self.refresh(network_session=network_session) + self.refresh_token(network_session=network_session) return self.token - def refresh(self, network_session: Optional[NetworkSession] = None) -> str: + def refresh_token( + self, network_session: Optional[NetworkSession] = None + ) -> AccessToken: """ Fetch a new access token :return: @@ -215,7 +219,7 @@ def refresh(self, network_session: Optional[NetworkSession] = None) -> str: ) token_response = AccessToken.from_dict(json.loads(response.text)) - self.token = token_response.access_token + self.token = token_response return self.token def as_user(self, user_id: str): diff --git a/box_sdk_gen/oauth.py b/box_sdk_gen/oauth.py index ad7006a..c242e75 100644 --- a/box_sdk_gen/oauth.py +++ b/box_sdk_gen/oauth.py @@ -3,9 +3,10 @@ from typing import Union, Optional from .auth import Authentication -from .auth_schemas import TokenRequest, TokenRequestGrantType, AccessToken +from .auth_schemas import TokenRequest, TokenRequestGrantType from .fetch import fetch, FetchResponse, FetchOptions from .network import NetworkSession +from .schemas import AccessToken class OAuthConfig: @@ -120,7 +121,9 @@ def get_tokens_authorization_code_grant( self.token = self._send_token_request(request_body, network_session) return self.token.access_token - def retrieve_token(self, network_session: Optional[NetworkSession] = None) -> str: + def retrieve_token( + self, network_session: Optional[NetworkSession] = None + ) -> AccessToken: """ Return a current token or get a new one when not available. :param network_session: An object to keep network session state @@ -131,9 +134,13 @@ def retrieve_token(self, network_session: Optional[NetworkSession] = None) -> st "Access and refresh tokens not available. Authenticate before making" " any API call first." ) - return self.token.access_token + return self.token - def refresh(self, network_session: Optional[NetworkSession] = None) -> str: + def refresh_token( + self, + network_session: Optional[NetworkSession] = None, + refresh_token: Optional[str] = None, + ) -> AccessToken: """ Refresh outdated access token with refresh token :param network_session: An object to keep network session state @@ -143,11 +150,11 @@ def refresh(self, network_session: Optional[NetworkSession] = None) -> str: grant_type=TokenRequestGrantType.REFRESH_TOKEN, client_id=self.config.client_id, client_secret=self.config.client_secret, - refresh_token=self.token.refresh_token, + refresh_token=refresh_token or self.token.refresh_token, ) self.token = self._send_token_request(request_body, network_session) - return self.token.access_token + return self.token @staticmethod def _send_token_request( diff --git a/test/auth.py b/test/auth.py index f9c4d66..7eab004 100644 --- a/test/auth.py +++ b/test/auth.py @@ -1,3 +1,5 @@ +from box_sdk_gen.schemas import AccessToken + from box_sdk_gen.utils import decode_base_64 from box_sdk_gen.utils import get_env_var @@ -65,8 +67,8 @@ def test_developer_token_auth(): ) auth: JWTAuth = JWTAuth(config=jwt_config) auth.as_user(user_id) - token: str = auth.retrieve_token() - dev_auth: DeveloperTokenAuth = DeveloperTokenAuth(token=token) + token: AccessToken = auth.retrieve_token() + dev_auth: DeveloperTokenAuth = DeveloperTokenAuth(token=token.access_token) client: Client = Client(auth=dev_auth) current_user: UserFull = client.users.get_user_me() assert current_user.id == user_id