"""

try:

with nogil:

check_status(HaveLibHdfs())

return True

except Exception:

"""

"""

if count < 1:

raise ValueError("IO thread count must be strictly positive")

"""

# Default chunk size for chunked reads.

# Use a large enough value for networked filesystems.

_default_chunk_size = 256 * 1024

def __cinit__(self):

self.own_file = False

self.is_readable = False

self.is_writable = False

self.is_seekable = False

self._is_appending = False

def __dealloc__(self):

if self.own_file:

self.close()

def __enter__(self):

return self

def __exit__(self, exc_type, exc_value, tb):

self.close()

def __repr__(self):

name = f"pyarrow.{self.__class__.__name__}"

return (f"<{name} "

f"closed={self.closed} "

f"own_file={self.own_file} "

f"is_seekable={self.is_seekable} "

f"is_writable={self.is_writable} "

f"is_readable={self.is_readable}>")

@property

def mode(self):

"""

# Emulate built-in file modes

if self.is_readable and self.is_writable:

return 'rb+'

elif self.is_readable:

return 'rb'

elif self.is_writable and self._is_appending:

return 'ab'

elif self.is_writable:

return 'wb'

else:

raise ValueError('File object is malformed, has no mode')

def readable(self):

self._assert_open()

return self.is_readable

def writable(self):

self._assert_open()

return self.is_writable

def seekable(self):

self._assert_open()

return self.is_seekable

def isatty(self):

self._assert_open()

return False

def fileno(self):

"""

raise UnsupportedOperation()

@property

def closed(self):

if self.is_readable:

return self.input_stream.get().closed()

elif self.is_writable:

return self.output_stream.get().closed()

else:

return True

def close(self):

if not self.closed:

with nogil:

if self.is_readable:

check_status(self.input_stream.get().Close())

else:

check_status(self.output_stream.get().Close())

cdef set_random_access_file(self, shared_ptr[CRandomAccessFile] handle):

self.input_stream = <shared_ptr[CInputStream]> handle

self.random_access = handle

self._assert_open()

if not self.is_readable:

# XXX UnsupportedOperation

"""

cdef int64_t size

handle = self.get_random_access_file()

with nogil:

size = GetResultValue(handle.get().GetSize())

"""

cdef:

shared_ptr[const CKeyValueMetadata] c_metadata

handle = self.get_input_stream()

with nogil:

c_metadata = GetResultValue(handle.get().ReadMetadata())

metadata = {}

if c_metadata.get() != nullptr:

for i in range(c_metadata.get().size()):

metadata[frombytes(c_metadata.get().key(i))] = \

c_metadata.get().value(i)

"""

cdef int64_t position

if self.is_readable:

rd_handle = self.get_random_access_file()

with nogil:

position = GetResultValue(rd_handle.get().Tell())

else:

wr_handle = self.get_output_stream()

with nogil:

position = GetResultValue(wr_handle.get().Tell())

"""

cdef int64_t offset

handle = self.get_random_access_file()

with nogil:

if whence == 0:

offset = position

elif whence == 1:

offset = GetResultValue(handle.get().Tell())

offset = offset + position

elif whence == 2:

offset = GetResultValue(handle.get().GetSize())

offset = offset + position

else:

with gil:

raise ValueError(f"Invalid value of whence: {whence}")

check_status(handle.get().Seek(offset))

"""

self._assert_open()

# For IOBase compatibility, flush() on an input stream is a no-op

if self.is_writable:

handle = self.get_output_stream()

with nogil:

"""

self._assert_writable()

handle = self.get_output_stream()

cdef shared_ptr[CBuffer] buf = as_c_buffer(data)

with nogil:

check_status(handle.get().WriteBuffer(buf))

"""

cdef:

int64_t c_nbytes

int64_t bytes_read = 0

PyObject* obj

if nbytes is None:

if not self.is_seekable:

# Cannot get file size => read chunkwise

bs = self._default_chunk_size

chunks = []

while True:

chunk = self.read(bs)

if not chunk:

break

chunks.append(chunk)

return b"".join(chunks)

c_nbytes = self.size() - self.tell()

else:

c_nbytes = nbytes

handle = self.get_input_stream()

# Allocate empty write space

obj = PyBytes_FromStringAndSizeNative(NULL, c_nbytes)

cdef uint8_t* buf = <uint8_t*> cp.PyBytes_AS_STRING(<object> obj)

with nogil:

bytes_read = GetResultValue(handle.get().Read(c_nbytes, buf))

if bytes_read < c_nbytes:

cp._PyBytes_Resize(&obj, <Py_ssize_t> bytes_read)

"""

