Python SpooledTemporaryFile 怎么手动触发写入磁盘

要让 Python 的 SpooledTemporaryFile 在数据尚未超过 max_size 时主动写入磁盘，直接调用它的公开方法 rollover()。这个调用会把当前内存缓冲中的内容迁移到底层 TemporaryFile，同时保留当前文件位置，之后仍然通过原来的对象继续读写。

官方文档：https://docs.python.org/3.14/library/tempfile.html

什么时候需要主动落盘

SpooledTemporaryFile 的价值是先把小数据留在内存里，超过阈值再切换到磁盘。但工程里经常会提前遇到“必须是真实文件”的边界：例如下游 C 扩展需要文件描述符、准备把对象交给只接受 OS 文件的接口，或希望在进入高内存阶段前主动释放内存缓冲。

这时不必写入无意义的大块数据来撞过阈值，也不需要读取 _file、_rolled 这类私有属性。rollover() 就是标准库为这个场景提供的显式入口，而且重复调用不会再次迁移。

先完成一个最小手动落盘实验

下面创建一个 1 MiB 阈值的二进制临时文件，只写入几十字节，然后在远未达到阈值时主动调用 rollover()。

from tempfile import SpooledTemporaryFile

payload = b"report-part-a\nreport-part-b\n"

# 使用上下文管理器，离开代码块时自动关闭底层临时文件
with SpooledTemporaryFile(max_size=1024 * 1024, mode="w+b") as temp_file:
    temp_file.write(payload)
    position_before = temp_file.tell()

    # 不等待超过 max_size，立即把内存缓冲迁移到磁盘临时文件
    temp_file.rollover()

    # rollover 保留当前文件位置；fileno 此时返回真实文件描述符
    position_after = temp_file.tell()
    descriptor = temp_file.fileno()

    print(position_before, position_after)
    print(descriptor)

    # 回到开头读取，确认迁移后的内容保持不变
    temp_file.seek(0)
    print(temp_file.read())

检查时重点看两件事：position_before 与 position_after 应一致，重新定位到开头后读出的字节应与 payload 相同。文件描述符的具体整数值由操作系统分配，不应写进业务判断。

为什么原来的对象还能继续读写

从调用方看，SpooledTemporaryFile 始终是同一个文件式对象。CPython 的实现会先保存旧缓冲区的当前位置，创建新的 TemporaryFile，复制现有内容，再把新文件定位回原来的位置。迁移完成后，write()、seek()、read()、truncate() 等操作都转发给新的底层文件。

这也解释了为什么手动落盘后通常不需要重新赋值：

from tempfile import SpooledTemporaryFile

with SpooledTemporaryFile(max_size=4096, mode="w+b") as buffer:
    buffer.write(b"header|")

    # 迁移后仍使用同一个 buffer 引用，不需要接收返回值
    result = buffer.rollover()
    assert result is None

    buffer.write(b"body")
    buffer.seek(0)

    # 迁移前后的数据对调用方保持为一个连续文件
    assert buffer.read() == b"header|body"

rollover() 的职责是改变内部存储位置，不是返回一个新的文件对象。把它写成 buffer = buffer.rollover() 会把变量变成 None，这是最常见的误用之一。

三种触发方式有什么区别

SpooledTemporaryFile 有三种常见的落盘触发来源，它们最后都让后续操作基于磁盘临时文件继续进行。

其中 fileno() 不是纯查询：官方文档明确说明，调用它会让内容写入磁盘。因此，如果你只是想知道对象当前是否仍在内存里，不要用 fileno() 做探测，因为探测动作本身会改变状态。

max_size 的边界要先想清楚

max_size 控制自动滚动的容量阈值，但手动调用 rollover() 不受它限制。根据 CPython 实现，自动检查要求 max_size 为非零值，并且当前文件位置超过该值；因此默认的 max_size=0 不会因为普通写入而按容量自动落盘，但 rollover() 与 fileno() 仍然可以触发。

from tempfile import SpooledTemporaryFile

# max_size=0 时不依赖容量自动滚动，适合由业务节点决定落盘时机
with SpooledTemporaryFile(max_size=0, mode="w+b") as temp_file:
    temp_file.write(b"small payload")

    # 在交给需要真实文件的组件之前显式迁移
    temp_file.rollover()
    descriptor = temp_file.fileno()
    print(descriptor)

还要注意，阈值判断使用的是当前文件位置而不是抽象的“累计写入字节数”。如果代码频繁 seek() 后覆盖已有区域，文件位置与累计写入量可能不是一回事。需要严格控制磁盘占用时，仍应结合业务数据规模设计阈值，而不是把 max_size 当作文件最终大小的硬上限。

不要依赖临时文件一定有路径

落盘不等于一定能拿到稳定、可见的文件路径。底层使用的是 TemporaryFile：在 POSIX 系统上，目录项可能根本不创建，或创建后立即删除；不同平台的 name 表现也可能不同。若第三方接口接收文件对象或文件描述符，直接传递即可。

如果外部程序明确要求一个可见路径，应改用 NamedTemporaryFile，并根据平台处理关闭、重开和删除策略，而不是强行从 SpooledTemporaryFile 推导路径。

把清理责任交给上下文管理器

无论对象仍在内存还是已经落盘，都建议放进 with。退出上下文时会关闭底层对象；迁移到磁盘后创建的临时资源也随之清理。发生异常时同样能执行关闭动作。

from tempfile import SpooledTemporaryFile

def prepare_upload(data: bytes) -> int:
    # with 负责异常路径上的关闭与临时资源清理
    with SpooledTemporaryFile(max_size=2 * 1024 * 1024, mode="w+b") as file_obj:
        file_obj.write(data)

        # 下游需要描述符前主动落盘，使状态切换发生在可控位置
        file_obj.rollover()
        file_obj.seek(0)

        # 示例只返回长度；真实项目可在此调用接收文件对象的下游接口
        return len(file_obj.read())

快速检查清单

• 需要立刻落盘时，调用 temp_file.rollover()，不要给返回值重新赋值。
• 调用前后继续使用同一个 SpooledTemporaryFile 对象。
• 需要重新读取时显式 seek(0)，因为 rollover 会保留原有文件位置。
• 记住 fileno() 本身也会触发落盘。
• 不要依赖私有属性判断状态，也不要假设落盘后一定有可见路径。
• 使用 with 保证关闭和临时资源清理。

常见问题

rollover() 可以重复调用吗？

可以。对象已经迁移后，再次调用会直接返回，不会重复复制内容。

rollover() 会把文件指针移到开头吗？

不会。实现会保存并恢复原位置。若下一步要完整读取，仍需自己调用 seek(0)。

只调用 flush() 能强制落盘吗？

不能把它等同于 rollover()。在对象仍使用内存缓冲时，flush() 只是刷新当前底层对象；需要保证切换到磁盘临时文件时，应明确调用 rollover()。