(self, buff, remote_path)
| 426 | threading.Thread(target=target).start() |
| 427 | |
| 428 | def upload_from(self, buff, remote_path): |
| 429 | |
| 430 | try: |
| 431 | urn = Urn(remote_path) |
| 432 | |
| 433 | if urn.is_dir(): |
| 434 | raise OptionNotValid(name="remote_path", value=remote_path) |
| 435 | |
| 436 | if not self.check(urn.parent()): |
| 437 | raise RemoteParentNotFound(urn.path()) |
| 438 | |
| 439 | url = {'hostname': self.webdav.hostname, 'root': self.webdav.root, 'path': urn.quote()} |
| 440 | options = { |
| 441 | 'URL': "{hostname}{root}{path}".format(**url), |
| 442 | 'HTTPHEADER': self.get_header('upload_from'), |
| 443 | 'UPLOAD': 1, |
| 444 | 'READFUNCTION': buff.read, |
| 445 | } |
| 446 | |
| 447 | request = self.Request(options=options) |
| 448 | |
| 449 | request.perform() |
| 450 | code = request.getinfo(pycurl.HTTP_CODE) |
| 451 | if code == "507": |
| 452 | raise NotEnoughSpace() |
| 453 | |
| 454 | request.close() |
| 455 | |
| 456 | except pycurl.error: |
| 457 | raise NotConnection(self.webdav.hostname) |
| 458 | |
| 459 | def upload(self, remote_path, local_path, progress=None): |
| 460 |
no test coverage detected