Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
buffering (only usable in text mode), and an integer > 1 to indicate
the size in bytes of a fixed-size chunk buffer.
:param int size:
Size of data to that will be transmitted.
:param bool block_transfer:
If block transfer should be used.
:param bool force_segment:
Force use of segmented download regardless of data size.
:returns:
A file like object.
"""
buffer_size = buffering if buffering > 1 else io.DEFAULT_BUFFER_SIZE
if "r" in mode:
if block_transfer:
raw_stream = BlockUploadStream(self, index, subindex)
else:
raw_stream = ReadableStream(self, index, subindex)
if buffering:
buffered_stream = io.BufferedReader(raw_stream, buffer_size=buffer_size)
else:
return raw_stream
if "w" in mode:
if block_transfer:
raw_stream = BlockDownloadStream(self, index, subindex, size)
else:
raw_stream = WritableStream(self, index, subindex, size, force_segment)
if buffering:
buffered_stream = io.BufferedWriter(raw_stream, buffer_size=buffer_size)
else:
return raw_stream
if "b" not in mode:
def close(self):
if self.closed:
return
super(BlockUploadStream, self).close()
if self._done:
request = bytearray(8)
request[0] = REQUEST_BLOCK_UPLOAD | END_BLOCK_TRANSFER
self.sdo_client.send_request(request)