Watch at Ease세상을 읽고, 일상의 답을 찾다.

기술 · 실용 가이드

Python SpooledTemporaryFile이 디스크로 가거나 비어 있나요? rollover와 offset 감사

library나 subprocess로 넘기기 전에 숨은 fileno rollover와 EOF cursor를 분리하세요.

SpooledTemporaryFile 세 가지 rollover trigger 도표
원본 일러스트: 작은 payload도 fileno 때문에 disk로 갑니다. Watch at Ease

SpooledTemporaryFile에는 독립된 두 함정이 있습니다. consumer가 fileno()를 호출하면 작은 payload도 disk로 가고, write 뒤 EOF cursor는 downstream에 빈 데이터를 줍니다.

public API로 trigger를 찾고 handoff 전에 seek(0), bytes 또는 descriptor contract를 정한 뒤 실제 consumer에서 delivery와 storage policy를 검증하세요.

1. storage state와 stream position을 분리하세요

consumer에 넘기는 object, max_size, write bytes, tell(), fileno() call을 기록하세요. memory에 있으면서 EOF일 수도, disk에 있으면서 같은 EOF일 수도 있습니다.

private _file/_rolled를 contract로 쓰지 말고 격리 test에서 temp directory를 관찰하며 소유한 consumer boundary만 감싸세요.

2. rollover trigger를 명시하세요

Python trigger는 nonzero max_size 초과, fileno(), 명시적 rollover()입니다. current CPython의 default 0은 size trigger를 끄지만 나머지는 disk로 보냅니다.

실제 library/framework/subprocess를 감사하세요. descriptor-only API는 숨게 fileno()를 부를 수 있어 no-disk policy와 호환되지 않습니다.

3. handoff 전에 public cursor를 되돌리세요

write 뒤 다음 reader 전에 seek(0)하세요. CPython issue의 empty stdin은 cursor가 끝에 있었기 때문이며 object가 잘못된 것이 아닙니다. rollover는 rewind하지 않습니다.

FastAPI UploadFile async code는 await upload.seek(0), sync code는 public file method를 쓰고 final close owner를 정하세요.

4. 하나의 handoff contract를 선택하세요

consumer가 bytes를 받으면 pipe, BytesIO 또는 read result로 no-disk를 유지하세요. descriptor가 필요하면 명시적으로 rollover하고 temp space, seek(0), handoff, owner close를 계약화하세요.

durable output은 caller가 정한 destination으로 복사하고 필요하면 fsync/atomic replace를 정하세요. temporary file은 persistence interface가 아닙니다.

5. 실제 boundary에서 두 축을 테스트하세요

distinct payload를 쓰고 tell(), rewind, real consumer, exact result를 확인하며 below-threshold와 forced-rollover case를 모두 실행하세요.

fileno() 기록 test double과 격리 temp directory 관찰을 추가해 delivery, storage policy, close cleanup을 각각 검증하세요.

spooled stream handoff와 cursor 승인 도표
원본 일러스트: cursor를 되돌리고 실제 consumer에서 bytes를 검증합니다. Watch at Ease