cdef:

shared_ptr[CInputStream] data

int64_t c_file_offset

int64_t c_nbytes

c_file_offset = file_offset

c_nbytes = nbytes

handle = self.get_random_access_file()

data = GetResultValue(

CRandomAccessFile.GetStream(handle, c_file_offset, c_nbytes))

stream = NativeFile()

stream.set_input_stream(data)

stream.is_readable = True

"""

cdef:

int64_t c_nbytes

int64_t c_offset

int64_t bytes_read = 0

PyObject* obj

c_nbytes = nbytes

c_offset = offset

handle = self.get_random_access_file()

# Allocate empty write space

obj = PyBytes_FromStringAndSizeNative(NULL, c_nbytes)

cdef uint8_t* buf = <uint8_t*> cp.PyBytes_AS_STRING(<object> obj)

with nogil:

bytes_read = GetResultValue(handle.get().

ReadAt(c_offset, c_nbytes, buf))

if bytes_read < c_nbytes:

cp._PyBytes_Resize(&obj, <Py_ssize_t> bytes_read)

"""Read and return up to n bytes.

if nbytes is None:

# The expectation when passing `nbytes=None` is not to read the

# entire file but to issue a single underlying read call up to

# a reasonable size (the use case being to read a bufferable

# amount of bytes, such as with io.TextIOWrapper).

nbytes = self._default_chunk_size

"""

cdef:

int64_t bytes_read

uint8_t* buf

Buffer py_buf

int64_t buf_len

handle = self.get_input_stream()

py_buf = py_buffer(b)

buf_len = py_buf.size

buf = py_buf.buffer.get().mutable_data()

with nogil:

bytes_read = GetResultValue(handle.get().Read(buf_len, buf))

"""NOT IMPLEMENTED. Read and return a line of bytes from the file.

"""NOT IMPLEMENTED. Read lines of the file

line = self.readline()

if not line:

raise StopIteration

"""

cdef:

int64_t c_nbytes

int64_t bytes_read = 0

shared_ptr[CBuffer] output

handle = self.get_input_stream()

if nbytes is None:

if not self.is_seekable:

# Cannot get file size => read chunkwise

return py_buffer(self.read())

c_nbytes = self.size() - self.tell()

else:

c_nbytes = nbytes

with nogil:

output = GetResultValue(handle.get().ReadBuffer(c_nbytes))

"""

"""

self._assert_writable()

for line in lines:

"""

cdef:

int64_t bytes_read = 0

uint8_t* buf

if not is_threading_enabled():

return self._download_nothreads(stream_or_path, buffer_size)

handle = self.get_input_stream()

buffer_size = buffer_size or DEFAULT_BUFFER_SIZE

write_queue = Queue(50)

if not hasattr(stream_or_path, 'read'):

stream = open(stream_or_path, 'wb')

def cleanup():

stream.close()

else:

stream = stream_or_path

def cleanup():

pass

done = False

exc_info = None

def bg_write():

try:

while not done or write_queue.qsize() > 0:

try:

buf = write_queue.get(timeout=0.01)

except QueueEmpty:

continue

stream.write(buf)

except Exception as e:

exc_info = sys.exc_info()

finally:

cleanup()

self.seek(0)

writer_thread = threading.Thread(target=bg_write)

# This isn't ideal -- PyBytes_FromStringAndSize copies the data from

# the passed buffer, so it's hard for us to avoid doubling the memory

buf = <uint8_t*> malloc(buffer_size)

if buf == NULL:

raise MemoryError(f"Failed to allocate {buffer_size} bytes")

writer_thread.start()

cdef int64_t total_bytes = 0

cdef int32_t c_buffer_size = buffer_size

try:

while True:

with nogil:

bytes_read = GetResultValue(

handle.get().Read(c_buffer_size, buf))

total_bytes += bytes_read

# EOF

if bytes_read == 0:

break

pybuf = cp.PyBytes_FromStringAndSize(<const char*>buf,

bytes_read)

if writer_thread.is_alive():

while write_queue.full():

time.sleep(0.01)

else:

break

write_queue.put_nowait(pybuf)

finally:

free(buf)

done = True

writer_thread.join()

if exc_info is not None:

"""

