파일 목록에 대해 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을 스트림 래퍼로 표시해야하지만 (구현 방법은 다르지만) 기존 스트림의 유형 변환이라고 봐야합니다. 두 스트림의 상태가 완전히 연결됩니다. 그러나 래핑 된 스트림을 detach()으로 바인딩 해제 할 수 있지만 BufferedIOBase 객체는 쓸모가 없습니다.

또한 io.open은 rb 일 때 이미 BufferedReader을 반환하므로 이중 버퍼링을 수행합니다. 대신 io.FileIO을 사용해야합니다.

1. 새로운 스트림과 새로운 기본 파일 디스크립터를 생성하고 파일 이름 대신 스트림 주위에 전달합니다

   당신은 몇 가지 선택이있다. 이것은 가장 쉽고 안전한 옵션입니다.

2. 원시 파일 설명자를 만들고 필요한 경우 스트림을 만듭니다. 이를 위해서는 여러 개의 스트림이 동일한 파일 설명자를 동시에 사용하지 않도록주의해야합니다 ( ). 예를 들면 : 당신의 BufferedIOBase 객체 전에

   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() 기본이되는 스트림은 스트림을 닫습니다.

   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() 위임을 해제 : 당신은 심지어 당신의 소멸자에서 이것을 구현할 수

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

   (! 스트림을 되감기하는 것을 잊지 마십시오)

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

저는 현재하고있는 일에 대한 큰 그림을 가지고 있지 않습니다 (아마도 다른 데스가 필요합니다. 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