Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
overwrite (bool): should this overwrite existing files, default `False`
Returns:
list of ids of the uploaded content
"""
warnings.warn("sendfiles2box to be removed in v0.16", stacklevel=2)
if isinstance(filenames, string_types):
filenames = [filenames]
if isinstance(remotefilenames, string_types):
remotefilenames = [remotefilenames]
if remotefilenames is None:
remotefilenames = [os.path.basename(f) for f in filenames]
iemprops = get_properties()
if "boxclient.access_token" not in iemprops:
return []
oauth = OAuth2(
client_id=iemprops["boxclient.client_id"],
client_secret=iemprops["boxclient.client_secret"],
access_token=iemprops["boxclient.access_token"],
refresh_token=iemprops["boxclient.refresh_token"],
store_tokens=_store_tokens,
)
client = Client(oauth)
folder_id = 0
for token in remotepath.split("/"):
if token.strip() == "":
continue
offset = 0
found = False
while not found:
LOG.debug("folder(%s).get_items(offset=%s)", folder_id, offset)
items = list(
def __init__(self):
oauth = OAuth2(
client_id=BOX_CLIENT_ID,
client_secret=BOX_CLIENT_SECRET,
access_token=BOX_ACCESS_TOKEN
)
self._client = Client(oauth)
def credentials_are_valid(self, user_settings, client):
from addons.box.models import Provider as Box # Avoid circular import
if self.node_settings.has_auth:
if Box(self.node_settings.external_account).refresh_oauth_key():
return True
if user_settings:
oauth = OAuth2(client_id=settings.BOX_KEY, client_secret=settings.BOX_SECRET, access_token=user_settings.external_accounts[0].oauth_key)
client = client or Client(oauth)
try:
client.user()
except (BoxAPIException, IndexError):
return False
return True
def connect(self, token: str) -> 'Client':
"""Connect to box using a bearer token
Args:
token: The bearer token to use for this connection
"""
if self.config['storage'] == 'box':
auth = OAuth2(None, None, access_token=token)
return Client(auth)
else:
raise Exception('Invalid Storage Option: ' +
self.config['storage'])
with self.exception_handler():
# get all items
files = []
offset = 0
while len(files) == offset:
files += self.app_folder.get_items(self.MAX_BOX_LIMIT, offset=offset)
offset += self.MAX_BOX_LIMIT
self.id_cache = {user_file.name: user_file.object_id for user_file in files}
# if this came from cache, it is a json string that needs to be converted
if type(user_credentials) in [unicode, str]:
user_credentials = json.loads(user_credentials)
self.access_token, self.refresh_token = user_credentials["access_token"], user_credentials["refresh_token"]
oauth = OAuth2(client_id=self.app_credentials["client_id"],
client_secret=self.app_credentials["client_secret"],
store_tokens=store_tokens_callback,
access_token=self.access_token,
refresh_token=self.refresh_token)
self.client = Client(oauth)
load_email()
make_app_folder()
prime_cache()
def start_connection(self):
self.oauth = OAuth2(client_id=self.app_credentials["client_id"], client_secret=self.app_credentials["client_secret"])
with self.exception_handler():
authorize_url, self.csrf_token = self.oauth.get_authorization_url(self.get_oauth_redirect_url())
return authorize_url
def _folder_data(self, folder_id):
# Split out from set_folder for ease of testing, due to
# outgoing requests. Should only be called by set_folder
try:
Provider(self.external_account).refresh_oauth_key(force=True)
except InvalidGrantError:
raise exceptions.InvalidAuthError()
try:
oauth = OAuth2(client_id=settings.BOX_KEY, client_secret=settings.BOX_SECRET, access_token=self.external_account.oauth_key)
client = Client(oauth)
folder_data = client.folder(self.folder_id).get()
except BoxAPIException:
raise exceptions.InvalidFolderError()
folder_name = folder_data['name'].replace('All Files', '') or '/ (Full Box)'
folder_path = '/'.join(
[x['name'] for x in folder_data['path_collection']['entries'] if x['name']] +
[folder_data['name']]
).replace('All Files', '') or '/'
return folder_name, folder_path