cdef:

int64_t bytes_read = 0

uint8_t* buf

handle = self.get_input_stream()

buffer_size = buffer_size or DEFAULT_BUFFER_SIZE

if not hasattr(stream_or_path, 'read'):

stream = open(stream_or_path, 'wb')

def cleanup():

stream.close()

else:

stream = stream_or_path

def cleanup():

pass

self.seek(0)

# This isn't ideal -- PyBytes_FromStringAndSize copies the data from

# the passed buffer, so it's hard for us to avoid doubling the memory

buf = <uint8_t*> malloc(buffer_size)

if buf == NULL:

raise MemoryError(f"Failed to allocate {buffer_size} bytes")

cdef int64_t total_bytes = 0

cdef int32_t c_buffer_size = buffer_size

try:

while True:

with nogil:

bytes_read = GetResultValue(

handle.get().Read(c_buffer_size, buf))

total_bytes += bytes_read

# EOF

if bytes_read == 0:

break

pybuf = cp.PyBytes_FromStringAndSize(<const char*>buf,

bytes_read)

# no background thread - write on main thread

stream.write(pybuf)

finally:

free(buf)

"""

if not is_threading_enabled():

return self._upload_nothreads(stream, buffer_size)

write_queue = Queue(50)

self._assert_writable()

buffer_size = buffer_size or DEFAULT_BUFFER_SIZE

done = False

exc_info = None

def bg_write():

try:

while not done or write_queue.qsize() > 0:

try:

buf = write_queue.get(timeout=0.01)

except QueueEmpty:

continue

self.write(buf)

except Exception as e:

exc_info = sys.exc_info()

writer_thread = threading.Thread(target=bg_write)

writer_thread.start()

try:

while True:

buf = stream.read(buffer_size)

if not buf:

break

if writer_thread.is_alive():

while write_queue.full():

time.sleep(0.01)

else:

break

write_queue.put_nowait(buf)

finally:

done = True

writer_thread.join()

if exc_info is not None:

"""

self._assert_writable()

buffer_size = buffer_size or DEFAULT_BUFFER_SIZE

while True:

buf = stream.read(buffer_size)

if not buf:

break

# no threading - just write

"""

cdef:

object handle

def __cinit__(self, handle, mode=None):

self.handle = handle

if mode is None:

try:

inferred_mode = handle.mode

except AttributeError:

# Not all file-like objects have a mode attribute

# (e.g. BytesIO)

try:

inferred_mode = 'w' if handle.writable() else 'r'

except AttributeError:

raise ValueError("could not infer open mode for file-like "

"object %r, please pass it explicitly"

% (handle,))

else:

inferred_mode = mode

if inferred_mode.startswith('w'):

kind = 'w'

elif inferred_mode.startswith('r'):

kind = 'r'

else:

raise ValueError(f'Invalid file mode: {mode}')

# If mode was given, check it matches the given file

if mode is not None:

if isinstance(handle, IOBase):

# Python 3 IO object

if kind == 'r':

if not handle.readable():

raise TypeError("readable file expected")

else:

if not handle.writable():

raise TypeError("writable file expected")

# (other duck-typed file-like objects are possible)

# If possible, check the file is a binary file

if isinstance(handle, TextIOBase):

raise TypeError("binary file expected, got text file")

if kind == 'r':

self.set_random_access_file(

shared_ptr[CRandomAccessFile](new PyReadableFile(handle)))

self.is_readable = True

else:

self.set_output_stream(

shared_ptr[COutputStream](new PyOutputStream(handle)))

self.is_writable = True

def truncate(self, pos=None):

"""

self.handle.truncate(pos)

def readline(self, size=None):

"""

return self.handle.readline(size)

def readlines(self, hint=None):

"""