-
Notifications
You must be signed in to change notification settings - Fork 4.1k
Expand file tree
/
Copy pathio.pxi
More file actions
2912 lines (2355 loc) · 84.2 KB
/
io.pxi
File metadata and controls
2912 lines (2355 loc) · 84.2 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
# Cython wrappers for IO interfaces defined in arrow::io and messaging in
# arrow::ipc
from libc.stdlib cimport malloc, free
from cpython.bytearray cimport PyByteArray_FromStringAndSize
import codecs
import pickle
import re
import sys
import threading
import time
import warnings
from io import BufferedIOBase, IOBase, TextIOBase, UnsupportedOperation
from queue import Queue, Empty as QueueEmpty
from pyarrow.lib cimport check_status, HaveLibHdfs
from pyarrow.util import _is_path_like, _stringify_path
# 64K
DEFAULT_BUFFER_SIZE = 2 ** 16
cdef extern from "Python.h":
# To let us get a PyObject* and avoid Cython auto-ref-counting
PyObject* PyBytes_FromStringAndSizeNative" PyBytes_FromStringAndSize"(
char *v, Py_ssize_t len) except NULL
def have_libhdfs():
"""
Return true if HDFS (HadoopFileSystem) library is set up correctly.
"""
try:
with nogil:
check_status(HaveLibHdfs())
return True
except Exception:
return False
def io_thread_count():
"""
Return the number of threads to use for I/O operations.
Many operations, such as scanning a dataset, will implicitly make
use of this pool. The number of threads is set to a fixed value at
startup. It can be modified at runtime by calling
:func:`set_io_thread_count()`.
See Also
--------
set_io_thread_count : Modify the size of this pool.
cpu_count : The analogous function for the CPU thread pool.
"""
return GetIOThreadPoolCapacity()
def set_io_thread_count(int count):
"""
Set the number of threads to use for I/O operations.
Many operations, such as scanning a dataset, will implicitly make
use of this pool.
Parameters
----------
count : int
The max number of threads that may be used for I/O.
Must be positive.
See Also
--------
io_thread_count : Get the size of this pool.
set_cpu_count : The analogous function for the CPU thread pool.
"""
if count < 1:
raise ValueError("IO thread count must be strictly positive")
check_status(SetIOThreadPoolCapacity(count))
cdef class NativeFile(_Weakrefable):
"""
The base class for all Arrow streams.
Streams are either readable, writable, or both.
They optionally support seeking.
While this class exposes methods to read or write data from Python, the
primary intent of using a Arrow stream is to pass it to other Arrow
facilities that will make use of it, such as Arrow IPC routines.
Be aware that there are subtle differences with regular Python files,
e.g. destroying a writable Arrow stream without closing it explicitly
will not flush any pending data.
"""
# 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):
"""
The file mode. Currently instances of NativeFile may support:
* rb: binary read
* wb: binary write
* rb+: binary read and write
* ab: binary append
"""
# 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):
"""
NOT IMPLEMENTED
"""
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.is_seekable = True
cdef set_input_stream(self, shared_ptr[CInputStream] handle):
self.input_stream = handle
self.random_access.reset()
self.is_seekable = False
cdef set_output_stream(self, shared_ptr[COutputStream] handle):
self.output_stream = handle
cdef shared_ptr[CRandomAccessFile] get_random_access_file(self) except *:
self._assert_readable()
self._assert_seekable()
return self.random_access
cdef shared_ptr[CInputStream] get_input_stream(self) except *:
self._assert_readable()
return self.input_stream
cdef shared_ptr[COutputStream] get_output_stream(self) except *:
self._assert_writable()
return self.output_stream
def _assert_open(self):
if self.closed:
raise ValueError("I/O operation on closed file")
def _assert_readable(self):
self._assert_open()
if not self.is_readable:
# XXX UnsupportedOperation
raise IOError("only valid on readable files")
def _assert_writable(self):
self._assert_open()
if not self.is_writable:
raise IOError("only valid on writable files")
def _assert_seekable(self):
self._assert_open()
if not self.is_seekable:
raise IOError("only valid on seekable files")
def size(self):
"""
Return file size
"""
cdef int64_t size
handle = self.get_random_access_file()
with nogil:
size = GetResultValue(handle.get().GetSize())
return size
def metadata(self):
"""
Return file metadata
"""
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)
return metadata
def tell(self):
"""
Return current stream position
"""
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())
return position
def seek(self, int64_t position, int whence=0):
"""
Change current file stream position
Parameters
----------
position : int
Byte offset, interpreted relative to value of whence argument
whence : int, default 0
Point of reference for seek offset
Notes
-----
Values of whence:
* 0 -- start of stream (the default); offset should be zero or positive
* 1 -- current stream position; offset may be negative
* 2 -- end of stream; offset is usually negative
Returns
-------
int
The new absolute stream position.
"""
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))
return self.tell()
def flush(self):
"""
Flush the stream, if applicable.
An error is raised if stream is not writable.
"""
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:
check_status(handle.get().Flush())
def write(self, data):
"""
Write data to the file.
Parameters
----------
data : bytes-like object or exporter of buffer protocol
Returns
-------
int
nbytes: number of bytes written
"""
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))
return buf.get().size()
def read(self, nbytes=None):
"""
Read and return up to n bytes.
If *nbytes* is None, then the entire remaining file contents are read.
Parameters
----------
nbytes : int, default None
Returns
-------
data : bytes
"""
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)
return PyObject_to_object(obj)
def get_stream(self, file_offset, nbytes):
"""
Return an input stream that reads a file segment independent of the
state of the file.
Allows reading portions of a random access file as an input stream
without interfering with each other.
Parameters
----------
file_offset : int
nbytes : int
Returns
-------
stream : NativeFile
"""
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
return stream
def read_at(self, nbytes, offset):
"""
Read indicated number of bytes at offset from the file
Parameters
----------
nbytes : int
offset : int
Returns
-------
data : bytes
"""
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)
return PyObject_to_object(obj)
def read1(self, nbytes=None):
"""Read and return up to n bytes.
Unlike read(), if *nbytes* is None then a chunk is read, not the
entire file.
Parameters
----------
nbytes : int, default None
The maximum number of bytes to read.
Returns
-------
data : 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
return self.read(nbytes)
def readall(self):
return self.read()
def readinto(self, b):
"""
Read into the supplied buffer
Parameters
----------
b : buffer-like object
A writable buffer object (such as a bytearray).
Returns
-------
written : int
number of bytes written
"""
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))
return bytes_read
def readline(self, size=None):
"""NOT IMPLEMENTED. Read and return a line of bytes from the file.
If size is specified, read at most size bytes.
Line terminator is always b"\\n".
Parameters
----------
size : int
maximum number of bytes read
"""
raise UnsupportedOperation()
def readlines(self, hint=None):
"""NOT IMPLEMENTED. Read lines of the file
Parameters
----------
hint : int
maximum number of bytes read until we stop
"""
raise UnsupportedOperation()
def __iter__(self):
self._assert_readable()
return self
def __next__(self):
line = self.readline()
if not line:
raise StopIteration
return line
def read_buffer(self, nbytes=None):
"""
Read from buffer.
Parameters
----------
nbytes : int, optional
maximum number of bytes read
"""
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))
return pyarrow_wrap_buffer(output)
def truncate(self):
"""
NOT IMPLEMENTED
"""
raise UnsupportedOperation()
def writelines(self, lines):
"""
Write lines to the file.
Parameters
----------
lines : iterable
Iterable of bytes-like objects or exporters of buffer protocol
"""
self._assert_writable()
for line in lines:
self.write(line)
def download(self, stream_or_path, buffer_size=None):
"""
Read this file completely to a local path or destination stream.
This method first seeks to the beginning of the file.
Parameters
----------
stream_or_path : str or file-like object
If a string, a local file path to write to; otherwise,
should be a writable stream.
buffer_size : int, optional
The buffer size to use for data transfers.
"""
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:
raise exc_info[0], exc_info[1], exc_info[2]
def _download_nothreads(self, stream_or_path, buffer_size=None):
"""
Internal method to do a download without separate threads, queues etc.
Called by download above if is_threading_enabled() == False
"""
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)
cleanup()
def upload(self, stream, buffer_size=None):
"""
Write from a source stream to this file.
Parameters
----------
stream : file-like object
Source stream to pipe to this file.
buffer_size : int, optional
The buffer size to use for data transfers.
"""
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:
raise exc_info[0], exc_info[1], exc_info[2]
def _upload_nothreads(self, stream, buffer_size=None):
"""
Internal method to do an upload without separate threads, queues etc.
Called by upload above if is_threading_enabled() == False
"""
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
self.write(buf)
BufferedIOBase.register(NativeFile)
# ----------------------------------------------------------------------
# Python file-like objects
cdef class PythonFile(NativeFile):
"""
A stream backed by a Python file object.
This class allows using Python file objects with arbitrary Arrow
functions, including functions written in another language than Python.
As a downside, there is a non-zero redirection cost in translating
Arrow stream calls to Python method calls. Furthermore, Python's
Global Interpreter Lock may limit parallelism in some situations.
Examples
--------
>>> import io
>>> import pyarrow as pa
>>> pa.PythonFile(io.BytesIO())
<pyarrow.PythonFile closed=False own_file=False is_seekable=False is_writable=True is_readable=False>
Create a stream for writing:
>>> buf = io.BytesIO()
>>> f = pa.PythonFile(buf, mode = 'w')
>>> f.writable()
True
>>> f.write(b'PythonFile')
10
>>> buf.getvalue()
b'PythonFile'
>>> f.close()
>>> f
<pyarrow.PythonFile closed=True own_file=False is_seekable=False is_writable=True is_readable=False>
Create a stream for reading:
>>> buf = io.BytesIO(b'PythonFile')
>>> f = pa.PythonFile(buf, mode = 'r')
>>> f.mode
'rb'
>>> f.read()
b'PythonFile'
>>> f
<pyarrow.PythonFile closed=False own_file=False is_seekable=True is_writable=False is_readable=True>
>>> f.close()
>>> f
<pyarrow.PythonFile closed=True own_file=False is_seekable=True is_writable=False is_readable=True>
"""
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):
"""
Parameters
----------
pos : int, optional
"""
self.handle.truncate(pos)
def readline(self, size=None):
"""
Read and return a line of bytes from the file.
If size is specified, read at most size bytes.
Parameters
----------
size : int
Maximum number of bytes read
"""
return self.handle.readline(size)
def readlines(self, hint=None):
"""