def _write_bytes()

in alibabacloud_oss_v2/filelike.py [0:0]

• 34 lines of code
• 8 McCabe index (conditional complexity)

    def _write_bytes(self, b):
        offset = self._offset
        hash_crc64 = self._hash_crc64
        error: Exception = None
        request = models.AppendObjectRequest(
            bucket=self._bucket,
            key=self._key,
            position=offset,
            request_payer=self._request_payer,
            body=b
        )

        self._apply_create_param_if_need(request)

        try:
            result = self._client.append_object(request)
            offset = result.next_position
            hash_crc64 = result.hash_crc64
        except Exception as err:
            error = err
            if isinstance(err, exceptions.OperationError):
                serr = err.unwrap()
                if isinstance(serr, exceptions.ServiceError):
                    if serr.code == 'PositionNotEqualToLength':
                        next_append = self._next_append_stat()
                        if next_append[0] >= 0 and offset + len(b) == next_append[0]:
                            error = None
                            offset = next_append[0]
                            hash_crc64 = next_append[1]

        if error:
            raise self._wrap_error('write', error)

        writern = offset - self._offset
        self._created = True
        self._offset = offset
        self._hash_crc64 = hash_crc64

        return writern