codegen output a7ba6f00842745a4a81cfd756b0a03af

box_sdk_gen/ccg_auth.py

@@ -3,6 +3,7 @@
 from typing import Union, Optional
 
 from .auth import Authentication
+from .token_storage import TokenStorage, InMemoryTokenStorage
 from .auth_schemas import (
  TokenRequestBoxSubjectType,
  TokenRequest,
@@ -20,6 +21,7 @@ def __init__(
  client_secret: str,
  enterprise_id: Union[None, str] = None,
  user_id: Union[None, str] = None,
+ token_storage: TokenStorage = None,
  ):
  """
  :param client_id:
@@ -45,14 +47,20 @@ def __init__(
 
  <https://developer.box.com/en/guides/applications/>
  <https://developer.box.com/en/guides/authentication/select/>
+ :param token_storage:
+ Object responsible for storing token. If no custom implementation provided,
+ the token will be stored in memory.
  """
+ if token_storage is None:
+ token_storage = InMemoryTokenStorage()
  if not enterprise_id and not user_id:
  raise Exception("Enterprise ID or User ID is needed")
 
  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 CCGAuth(Authentication):
@@ -62,7 +70,7 @@ def __init__(self, config: CCGConfig):
  Configuration object of Client Credentials Grant auth.
  """
  self.config = config
- self.token: Union[None, AccessToken] = None
+ self.token_storage = config.token_storage
 
  if config.user_id:
  self.subject_id = self.config.user_id
@@ -79,9 +87,10 @@ def retrieve_token(
  :param network_session: An object to keep network session state
  :return: Access token
  """
- if self.token is None:
- self.refresh_token(network_session=network_session)
- return self.token
+ token = self.token_storage.get()
+ if token is None:
+ return self.refresh_token(network_session=network_session)
+ return token
 
  def refresh_token(
  self, network_session: Optional[NetworkSession] = None
@@ -109,9 +118,9 @@ def refresh_token(
  ),
  )
 
- token_response = AccessToken.from_dict(json.loads(response.text))
- self.token = token_response
- return token_response
+ new_token = AccessToken.from_dict(json.loads(response.text))
+ self.token_storage.store(new_token)
+ return new_token
 
  def as_user(self, user_id: str):
  """
@@ -129,7 +138,7 @@ def as_user(self, user_id: str):
  """
  self.subject_id = user_id
  self.subject_type = TokenRequestBoxSubjectType.USER
- self.token = None
+ self.token_storage.clear()
 
  def as_enterprise(self, enterprise_id: str):
  """
@@ -140,4 +149,4 @@ def as_enterprise(self, enterprise_id: str):
  """
  self.subject_id = enterprise_id
  self.subject_type = TokenRequestBoxSubjectType.ENTERPRISE
- self.token = None
+ self.token_storage.clear()

box_sdk_gen/fetch.py

@@ -42,6 +42,7 @@ class FetchOptions:
  params: Dict[str, str] = None
  headers: Dict[str, str] = None
  body: str = None
+ file_stream: ByteStream = None
  multipart_data: List[MultipartItem] = None
  content_type: str = ""
  response_format: Optional[str] = None
@@ -100,7 +101,7 @@ def fetch(url: str, options: FetchOptions) -> FetchResponse:
  method=options.method,
  url=url,
  headers=headers,
- body=options.body,
+ body=options.file_stream or options.body,
  content_type=options.content_type,
  params=params,
  multipart_data=options.multipart_data,

box_sdk_gen/jwt_auth.py

@@ -14,6 +14,7 @@
  jwt, default_backend, serialization = None, None, None
 
 from .auth import Authentication
+from .token_storage import TokenStorage, InMemoryTokenStorage
 from .auth_schemas import (
  TokenRequestBoxSubjectType,
  TokenRequest,
@@ -35,6 +36,7 @@ def __init__(
  enterprise_id: Optional[str] = None,
  user_id: Optional[str] = None,
  jwt_algorithm: str = 'RS256',
+ token_storage: TokenStorage = None,
  **_kwargs
  ):
  """
@@ -69,7 +71,12 @@ def __init__(
  <https://developer.box.com/en/guides/authentication/select/>
  :param jwt_algorithm:
  Which algorithm to use for signing the JWT assertion. Must be one of 'RS256', 'RS384', 'RS512'.
+ :param token_storage:
+ Object responsible for storing token. If no custom implementation provided,
+ the token will be stored in memory.
  """
+ if token_storage is None:
+ token_storage = InMemoryTokenStorage()
  if not enterprise_id and not user_id:
  raise Exception("Enterprise ID or User ID is needed")
 
@@ -81,17 +88,21 @@ def __init__(
  self.private_key = private_key
  self.private_key_passphrase = private_key_passphrase
  self.jwt_algorithm = jwt_algorithm
+ self.token_storage = token_storage
 
  @classmethod
  def from_config_json_string(
- cls, config_json_string: str, **kwargs: Any
+ cls, config_json_string: str, token_storage: TokenStorage = None, **kwargs: Any
  ) -> 'JWTConfig':
  """
  Create an auth instance as defined by a string content of JSON file downloaded from the Box Developer Console.
  See https://developer.box.com/en/guides/authentication/jwt/ for more information.
 
  :param config_json_string:
  String content of JSON file containing the configuration.
+ :param token_storage:
+ Object responsible for storing token. If no custom implementation provided,
+ the token will be stored in memory.
  :return:
  Auth instance configured as specified by the config dictionary.
  """
@@ -111,22 +122,30 @@ def from_config_json_string(
  private_key_passphrase=config_dict['boxAppSettings']['appAuth'].get(
  'passphrase', None
  ),
+ token_storage=token_storage,
  **kwargs
  )
 
  @classmethod
- def from_config_file(cls, config_file_path: str, **kwargs: Any) -> 'JWTConfig':
+ def from_config_file(
+ cls, config_file_path: str, token_storage: TokenStorage = None, **kwargs: Any
+ ) -> 'JWTConfig':
  """
  Create an auth instance as defined by a JSON file downloaded from the Box Developer Console.
  See https://developer.box.com/en/guides/authentication/jwt/ for more information.
 
  :param config_file_path:
  Path to the JSON file containing the configuration.
+ :param token_storage:
+ Object responsible for storing token. If no custom implementation provided,
+ the token will be stored in memory.
  :return:
  Auth instance configured as specified by the JSON file.
  """
  with open(config_file_path, encoding='utf-8') as config_file:
- return cls.from_config_json_string(config_file.read(), **kwargs)
+ return cls.from_config_json_string(
+ config_file.read(), token_storage, **kwargs
+ )
 
 
 class JWTAuth(Authentication):
@@ -142,7 +161,7 @@ def __init__(self, config: JWTConfig):
  )
 
  self.config = config
- self.token: Union[None, AccessToken] = None
+ self.token_storage = config.token_storage
 
  if config.enterprise_id:
  self.subject_type = TokenRequestBoxSubjectType.ENTERPRISE
@@ -163,9 +182,10 @@ def retrieve_token(
  :param network_session: An object to keep network session state
  :return: Access token
  """
- if self.token is None:
- self.refresh_token(network_session=network_session)
- return self.token
+ token = self.token_storage.get()
+ if token is None:
+ return self.refresh_token(network_session=network_session)
+ return token
 
  def refresh_token(
  self, network_session: Optional[NetworkSession] = None
@@ -218,9 +238,9 @@ def refresh_token(
  ),
  )
 
- token_response = AccessToken.from_dict(json.loads(response.text))
- self.token = token_response
- return self.token
+ new_token = AccessToken.from_dict(json.loads(response.text))
+ self.token_storage.store(new_token)
+ return new_token
 
  def as_user(self, user_id: str):
  """
@@ -238,7 +258,7 @@ def as_user(self, user_id: str):
  """
  self.subject_id = user_id
  self.subject_type = TokenRequestBoxSubjectType.USER
- self.token = None
+ self.token_storage.clear()
 
  def as_enterprise(self, enterprise_id: str):
  """
@@ -249,7 +269,7 @@ def as_enterprise(self, enterprise_id: str):
  """
  self.subject_id = enterprise_id
  self.subject_type = TokenRequestBoxSubjectType.ENTERPRISE
- self.token = None
+ self.token_storage.clear()
 
  @classmethod
  def _get_rsa_private_key(