POC

box_sdk_gen/__init__.py

@@ -0,0 +1,15 @@
+from box_sdk_gen.box import *
+
+from box_sdk_gen.serialization import *
+
+from box_sdk_gen.internal import *
+
+from box_sdk_gen.networking import *
+
+from box_sdk_gen.schemas import *
+
+from box_sdk_gen.parameters import *
+
+from box_sdk_gen.managers import *
+
+from box_sdk_gen.client import *

box_sdk_gen/box/__init__.py

@@ -0,0 +1,11 @@
+from box_sdk_gen.box.errors import *
+
+from box_sdk_gen.box.token_storage import *
+
+from box_sdk_gen.box.developer_token_auth import *
+
+from box_sdk_gen.box.oauth import *
+
+from box_sdk_gen.box.jwt_auth import *
+
+from box_sdk_gen.box.ccg_auth import *

box_sdk_gen/box/ccg_auth.py

@@ -0,0 +1,149 @@
+from typing import Optional
+
+from typing import List
+
+from box_sdk_gen.schemas.post_o_auth_2_token import PostOAuth2TokenGrantTypeField
+
+from box_sdk_gen.schemas.post_o_auth_2_token import PostOAuth2TokenSubjectTokenTypeField
+
+from box_sdk_gen.schemas.access_token import AccessToken
+
+from box_sdk_gen.schemas.post_o_auth_2_token import PostOAuth2TokenBoxSubjectTypeField
+
+from box_sdk_gen.networking.auth import Authentication
+
+from box_sdk_gen.networking.network import NetworkSession
+
+from box_sdk_gen.box.token_storage import TokenStorage
+
+from box_sdk_gen.box.token_storage import InMemoryTokenStorage
+
+from box_sdk_gen.managers.authorization import AuthorizationManager
+
+from box_sdk_gen.box.errors import BoxSDKError
+
+from box_sdk_gen.schemas.post_o_auth_2_token import PostOAuth2Token
+
+from box_sdk_gen.schemas.post_o_auth_2_revoke import PostOAuth2Revoke
+
+class CCGConfig:
+    def __init__(self, client_id: str, client_secret: str, *, enterprise_id: Optional[str] = None, user_id: Optional[str] = None, token_storage: TokenStorage = None):
+        """
+        :param client_id: Box API key used for identifying the application the user is authenticating with
+        :type client_id: str
+        :param client_secret: Box API secret used for making auth requests.
+        :type client_secret: str
+        :param enterprise_id: The ID of the Box Developer Edition enterprise., defaults to None
+        :type enterprise_id: Optional[str], optional
+        :param user_id: The user id to authenticate. This value is not required. But if it is provided, then the user will be auto-authenticated at the time of the first API call., defaults to None
+        :type user_id: Optional[str], optional
+        :param token_storage: Object responsible for storing token. If no custom implementation provided,the token will be stored in memory., defaults to None
+        :type token_storage: TokenStorage, optional
+        """
+        if token_storage is None:
+            token_storage = InMemoryTokenStorage()
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.enterprise_id = enterprise_id
+        self.user_id = user_id
+        self.token_storage = token_storage
+
+class BoxCCGAuth(Authentication):
+    def __init__(self, config: CCGConfig, **kwargs):
+        """
+        :param config: Configuration object of Client Credentials Grant auth.
+        :type config: CCGConfig
+        """
+        super().__init__(**kwargs)
+        self.config = config
+        self.token_storage = self.config.token_storage
+        self.subject_id = self.config.user_id if not self.config.user_id == None else self.config.enterprise_id
+        self.subject_type = PostOAuth2TokenBoxSubjectTypeField.USER if not self.config.user_id == None else PostOAuth2TokenBoxSubjectTypeField.ENTERPRISE
+    def refresh_token(self, *, network_session: Optional[NetworkSession] = None) -> AccessToken:
+        """
+        Get a new access token using CCG auth
+        :param network_session: An object to keep network session state, defaults to None
+        :type network_session: Optional[NetworkSession], optional
+        """
+        auth_manager: AuthorizationManager = AuthorizationManager(network_session=network_session if not network_session == None else NetworkSession())
+        token: AccessToken = auth_manager.request_access_token(PostOAuth2TokenGrantTypeField.CLIENT_CREDENTIALS, client_id=self.config.client_id, client_secret=self.config.client_secret, box_subject_type=self.subject_type, box_subject_id=self.subject_id)
+        self.token_storage.store(token)
+        return token
+    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, defaults to None
+        :type network_session: Optional[NetworkSession], optional
+        """
+        old_token: Optional[AccessToken] = self.token_storage.get()
+        if old_token == None:
+            new_token: AccessToken = self.refresh_token(network_session=network_session)
+            return new_token
+        return old_token
+    def retrieve_authorization_header(self, *, network_session: Optional[NetworkSession] = None) -> str:
+        token: AccessToken = self.retrieve_token(network_session=network_session)
+        return ''.join(['Bearer ', token.access_token])
+    def with_user_subject(self, user_id: str, *, token_storage: TokenStorage = None) -> 'BoxCCGAuth':
+        """
+        Create a new BoxCCGAuth instance that uses the provided user ID as the subject ID.
+
+        May be one of this application's created App User. Depending on the configured User Access Level, may also be any other App User or Managed User in the enterprise.
+
+
+        <https://developer.box.com/en/guides/applications/>
+
+
+        <https://developer.box.com/en/guides/authentication/select/>
+
+        :param user_id: The id of the user to authenticate
+        :type user_id: str
+        :param token_storage: Object responsible for storing token in newly created BoxCCGAuth. If no custom implementation provided, the token will be stored in memory., defaults to None
+        :type token_storage: TokenStorage, optional
+        """
+        if token_storage is None:
+            token_storage = InMemoryTokenStorage()
+        new_config: CCGConfig = CCGConfig(client_id=self.config.client_id, client_secret=self.config.client_secret, enterprise_id=self.config.enterprise_id, user_id=user_id, token_storage=token_storage)
+        return BoxCCGAuth(config=new_config)
+    def with_enterprise_subject(self, enterprise_id: str, *, token_storage: TokenStorage = None) -> 'BoxCCGAuth':
+        """
+        Create a new BoxCCGAuth instance that uses the provided enterprise ID as the subject ID.
+        :param enterprise_id: The id of the enterprise to authenticate
+        :type enterprise_id: str
+        :param token_storage: Object responsible for storing token in newly created BoxCCGAuth. If no custom implementation provided, the token will be stored in memory., defaults to None
+        :type token_storage: TokenStorage, optional
+        """
+        if token_storage is None:
+            token_storage = InMemoryTokenStorage()
+        new_config: CCGConfig = CCGConfig(client_id=self.config.client_id, client_secret=self.config.client_secret, enterprise_id=enterprise_id, user_id=None, token_storage=token_storage)
+        return BoxCCGAuth(config=new_config)
+    def downscope_token(self, scopes: List[str], *, resource: Optional[str] = None, shared_link: Optional[str] = None, network_session: Optional[NetworkSession] = None) -> AccessToken:
+        """
+        Downscope access token to the provided scopes. Returning a new access token with the provided scopes, with the original access token unchanged.
+        :param scopes: The scope(s) to apply to the resulting token.
+        :type scopes: List[str]
+        :param resource: The file or folder to get a downscoped token for. If None and shared_link None, the resulting token will not be scoped down to just a single item. The resource should be a full URL to an item, e.g. https://api.box.com/2.0/files/123456., defaults to None
+        :type resource: Optional[str], optional
+        :param shared_link: The shared link to get a downscoped token for. If None and item None, the resulting token will not be scoped down to just a single item., defaults to None
+        :type shared_link: Optional[str], optional
+        :param network_session: An object to keep network session state, defaults to None
+        :type network_session: Optional[NetworkSession], optional
+        """
+        token: Optional[AccessToken] = self.retrieve_token(network_session=network_session)
+        if token == None:
+            raise BoxSDKError(message='No access token is available. Make an API call to retrieve a token before calling this method.')
+        auth_manager: AuthorizationManager = AuthorizationManager(network_session=network_session if not network_session == None else NetworkSession())
+        downscoped_token: AccessToken = auth_manager.request_access_token(PostOAuth2TokenGrantTypeField.URN_IETF_PARAMS_OAUTH_GRANT_TYPE_TOKEN_EXCHANGE, subject_token=token.access_token, subject_token_type=PostOAuth2TokenSubjectTokenTypeField.URN_IETF_PARAMS_OAUTH_TOKEN_TYPE_ACCESS_TOKEN, resource=resource, scope=' '.join(scopes), box_shared_link=shared_link)
+        return downscoped_token
+    def revoke_token(self, *, network_session: Optional[NetworkSession] = None) -> None:
+        """
+        Revoke the current access token and remove it from token storage.
+        :param network_session: An object to keep network session state, defaults to None
+        :type network_session: Optional[NetworkSession], optional
+        """
+        old_token: Optional[AccessToken] = self.token_storage.get()
+        if old_token == None:
+            return None
+        auth_manager: AuthorizationManager = AuthorizationManager(network_session=network_session if not network_session == None else NetworkSession())
+        auth_manager.revoke_access_token(client_id=self.config.client_id, client_secret=self.config.client_secret, token=old_token.access_token)
+        self.token_storage.clear()
+        return None

box_sdk_gen/box/developer_token_auth.py

@@ -0,0 +1,92 @@
+from typing import Optional
+
+from typing import List
+
+from box_sdk_gen.schemas.post_o_auth_2_token import PostOAuth2TokenGrantTypeField
+
+from box_sdk_gen.schemas.post_o_auth_2_token import PostOAuth2TokenSubjectTokenTypeField
+
+from box_sdk_gen.schemas.access_token import AccessToken
+
+from box_sdk_gen.networking.auth import Authentication
+
+from box_sdk_gen.networking.network import NetworkSession
+
+from box_sdk_gen.box.errors import BoxSDKError
+
+from box_sdk_gen.box.token_storage import TokenStorage
+
+from box_sdk_gen.box.token_storage import InMemoryTokenStorage
+
+from box_sdk_gen.managers.authorization import AuthorizationManager
+
+from box_sdk_gen.schemas.post_o_auth_2_token import PostOAuth2Token
+
+from box_sdk_gen.schemas.post_o_auth_2_revoke import PostOAuth2Revoke
+
+class DeveloperTokenConfig:
+    def __init__(self, *, client_id: Optional[str] = None, client_secret: Optional[str] = None):
+        self.client_id = client_id
+        self.client_secret = client_secret
+
+class BoxDeveloperTokenAuth(Authentication):
+    def __init__(self, token: str, *, config: DeveloperTokenConfig = None, **kwargs):
+        """
+        :param config: Configuration object of DeveloperTokenAuth., defaults to None
+        :type config: DeveloperTokenConfig, optional
+        """
+        super().__init__(**kwargs)
+        self.token = token
+        self.config = config
+        self.token_storage = InMemoryTokenStorage(token=AccessToken(access_token=self.token))
+    def retrieve_token(self, *, network_session: Optional[NetworkSession] = None) -> AccessToken:
+        """
+        Retrieves stored developer token
+        :param network_session: An object to keep network session state, defaults to None
+        :type network_session: Optional[NetworkSession], optional
+        """
+        token: Optional[AccessToken] = self.token_storage.get()
+        if token == None:
+            raise BoxSDKError(message='No access token is available.')
+        return token
+    def refresh_token(self, *, network_session: Optional[NetworkSession] = None) -> AccessToken:
+        """
+        Developer token cannot be refreshed
+        :param network_session: An object to keep network session state, defaults to None
+        :type network_session: Optional[NetworkSession], optional
+        """
+        raise BoxSDKError(message='Developer token has expired. Please provide a new one.')
+    def retrieve_authorization_header(self, *, network_session: Optional[NetworkSession] = None) -> str:
+        token: AccessToken = self.retrieve_token(network_session=network_session)
+        return ''.join(['Bearer ', token.access_token])
+    def revoke_token(self, *, network_session: Optional[NetworkSession] = None) -> None:
+        """
+        Revoke an active Access Token, effectively logging a user out that has been previously authenticated.
+        :param network_session: An object to keep network session state, defaults to None
+        :type network_session: Optional[NetworkSession], optional
+        """
+        token: Optional[AccessToken] = self.token_storage.get()
+        if token == None:
+            return None
+        auth_manager: AuthorizationManager = AuthorizationManager(network_session=network_session if not network_session == None else NetworkSession())
+        auth_manager.revoke_access_token(client_id=self.config.client_id, client_secret=self.config.client_secret, token=token.access_token)
+        self.token_storage.clear()
+        return None
+    def downscope_token(self, scopes: List[str], *, resource: Optional[str] = None, shared_link: Optional[str] = None, network_session: Optional[NetworkSession] = None) -> AccessToken:
+        """
+        Downscope access token to the provided scopes. Returning a new access token with the provided scopes, with the original access token unchanged.
+        :param scopes: The scope(s) to apply to the resulting token.
+        :type scopes: List[str]
+        :param resource: The file or folder to get a downscoped token for. If None and shared_link None, the resulting token will not be scoped down to just a single item. The resource should be a full URL to an item, e.g. https://api.box.com/2.0/files/123456., defaults to None
+        :type resource: Optional[str], optional
+        :param shared_link: The shared link to get a downscoped token for. If None and item None, the resulting token will not be scoped down to just a single item., defaults to None
+        :type shared_link: Optional[str], optional
+        :param network_session: An object to keep network session state, defaults to None
+        :type network_session: Optional[NetworkSession], optional
+        """
+        token: Optional[AccessToken] = self.retrieve_token(network_session=network_session)
+        if token == None or token.access_token == None:
+            raise BoxSDKError(message='No access token is available.')
+        auth_manager: AuthorizationManager = AuthorizationManager(network_session=network_session if not network_session == None else NetworkSession())
+        downscoped_token: AccessToken = auth_manager.request_access_token(PostOAuth2TokenGrantTypeField.URN_IETF_PARAMS_OAUTH_GRANT_TYPE_TOKEN_EXCHANGE, subject_token=token.access_token, subject_token_type=PostOAuth2TokenSubjectTokenTypeField.URN_IETF_PARAMS_OAUTH_TOKEN_TYPE_ACCESS_TOKEN, resource=resource, scope=' '.join(scopes), box_shared_link=shared_link)
+        return downscoped_token

box_sdk_gen/box/errors.py

@@ -0,0 +1,78 @@
+import pprint
+from datetime import datetime
+from typing import Any, Dict, Optional
+
+from ..internal.logging import DataSanitizer
+from ..internal.errors import GeneratedCodeError
+
+class BoxSDKError(GeneratedCodeError):
+    def __init__(self, message: str, timestamp: Optional[datetime] = None, error: Optional[Exception] = None, **kwargs):
+        super().__init__(message, **kwargs)
+        self.name = 'BoxSDKError'
+        self.message = message
+        self.timestamp = timestamp if timestamp is not None else datetime.now()
+        self.error = error
+
+    def __str__(self):
+        return ''.join((
+            f'\nTimestamp: {self.timestamp}',
+            f'\nUnderlying error: {self.error}',
+            f'\nMessage: {self.message}',
+        ))
+
+class RequestInfo:
+    def __init__(self, method: str, url: str, query_params: Dict[str, str], headers: Dict[str, str], body: Optional[str] = None):
+        self.method = method
+        self.url = url
+        self.query_params = query_params
+        self.headers = headers
+        self.body = body
+
+    def print(self, data_sanitizer: DataSanitizer):
+        return ''.join((
+            f'\n\tMethod: {self.method}',
+            f'\n\tURL: {self.url}',
+            f'\n\tQuery params: \n{pprint.pformat(self.query_params, indent=8)}',
+            f'\n\tHeaders: \n{pprint.pformat(data_sanitizer.sanitize_headers(self.headers), indent=8)}',
+            ''.join(['\n\tBody: ', '\n' if self.body else '', pprint.pformat(self.body, indent=8)]),
+        ))
+
+class ResponseInfo:
+    def __init__(self, status_code: int, headers: Dict[str, str], body: Dict = None, raw_body: Optional[str] = None, code: Optional[str] = None, context_info: Optional[Dict[str, Any]] = None, request_id: Optional[str] = None, help_url: Optional[str] = None):
+        self.status_code = status_code
+        self.headers = headers
+        self.body = body
+        self.raw_body = raw_body
+        self.code = code
+        self.context_info = context_info
+        self.request_id = request_id
+        self.help_url = help_url
+
+    def print(self, data_sanitizer: DataSanitizer):
+        return ''.join((
+            f'\n\tStatus code: {self.status_code}',
+            f'\n\tHeaders: \n{pprint.pformat(data_sanitizer.sanitize_headers(self.headers), indent=8)}',
+            f'\n\tCode: {self.code}',
+            f'\n\tContext Info: \n{pprint.pformat(self.context_info, indent=8)}',
+            f'\n\tRequest Id: {self.request_id}',
+            f'\n\tHelp Url: {self.help_url}',
+            ''.join(['\n\tBody: ', '\n' if self.body else '', pprint.pformat(data_sanitizer.sanitize_body(self.body), indent=8)]),
+        ))
+
+
+class BoxAPIError(BoxSDKError):
+    def __init__(self, request_info: RequestInfo, response_info: ResponseInfo, message: str, timestamp: Optional[datetime] = None, error: Optional[str] = None, *, data_sanitizer: DataSanitizer = None, **kwargs):
+        super().__init__(message=message, timestamp=timestamp, error=error, **kwargs)
+        if data_sanitizer is None:
+            data_sanitizer = DataSanitizer()
+        self.name = 'BoxAPIError'
+        self.request_info = request_info
+        self.response_info = response_info
+        self.data_sanitizer = data_sanitizer
+
+    def __str__(self):
+        return ''.join([
+            f'\t{super(BoxAPIError, self).__str__()}',
+            f'\nRequest: {self.request_info.print(self.data_sanitizer)}',
+            f'\nResponse: {self.response_info.print(self.data_sanitizer)}',
+        ])