(self, item, href)
| 118 | return href, etag |
| 119 | |
| 120 | def _upload_impl(self, item, href): |
| 121 | fpath = self._get_filepath(href) |
| 122 | try: |
| 123 | with atomic_write(fpath, mode='wb', overwrite=False) as f: |
| 124 | f.write(item.raw.encode(self.encoding)) |
| 125 | return fpath, get_etag_from_file(f) |
| 126 | except OSError as e: |
| 127 | if e.errno == errno.EEXIST: |
| 128 | raise exceptions.AlreadyExistingError(existing_href=href) |
| 129 | else: |
| 130 | raise |
| 131 | |
| 132 | def update(self, href, item, etag): |
| 133 | fpath = self._get_filepath(href) |
no test coverage detected