ファイルのリストのためにBufferedReaderがファイルを閉じるのを防ぐ

Question

私はBufferedReaderを継承するクラスとファイルストリームのリストを持っています b.close（）は最後のストリームを除くすべてのものに対して呼び出されます。 どうすればいいですか？実装BufferedIOBaseクラスでIOBaseオブジェクトをラップ

 

class TestReader(BufferedReader): 
    pass 

def test(streams): 
    for stream in streams: 
     b=TestReader(stream) 
     do_something(b) 
    #all the streams except streams[-1] are closed, how do I prevent this? 

streams=[open('test1.txt','rb'),open('test2.txt','rb')] 
test(streams) 
streams.do_something_else() 

Answer 1

にもかかわらず

おかげで、そのインターフェースは、ストリーム（IOBaseから継承すべて）ですので、IOBaseオブジェクトの通常の動作は、自分自身を閉じることです彼らが範囲外に出たとき。 BufferedIOBaseは、close()呼び出しを基になるストリームに委譲します。

BufferedReaderをストリームラッパー（これは実装されていますが）として表示するのではなく、既存のストリームのタイプキャストとして表示してください。 2つのストリームの状態は完全に結合されています。ただし、ラップされたストリームをdetach()でアンバインドできますが、これによりBufferedIOBaseオブジェクトは役に立たなくなります。

さらに、io.openは、モードがrbのときにすでにBufferedReaderを返しているので、ダブルバッファリングしています。代わりにio.FileIOを使用してください。

あなたは、いくつかの選択肢があります。

1. は、新たなストリームと新しい基盤となるファイルディスクリプタを作成し、ファイル名の代わりに、ストリームの周りに渡します。これはあなたの最も簡単で安全なオプションです。

2. 生ファイル記述子を作成し、必要に応じてストリームを作成します。これは、複数のストリームが同じファイル記述子を同時に使用していないことに注意してください。たとえば、次のように

   fd = os.open('test.txt', os.O_RDONLY) 
   file1 = FileIO(fd, 'r', closefd=False) 
   file2 = FileIO(fd, 'r', closefd=False) 
   
   file1.read(100) 
   assert file1.tell() == 100 
   file2.read(100) 
   assert file1.tell() == 200 
   

3. detach()あなたBufferedIOBaseオブジェクトがそのストリームをクローズする前に、基本となるストリーム。 （！ストリームを巻き戻すのを忘れないでください）

   def test(streams): 
       for stream in streams: 
        b=TestReader(stream) 
        do_something(b) 
        wrappedstream = b.detach() 
        assert wrappedstream is stream 
   

   あなたもあなたのデストラクタでこれを実装することができます。

   class TestReader(BufferedReader): 
       def __del__(self): 
        self.detach() 
        # self.raw will not be closed, 
        # rather left in the state it was in at detachment 
   

   それとも、あなたは意味論は間違っていると思えば完全にclose()委任を無効にします。

   class TestReader(BufferedReader): 
       def close(self): 
        self.closed = True 
   

を

私はあなたがしていることの大きな写真を持っていません（おそらくあなたは別のdes IGN）が、これは私が私が見るのコードを実装する方法を次のとおりです。

from io import FileIO, BufferedReader 
import io 
import os 

class TestReader(BufferedReader): 
    pass 

def test(streams): 
    for stream in streams: 
     b = TestReader(stream) 

def test_reset(streams): 
    """Will try to leave stream state unchanged""" 
    for stream in streams: 
     pos = stream.tell() 
     b = TestReader(stream) 
     do_something(b) 
     b.detach() 
     stream.seek(pos) 



filenames = ['test1.txt', 'test2.txt'] 

# option 1: just make new streams 

streams = [FileIO(name, 'r') for name in filenames] 
test(streams) 
streams = [io.open(name, 'rb') for name in filenames] 
#etc 


# option 2: use file descriptors 
fds = [os.open(name, os.O_RDONLY) for name in filenames] 
#closefd = False means "do not close fd on __del__ or __exit__" 
#this is only an option when you pass a fd instead of a file name 
streams = [FileIO(fd, 'r', closefd=False) for fd in fds] 
test(streams) 
streams = [] 
for fd in fds: 
    os.lseek(fd, 0, os.SEEK_SET) 
    streams.append(io.open(fd, 'rb', closefd=False)) 
    # you can also .seek(0) on the BufferedReader objects 
    # instead of os.lseek on the fds 


# option 3: detach 

streams = [FileIO(name, 'r') for name in filenames] 
test_reset(streams) 
# streams[*] should still be in the same state as when you passed it in