(self, remote_path, local_path, progress=None)
| 487 | self.upload(local_path=_local_path, remote_path=_remote_path, progress=progress) |
| 488 | |
| 489 | def upload_file(self, remote_path, local_path, progress=None): |
| 490 | |
| 491 | try: |
| 492 | if not os.path.exists(local_path): |
| 493 | raise LocalResourceNotFound(local_path) |
| 494 | |
| 495 | urn = Urn(remote_path) |
| 496 | |
| 497 | if urn.is_dir(): |
| 498 | raise OptionNotValid(name="remote_path", value=remote_path) |
| 499 | |
| 500 | if os.path.isdir(local_path): |
| 501 | raise OptionNotValid(name="local_path", value=local_path) |
| 502 | |
| 503 | if not self.check(urn.parent()): |
| 504 | raise RemoteParentNotFound(urn.path()) |
| 505 | |
| 506 | with open(local_path, "rb") as local_file: |
| 507 | |
| 508 | url = {'hostname': self.webdav.hostname, 'root': self.webdav.root, 'path': urn.quote()} |
| 509 | options = { |
| 510 | 'URL': "{hostname}{root}{path}".format(**url), |
| 511 | 'HTTPHEADER': self.get_header('upload_file'), |
| 512 | 'UPLOAD': 1, |
| 513 | 'READFUNCTION': local_file.read, |
| 514 | 'NOPROGRESS': 0 if progress else 1 |
| 515 | } |
| 516 | |
| 517 | if progress: |
| 518 | options["PROGRESSFUNCTION"] = progress |
| 519 | |
| 520 | file_size = os.path.getsize(local_path) |
| 521 | if file_size > self.large_size: |
| 522 | options['INFILESIZE_LARGE'] = file_size |
| 523 | else: |
| 524 | options['INFILESIZE'] = file_size |
| 525 | |
| 526 | request = self.Request(options=options) |
| 527 | |
| 528 | request.perform() |
| 529 | code = request.getinfo(pycurl.HTTP_CODE) |
| 530 | if code == "507": |
| 531 | raise NotEnoughSpace() |
| 532 | |
| 533 | request.close() |
| 534 | |
| 535 | except pycurl.error: |
| 536 | raise NotConnection(self.webdav.hostname) |
| 537 | |
| 538 | def upload_sync(self, remote_path, local_path, callback=None): |
| 539 |
no test coverage detected