abfs.py 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744
  1. #!/usr/bin/env python
  2. # Licensed to Cloudera, Inc. under one
  3. # or more contributor license agreements. See the NOTICE file
  4. # distributed with this work for additional information
  5. # regarding copyright ownership. Cloudera, Inc. licenses this file
  6. # to you under the Apache License, Version 2.0 (the
  7. # "License"); you may not use this file except in compliance
  8. # with the License. You may obtain a copy of the License at
  9. #
  10. # http://www.apache.org/licenses/LICENSE-2.0
  11. #
  12. # Unless required by applicable law or agreed to in writing, software
  13. # distributed under the License is distributed on an "AS IS" BASIS,
  14. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. # See the License for the specific language governing permissions and
  16. # limitations under the License.
  17. """
  18. Interfaces for ABFS
  19. """
  20. import os
  21. import logging
  22. import threading
  23. import urllib.error
  24. import urllib.request
  25. from builtins import object
  26. from math import ceil
  27. from posixpath import join
  28. from urllib.parse import quote as urllib_quote, urlparse as lib_urlparse
  29. from django.http.multipartparser import MultiPartParser
  30. import azure.abfs.__init__ as Init_ABFS
  31. from azure.abfs.abfsfile import ABFSFile
  32. from azure.abfs.abfsstats import ABFSStat
  33. from azure.conf import PERMISSION_ACTION_ABFS, is_raz_abfs
  34. from desktop.conf import RAZ
  35. from desktop.lib.rest import http_client, resource
  36. from desktop.lib.rest.raz_http_client import RazHttpClient
  37. from hadoop.fs.exceptions import WebHdfsException
  38. from hadoop.hdfs_site import get_umask_mode
  39. LOG = logging.getLogger()
  40. # Azure has a 30MB block limit on upload.
  41. UPLOAD_CHUCK_SIZE = 30 * 1000 * 1000
  42. class ABFSFileSystemException(IOError):
  43. def __init__(self, *args, **kwargs):
  44. super(ABFSFileSystemException, self).__init__(*args, **kwargs)
  45. class ABFS(object):
  46. def __init__(
  47. self,
  48. url,
  49. fs_defaultfs,
  50. logical_name=None,
  51. hdfs_superuser=None,
  52. security_enabled=False,
  53. ssl_cert_ca_verify=True,
  54. temp_dir="/tmp",
  55. umask=0o1022,
  56. hdfs_supergroup=None,
  57. access_token=None,
  58. token_type=None,
  59. expiration=None,
  60. username=None,
  61. ):
  62. self._url = url
  63. self._superuser = hdfs_superuser
  64. self._security_enabled = security_enabled
  65. self._ssl_cert_ca_verify = ssl_cert_ca_verify
  66. self._temp_dir = temp_dir
  67. self._umask = umask
  68. self.is_sentry_managed = lambda path: False
  69. self._fs_defaultfs = fs_defaultfs
  70. self._logical_name = logical_name
  71. self._supergroup = hdfs_supergroup
  72. self._access_token = access_token
  73. self._token_type = token_type
  74. split = lib_urlparse(fs_defaultfs)
  75. self._scheme = split.scheme
  76. self._netloc = split.netloc
  77. self._is_remote = True
  78. self._has_trash_support = False
  79. self._filebrowser_action = PERMISSION_ACTION_ABFS
  80. self.expiration = expiration
  81. self._user = username
  82. # To store user info
  83. self._thread_local = threading.local() # Unused
  84. self._root = self.get_client(url)
  85. LOG.debug("Initializing ABFS : %s (security: %s, superuser: %s)" % (self._url, self._security_enabled, self._superuser))
  86. @classmethod
  87. def from_config(cls, hdfs_config, auth_provider):
  88. credentials = auth_provider.get_credentials()
  89. return cls(
  90. url=hdfs_config.WEBHDFS_URL.get(),
  91. fs_defaultfs=hdfs_config.FS_DEFAULTFS.get(),
  92. logical_name=None,
  93. security_enabled=False,
  94. ssl_cert_ca_verify=False,
  95. temp_dir=None,
  96. umask=get_umask_mode(),
  97. hdfs_supergroup=None,
  98. access_token=credentials.get('access_token'),
  99. token_type=credentials.get('token_type'),
  100. expiration=int(credentials.get('expires_on')) * 1000 if credentials.get('expires_on') is not None else None,
  101. username=credentials.get('username'),
  102. )
  103. def get_client(self, url):
  104. if RAZ.IS_ENABLED.get():
  105. client = RazHttpClient(self._user, url, exc_class=WebHdfsException, logger=LOG)
  106. else:
  107. client = http_client.HttpClient(url, exc_class=WebHdfsException, logger=LOG)
  108. return resource.Resource(client)
  109. def _getheaders(self):
  110. headers = {
  111. "x-ms-version": "2019-12-12" # For latest SAS support
  112. }
  113. if self._token_type and self._access_token:
  114. headers["Authorization"] = self._token_type + " " + self._access_token
  115. return headers
  116. @property
  117. def superuser(self):
  118. return self._superuser
  119. @property
  120. def supergroup(self):
  121. return self._supergroup
  122. # Parse info about filesystems, directories, and files
  123. # --------------------------------
  124. def isdir(self, path):
  125. """Check if the given path is a directory or not."""
  126. try:
  127. stats = self.stats(path)
  128. return stats.isDir
  129. except Exception as e:
  130. # If checking stats for path here gives 404 error, it means the path does not exist and therefore is not a directory.
  131. if e.code == 404:
  132. return False
  133. raise e
  134. def isfile(self, path):
  135. """Check if the given path is a file or not."""
  136. try:
  137. stats = self.stats(path)
  138. return not stats.isDir
  139. except Exception as e:
  140. # If checking stats for path here gives 404 error, it means the path does not exist and therefore is not a file.
  141. if e.code == 404:
  142. return False
  143. raise e
  144. def exists(self, path):
  145. """
  146. Test if a path exists
  147. """
  148. try:
  149. if ABFS.isroot(path):
  150. return True
  151. self.stats(path)
  152. except WebHdfsException as e:
  153. if e.code == 404:
  154. return False
  155. raise WebHdfsException
  156. except IOError:
  157. return False
  158. return True
  159. def stats(self, path, params=None, **kwargs):
  160. """
  161. List the stat of the actual file/directory
  162. Returns the ABFFStat object
  163. """
  164. if ABFS.isroot(path):
  165. return ABFSStat.for_root(path)
  166. try:
  167. file_system, dir_name = Init_ABFS.parse_uri(path)[:2]
  168. except Exception:
  169. raise IOError
  170. if dir_name == '':
  171. return ABFSStat.for_filesystem(self._statsf(file_system, params, **kwargs), path)
  172. return ABFSStat.for_single(self._stats(file_system + '/' + dir_name, params, **kwargs), path)
  173. def listdir_stats(self, path, params=None, **kwargs):
  174. """
  175. List the stats for the directories inside the specified path
  176. Returns the Multiple ABFFStat object #note change later for recursive cases
  177. """
  178. if ABFS.isroot(path):
  179. return self.listfilesystems_stats(params=None, **kwargs)
  180. dir_stats = []
  181. file_system, directory_name, account = Init_ABFS.parse_uri(path)
  182. root = Init_ABFS.ABFS_ROOT
  183. if path.lower().startswith(Init_ABFS.ABFS_ROOT_S):
  184. root = Init_ABFS.ABFS_ROOT_S
  185. if params is None:
  186. params = {}
  187. if 'recursive' not in params:
  188. params['recursive'] = 'false'
  189. params['resource'] = 'filesystem'
  190. if directory_name != "":
  191. params['directory'] = directory_name
  192. while True:
  193. res = self._root._invoke("GET", file_system, params, headers=self._getheaders(), **kwargs)
  194. resp = self._root._format_response(res)
  195. if account:
  196. file_system += account
  197. for x in resp['paths']:
  198. dir_stats.append(ABFSStat.for_directory(res.headers, x, root + file_system + "/" + x['name']))
  199. # If the number of paths returned exceeds the 5000, a continuation token is provided in the response header x-ms-continuation,
  200. # which must be used in subsequent invocations to continue listing the paths.
  201. if 'x-ms-continuation' in res.headers:
  202. params['continuation'] = res.headers['x-ms-continuation']
  203. else:
  204. break
  205. return dir_stats
  206. def listfilesystems_stats(self, root=Init_ABFS.ABFS_ROOT, params=None, **kwargs):
  207. """
  208. Lists the stats inside the File Systems, No functionality for params
  209. """
  210. stats = []
  211. if params is None:
  212. params = {}
  213. params["resource"] = "account"
  214. res = self._root._invoke("GET", params=params, headers=self._getheaders())
  215. resp = self._root._format_response(res)
  216. for x in resp['filesystems']:
  217. stats.append(ABFSStat.for_filesystems(res.headers, x, root))
  218. return stats
  219. def _stats(self, schemeless_path, params=None, **kwargs):
  220. """
  221. Container function for both stats,
  222. Returns the header of the result
  223. """
  224. if params is None:
  225. params = {}
  226. params['action'] = 'getStatus'
  227. res = self._root._invoke('HEAD', schemeless_path, params, headers=self._getheaders(), **kwargs)
  228. return res.headers
  229. def _statsf(self, schemeless_path, params=None, **kwargs):
  230. """
  231. Continer function for both stats but if it's a file system
  232. Returns the header of the result
  233. """
  234. if params is None:
  235. params = {}
  236. # For RAZ ABFS, the root path stats should have 'getAccessControl' param.
  237. if is_raz_abfs():
  238. params['action'] = 'getAccessControl'
  239. else:
  240. params['resource'] = 'filesystem'
  241. res = self._root._invoke('HEAD', schemeless_path, params, headers=self._getheaders(), **kwargs)
  242. return res.headers
  243. def listdir(self, path, params=None, glob=None, **kwargs):
  244. """
  245. Lists the names inside the current directories
  246. """
  247. if ABFS.isroot(path):
  248. return self.listfilesystems(params=params, **kwargs)
  249. listofDir = self.listdir_stats(path, params)
  250. return [x.name for x in listofDir]
  251. def listfilesystems(self, root=Init_ABFS.ABFS_ROOT, params=None, **kwargs):
  252. """
  253. Lists the names of the File Systems, limited arguements
  254. """
  255. listofFileSystems = self.listfilesystems_stats(root=root, params=params)
  256. return [x.name for x in listofFileSystems]
  257. @staticmethod
  258. def get_home_dir():
  259. """
  260. Attempts to go to the directory set by the user in the configuration file. If not defaults to abfs://
  261. """
  262. return Init_ABFS.get_abfs_home_directory()
  263. # Find or alter information about the URI path
  264. # --------------------------------
  265. @staticmethod
  266. def isroot(path):
  267. """
  268. Checks if the path is the root path
  269. """
  270. return Init_ABFS.is_root(path)
  271. @staticmethod
  272. def normpath(path):
  273. """
  274. Normalizes a path
  275. """
  276. resp = Init_ABFS.normpath(path)
  277. return resp
  278. @staticmethod
  279. def netnormpath(path):
  280. """
  281. Normalizes a path
  282. """
  283. return Init_ABFS.normpath(path)
  284. @staticmethod
  285. def parent_path(path):
  286. """
  287. Returns the Parent Path
  288. """
  289. return Init_ABFS.parent_path(path)
  290. @staticmethod
  291. def join(first, *comp_list):
  292. """
  293. Joins two paths together
  294. """
  295. return Init_ABFS.join(first, *comp_list)
  296. # Create Files,directories, or File Systems
  297. # --------------------------------
  298. def mkdir(self, path, params=None, headers=None, *args, **kwargs):
  299. """
  300. Makes a directory
  301. """
  302. if params is None:
  303. params = {}
  304. params['resource'] = 'directory'
  305. self._create_path(path, params=params, headers=params, overwrite=False)
  306. def create(self, path, overwrite=False, data=None, headers=None, *args, **kwargs):
  307. """
  308. Makes a File (Put text in data if adding data)
  309. """
  310. params = {'resource': 'file'}
  311. self._create_path(path, params=params, headers=headers, overwrite=overwrite)
  312. if data:
  313. self._writedata(path, data, len(data))
  314. def create_home_dir(self, home_path):
  315. # When ABFS raz is enabled, try to create user home directory
  316. if is_raz_abfs():
  317. LOG.debug('Attempting to create user directory for path: %s' % home_path)
  318. try:
  319. if not self.exists(home_path):
  320. self.mkdir(home_path)
  321. else:
  322. LOG.debug('Skipping user directory creation, the path already exists: %s' % home_path)
  323. except Exception as e:
  324. LOG.exception('Failed to create user home directory for path %s with error: %s' % (home_path, str(e)))
  325. else:
  326. LOG.info('Create home directory is not available for Azure filesystem')
  327. def _create_path(self, path, params=None, headers=None, overwrite=False):
  328. """
  329. Container method for Create
  330. """
  331. file_system, dir_name = Init_ABFS.parse_uri(path)[:2]
  332. if dir_name == '':
  333. return self._create_fs(file_system)
  334. no_scheme = file_system + '/' + dir_name
  335. additional_header = self._getheaders()
  336. if headers is not None:
  337. additional_header.update(headers)
  338. if not overwrite:
  339. additional_header['If-None-Match'] = '*'
  340. self._root.put(no_scheme, params, headers=additional_header)
  341. def _create_fs(self, file_system):
  342. """
  343. Creates a File System
  344. """
  345. self._root.put(file_system, {'resource': 'filesystem'}, headers=self._getheaders())
  346. # Read Files
  347. # --------------------------------
  348. def read(self, path, offset='0', length=0, *args, **kwargs):
  349. """
  350. Read data from a file
  351. """
  352. path = Init_ABFS.strip_scheme(path)
  353. headers = self._getheaders()
  354. if length != 0 and length != '0':
  355. headers['range'] = 'bytes=%s-%s' % (str(offset), str(int(offset) + int(length) - 1))
  356. return self._root.get(path, headers=headers)
  357. def open(self, path, option='r', *args, **kwargs):
  358. """
  359. Returns an ABFSFile object that pretends that a file is open
  360. """
  361. return ABFSFile(self, path, option)
  362. # Alter Files
  363. # --------------------------------
  364. def append(self, path, data, offset=0):
  365. if not data:
  366. LOG.warning("There is no data to append to")
  367. return
  368. self._append(path, data)
  369. return self.flush(path, {'position': int(len(data)) + int(offset)})
  370. def _append(self, path, data, size=0, offset=0, params=None, **kwargs):
  371. """
  372. Appends the data to a file
  373. """
  374. path = Init_ABFS.strip_scheme(path)
  375. if params is None:
  376. LOG.warning("Params not specified, Append will take longer")
  377. resp = self._stats(path)
  378. params = {'position': int(resp['Content-Length']) + offset, 'action': 'append'}
  379. else:
  380. params['action'] = 'append'
  381. headers = {}
  382. if size == 0 or size == '0':
  383. headers['Content-Length'] = str(len(data.getvalue()))
  384. if headers['Content-Length'] == '0':
  385. return
  386. else:
  387. headers['Content-Length'] = str(size)
  388. return self._patching_sl(path, params, data, headers, **kwargs)
  389. def flush(self, path, params=None, headers=None, **kwargs):
  390. """
  391. Flushes the data(i.e. writes appended data to File)
  392. """
  393. path = Init_ABFS.strip_scheme(path)
  394. if params is None:
  395. LOG.warning("Params not specified")
  396. params = {'position': 0}
  397. if 'position' not in params:
  398. LOG.warning("Position is not specified")
  399. params['position'] = 0
  400. params['action'] = 'flush'
  401. if headers is None:
  402. headers = {}
  403. headers['Content-Length'] = '0'
  404. self._patching_sl(path, params, header=headers, **kwargs)
  405. # Remove Filesystems, directories. or Files
  406. # --------------------------------
  407. def remove(self, path, skip_trash=True):
  408. """
  409. Removes an item indicated in the path
  410. Also removes empty directories
  411. """
  412. self._delete(path, recursive='false', skip_trash=skip_trash)
  413. def rmtree(self, path, skip_trash=True):
  414. """
  415. Remove everything in a given directory
  416. """
  417. self._delete(path, recursive='true', skip_trash=skip_trash)
  418. def _delete(self, path, recursive='false', skip_trash=True):
  419. """
  420. Wrapper function for calling delete, no support for trash or
  421. """
  422. if not skip_trash:
  423. raise NotImplementedError("Trash not implemented for ABFS")
  424. if ABFS.isroot(path):
  425. raise RuntimeError("Cannot Remove Root")
  426. file_system, dir_name = Init_ABFS.parse_uri(path)[:2]
  427. if dir_name == '':
  428. return self._root.delete(file_system, {'resource': 'filesystem'}, headers=self._getheaders())
  429. new_path = file_system + '/' + dir_name
  430. param = None
  431. if self.isdir(path):
  432. param = {'recursive': recursive}
  433. self._root.delete(new_path, param, headers=self._getheaders())
  434. def restore(self, path):
  435. raise NotImplementedError("")
  436. # Edit permissions of Filesystems, directories. or Files
  437. # --------------------------------
  438. def chown(self, path, user=None, group=None, *args, **kwargs):
  439. """
  440. Changes ownership (not implemented)
  441. """
  442. headers = {}
  443. if user is not None:
  444. headers['x-ms-owner'] = user
  445. if group is not None:
  446. headers['x-ms-group'] = group
  447. self.setAccessControl(path, headers=headers, **kwargs)
  448. def chmod(self, path, permissionNumber=None, *args, **kwargs):
  449. """
  450. Set File Permissions (passing as an int converts said integer to octal. Passing as a string assumes the string is in octal)
  451. """
  452. header = {}
  453. if permissionNumber is not None:
  454. if isinstance(permissionNumber, basestring):
  455. header['x-ms-permissions'] = str(permissionNumber)
  456. else:
  457. header['x-ms-permissions'] = oct(permissionNumber)
  458. self.setAccessControl(path, headers=header)
  459. def setAccessControl(self, path, headers, **kwargs):
  460. """
  461. Set Access Controls (Can do both chmod and chown) (not implemented)
  462. """
  463. path = Init_ABFS.strip_scheme(path)
  464. params = {'action': 'setAccessControl'}
  465. if headers is None:
  466. headers = {}
  467. self._patching_sl(path, params, header=headers, **kwargs)
  468. def mktemp(self, subdir='', prefix='tmp', basedir=None):
  469. raise NotImplementedError("")
  470. def purge_trash(self):
  471. raise NotImplementedError("")
  472. # Handle file systems interactions
  473. # --------------------------------
  474. def copy(self, src, dst, *args, **kwargs):
  475. """
  476. General Copying
  477. """
  478. if self.isfile(src):
  479. return self.copyfile(src, dst)
  480. self.copy_remote_dir(src, dst)
  481. def copyfile(self, src, dst, *args, **kwargs):
  482. """
  483. Copies a File to another location
  484. """
  485. new_path = dst + '/' + Init_ABFS.strip_path(src)
  486. self.create(new_path)
  487. chunk_size = self.get_upload_chuck_size()
  488. file = self.read(src)
  489. size = len(file)
  490. self._writedata(new_path, file, size)
  491. def copy_remote_dir(self, src, dst, *args, **kwargs):
  492. """
  493. Copies the entire contents of a directory to another location
  494. """
  495. dst = dst + '/' + Init_ABFS.strip_path(src)
  496. self.mkdir(dst)
  497. other_files = self.listdir(src)
  498. for x in other_files:
  499. x = src + '/' + Init_ABFS.strip_path(x)
  500. self.copy(x, dst)
  501. def rename(self, old, new):
  502. """
  503. Renames a file
  504. """
  505. rename_source = Init_ABFS.strip_scheme(old)
  506. headers = {'x-ms-rename-source': '/' + urllib_quote(rename_source)}
  507. try:
  508. self._create_path(new, headers=headers, overwrite=True)
  509. except WebHdfsException as e:
  510. if e.code == 409:
  511. self.copy(old, new)
  512. self.rmtree(old)
  513. else:
  514. raise e
  515. def rename_star(self, old_dir, new_dir):
  516. """
  517. Renames a directory
  518. """
  519. self.rename(old_dir, new_dir)
  520. # Deprecated
  521. def upload(self, file, path, *args, **kwargs):
  522. """
  523. Upload is done by the client
  524. """
  525. pass
  526. def upload_v1(self, META, input_data, destination, username):
  527. from azure.abfs.upload import ABFSNewFileUploadHandler # Circular dependency
  528. abfs_upload_handler = ABFSNewFileUploadHandler(destination, username)
  529. parser = MultiPartParser(META, input_data, [abfs_upload_handler])
  530. return parser.parse()
  531. def copyFromLocal(self, local_src, remote_dst, *args, **kwargs):
  532. """
  533. Copy a directory or file from Local (Testing)
  534. """
  535. local_src = local_src.endswith('/') and local_src[:-1] or local_src
  536. remote_dst = remote_dst.endswith('/') and remote_dst[:-1] or remote_dst
  537. if os.path.isdir(local_src):
  538. self._local_copy_dir(local_src, remote_dst)
  539. else:
  540. (basename, filename) = os.path.split(local_src)
  541. self._local_copy_file(local_src, self.isdir(remote_dst) and self.join(remote_dst, filename) or remote_dst)
  542. def _local_copy_dir(self, local_src, remote_dst):
  543. """
  544. A wraper function for copying local directories
  545. """
  546. self.mkdir(remote_dir)
  547. for f in os.listdir(local_dir):
  548. local_src = os.path.join(local_dir, f)
  549. remote_dst = self.join(remote_dir, f)
  550. if os.path.isdir(local_src):
  551. self._copy_dir(local_src, remote_dst, mode)
  552. else:
  553. self._copy_file(local_src, remote_dst)
  554. def _local_copy_file(self, local_src, remote_dst, chunk_size=UPLOAD_CHUCK_SIZE):
  555. """
  556. A wraper function for copying local Files
  557. """
  558. if os.path.isfile(local_src):
  559. if self.exists(remote_dst):
  560. LOG.info('%s already exists. Skipping.' % remote_dst)
  561. return
  562. src = file(local_src)
  563. try:
  564. try:
  565. self.create(remote_dst)
  566. chunk = src.read(chunk_size)
  567. offset = 0
  568. while chunk:
  569. size = len(chunk)
  570. self._append(remote_dst, chunk, size=size, params={'position': offset})
  571. offset += size
  572. chunk = src.read(chunk_size)
  573. self.flush(remote_dst, params={'position': offset})
  574. except Exception:
  575. LOG.exception(_('Copying %s -> %s failed.') % (local_src, remote_dst))
  576. raise
  577. finally:
  578. src.close()
  579. else:
  580. LOG.info(_('Skipping %s (not a file).') % local_src)
  581. def check_access(self, path, *args, **kwargs):
  582. """
  583. Check access of a file/directory (Work in Progress/Not Ready)
  584. """
  585. raise NotImplementedError("")
  586. def mkswap(self, filename, subdir='', suffix='swp', basedir=None):
  587. """
  588. Makes a directory and returns a potential filename for that directory
  589. """
  590. base = self.join(basedir or self._temp_dir, subdir)
  591. if not self.isdir(base):
  592. self.mkdir(base)
  593. candidate = self.join(base, "%s.%s" % (filename, suffix))
  594. return candidate
  595. def setuser(self, user):
  596. """
  597. Changes the User
  598. """
  599. self._user = user
  600. def get_upload_chuck_size(self):
  601. """
  602. Gets the maximum size allowed to upload
  603. """
  604. return UPLOAD_CHUCK_SIZE
  605. def filebrowser_action(self):
  606. return self._filebrowser_action
  607. # Other Methods to condense stuff
  608. # ----------------------------
  609. # Write Files on creation
  610. # ----------------------------
  611. def _writedata(self, path, data, size):
  612. """
  613. Adds text to a given file
  614. """
  615. chunk_size = self.get_upload_chuck_size()
  616. cycles = ceil(float(size) / chunk_size)
  617. for i in range(0, cycles):
  618. chunk = size % chunk_size
  619. if i != cycles or chunk == 0:
  620. length = chunk_size
  621. else:
  622. length = chunk
  623. self._append(path, data[i * chunk_size : i * chunk_size + length], length)
  624. self.flush(path, {'position': int(size)})
  625. # Use Patch HTTP request
  626. # ----------------------------
  627. def _patching_sl(self, schemeless_path, param, data=None, header=None, **kwargs):
  628. """
  629. A wraper function for patch
  630. """
  631. if header is None:
  632. header = {}
  633. header.update(self._getheaders())
  634. return self._root.invoke('PATCH', schemeless_path, param, data, headers=header, **kwargs)