def _iter_part()

in alibabacloud_oss_v2/uploader.py [0:0]

• 26 lines of code
• 7 McCabe index (conditional complexity)

    def _iter_part(self, upload_ctx: _UploadContext):
        start_part_num = upload_ctx.start_num
        reader = self._reader
        if self._reader_seekable:
            reader = io_utils.ReadAtReader(reader)

        def next_body():
            n = self._options.part_size
            if self._reader_seekable:
                bytes_left = self._total_size - self._reader_pos
                if bytes_left < n:
                    n = bytes_left
                body = io_utils.SectionReader(reader, self._reader_pos, n)
            else:
                body = reader.read(n)

            self._reader_pos += len(body)
            return body

        while len(self._upload_errors) == 0:
            try:
                body = next_body()
                if len(body) == 0:
                    break
            except Exception as err:
                self._save_error(err)
                break

            start_part_num += 1
            yield upload_ctx.upload_id, start_part_num, body