def _write_any()

in alibabacloud_oss_v2/filelike.py [0:0]


    def _write_any(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
        )

        blen = utils.guess_content_length(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' and blen is not None:
                        next_append = self._next_append_stat()
                        if next_append[0] >= 0 and offset + blen == 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