How to use the b2sdk.transfer.emerge.write_intent.WriteIntent function in b2sdk

To help you get started, we’ve selected a few b2sdk examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github Backblaze / b2-sdk-python / test / internal / test_emerge_planner.py View on Github external
def test_single_multipart_upload(self):
        self.assertGreater(self.recommended_size, 2 * self.min_size)

        remainder = 2 * self.min_size
        source = UploadSource(self.recommended_size * 5 + remainder)
        expected_part_sizes = [self.recommended_size] * 5 + [remainder]

        self.verify_emerge_plan_for_write_intents(
            [WriteIntent(source)],
            self.split_source_to_part_defs(source, expected_part_sizes),
        )
github Backblaze / b2-sdk-python / test / internal / test_emerge_planner.py View on Github external
def test_single_part_upload(self):
        source = UploadSource(self.recommended_size)

        self.verify_emerge_plan_for_write_intents(
            [WriteIntent(source)],
            [part(source)],
        )
github Backblaze / b2-sdk-python / test / internal / test_emerge_planner.py View on Github external
def test_local_small_copy_overlap(self):
        self.assertGreater(self.recommended_size, self.min_size * 3 - 3)
        source_upload = UploadSource(self.recommended_size)
        small_size = self.min_size - 1
        source_copy1 = CopySource(small_size)
        source_copy2 = CopySource(small_size)
        source_copy3 = CopySource(small_size)
        write_intents = [
            WriteIntent(source_upload),
            WriteIntent(source_copy1),
            WriteIntent(source_copy2, destination_offset=small_size),
            WriteIntent(source_copy3, destination_offset=2 * small_size),
        ]

        self.verify_emerge_plan_for_write_intents(
            write_intents,
            [part(source_upload)],
        )
github Backblaze / b2-sdk-python / test / internal / test_emerge_planner.py View on Github external
def test_local_remote_overlap_end(self):
        source_upload = UploadSource(self.recommended_size * 2)
        source_copy = CopySource(self.recommended_size)
        write_intents = [
            WriteIntent(source_upload),
            WriteIntent(source_copy, destination_offset=self.recommended_size),
        ]

        self.verify_emerge_plan_for_write_intents(
            write_intents,
            [
                part(source_upload, 0, self.recommended_size),
                part(source_copy),
            ],
github Backblaze / b2-sdk-python / test / internal / test_emerge_planner.py View on Github external
def test_basic_local_overlap(self):
        source1 = UploadSource(self.recommended_size * 2)
        source2 = UploadSource(self.recommended_size * 2)
        write_intents = [
            WriteIntent(source1),
            WriteIntent(source2, destination_offset=self.recommended_size),
        ]

        self.verify_emerge_plan_for_write_intents(
            write_intents,
            [part(source1, 0, self.recommended_size)] +
            self.split_source_to_part_defs(source2, [self.recommended_size] * 2),
        )
github Backblaze / b2-sdk-python / test / internal / test_emerge_planner.py View on Github external
def test_local_small_copy_overlap(self):
        self.assertGreater(self.recommended_size, self.min_size * 3 - 3)
        source_upload = UploadSource(self.recommended_size)
        small_size = self.min_size - 1
        source_copy1 = CopySource(small_size)
        source_copy2 = CopySource(small_size)
        source_copy3 = CopySource(small_size)
        write_intents = [
            WriteIntent(source_upload),
            WriteIntent(source_copy1),
            WriteIntent(source_copy2, destination_offset=small_size),
            WriteIntent(source_copy3, destination_offset=2 * small_size),
        ]

        self.verify_emerge_plan_for_write_intents(
            write_intents,
            [part(source_upload)],
        )
github Backblaze / b2-sdk-python / test / internal / test_emerge_planner.py View on Github external
def test_raise_on_hole(self):
        source_upload1 = UploadSource(self.recommended_size)
        source_upload2 = UploadSource(self.recommended_size)
        source_copy1 = CopySource(self.recommended_size)
        source_copy2 = CopySource(self.recommended_size)

        write_intents = [
            WriteIntent(source_upload1),
            WriteIntent(source_upload2, destination_offset=self.recommended_size + 2 * MEGABYTE),
            WriteIntent(source_copy1, destination_offset=MEGABYTE),
            WriteIntent(source_copy2, destination_offset=self.recommended_size + 3 * MEGABYTE),
        ]

        hole_msg = ('Cannot emerge file with holes. '
                    'Found hole range: ({}, {})'.format(
                        write_intents[2].destination_end_offset,
                        write_intents[1].destination_offset,
                    ))
        with self.assertRaises(ValueError, hole_msg):
            self.planner.get_emerge_plan(write_intents)
github Backblaze / b2-sdk-python / b2sdk / bucket.py View on Github external
open (and re-open) the file.  The result of opening should be a binary
        file whose read() method returns bytes.

        :param b2sdk.v1.UploadSource upload_source: an object that opens the source of the upload
        :param str file_name: the file name of the new B2 file
        :param str,None content_type: the MIME type, or ``None`` to accept the default based on file extension of the B2 file name
        :param dict,None file_infos: a file info to store with the file or ``None`` to not store anything
        :param int,None min_part_size: the smallest part size to use or ``None`` to determine automatically
        :param b2sdk.v1.AbstractProgressListener,None progress_listener: a progress listener object to use, or ``None`` to not report progress

        The function `opener` should return a file-like object, and it
        must be possible to call it more than once in case the upload
        is retried.
        """
        return self.create_file(
            [WriteIntent(upload_source)],
            file_name,
            content_type=content_type,
            file_info=file_info,
            progress_listener=progress_listener,
            # FIXME: Bucket.upload documents wrong logic
            recommended_upload_part_size=min_part_size,
        )