ZetCode

Python os.fdopen 函数

上次修改时间:2025 年 4 月 11 日

本综合指南探讨了 Python 的 os.fdopen 函数,该函数从文件描述符创建一个文件对象。我们将涵盖描述符处理、文件对象创建和实用的底层 I/O 示例。

基本定义

os.fdopen 函数从现有的文件描述符 (fd) 创建一个 Python 文件对象。 它在底层描述符上提供更高级的文件操作。

关键参数:fd(文件描述符),mode(可选的文件模式字符串),buffering(可选的缓冲区大小)。 返回一个包装 fd 的文件对象。

基本的文件描述符到文件对象的转换

此示例演示了使用 os.fdopen 将文件描述符转换为文件对象的最简单方法。 我们首先使用 os.open 创建一个描述符。

basic_conversion.py
import os

# Create a file descriptor
fd = os.open("example.txt", os.O_RDWR | os.O_CREAT)

# Convert descriptor to file object
file_obj = os.fdopen(fd, "r+")

# Use the file object
file_obj.write("Hello World\n")
file_obj.seek(0)
print(file_obj.read())

# Close the file object (also closes the descriptor)
file_obj.close()

该示例显示了完整的生命周期:创建描述符,转换为文件对象,执行 I/O,然后关闭。 请注意,关闭文件对象也会关闭底层描述符。

当您需要在同一个程序中同时使用底层描述符操作和高级文件对象方法时,此方法很有用。

保留文件描述符

默认情况下,当文件对象关闭时,os.fdopen 会关闭描述符。 我们可以使用 closefd=False 参数阻止这种情况发生。

preserve_descriptor.py
import os

# Create descriptor
fd = os.open("data.txt", os.O_RDWR | os.O_CREAT)

# Convert to file object without closing fd
file_obj = os.fdopen(fd, "r+", closefd=False)

# Use file object
file_obj.write("Preserved descriptor\n")
file_obj.close()

# Descriptor is still valid
os.write(fd, b"Additional data\n")
os.close(fd)  # Must close manually

此示例在关闭文件对象后保持描述符有效。 当您需要在文件对象操作后继续使用描述符时,这很有用。

请记住在完成后手动关闭描述符,以避免资源泄漏。

使用标准流

os.fdopen 可以包装标准流描述符 (0, 1, 2) 以进行更方便的 I/O 操作。 此示例演示了包装 stdin。

std_streams.py
import os
import sys

# Wrap stdin (fd 0)
stdin = os.fdopen(0, "r")  # Equivalent to sys.stdin

print("Enter your name:")
name = stdin.readline().strip()
print(f"Hello, {name}!")

# Wrap stdout (fd 1)
stdout = os.fdopen(1, "w")  # Equivalent to sys.stdout
stdout.write(f"Greetings, {name}!\n")
stdout.flush()

该示例演示了如何为标准输入和输出创建文件对象。 在 sys.stdin/stdout 不可用的情况下,这可能很有用。

请注意,这些文件对象与 sys.stdin/stdout 共享相同的底层描述符,因此操作会影响两者。

fdopen 的缓冲模式

os.fdopen 支持不同的缓冲模式,例如常规的 open()。 此示例演示了行缓冲和无缓冲。

buffering_modes.py
import os

# Create descriptor
fd = os.open("output.log", os.O_WRONLY | os.O_CREAT)

# Line buffered file object
line_buffered = os.fdopen(fd, "w", buffering=1)
line_buffered.write("Line 1\n")  # Flushed immediately
line_buffered.write("Line 2\n")

# No buffering (direct write)
fd2 = os.open("direct.log", os.O_WRONLY | os.O_CREAT)
unbuffered = os.fdopen(fd2, "w", buffering=0)
unbuffered.write("Immediate write")
unbuffered.close()

第一个文件对象使用行缓冲 (1),在每个换行符后刷新。 第二个使用无缓冲 (0),直接写入磁盘。

缓冲选择会影响性能和数据安全性 - 无缓冲速度较慢,但可确保立即写入。

使用 fdopen 进行错误处理

此示例演示了在使用文件描述符和 os.fdopen 时进行正确的错误处理。 我们检查无效的描述符。

error_handling.py
import os
import errno

try:
    # Try to open invalid descriptor
    bad_fd = 999
    file_obj = os.fdopen(bad_fd, "r")
except OSError as e:
    if e.errno == errno.EBADF:
        print("Error: Bad file descriptor")
    else:
        print(f"Unexpected error: {e}")

# Safe approach
try:
    fd = os.open("data.txt", os.O_RDONLY)
    file_obj = os.fdopen(fd, "r")
    print(file_obj.read())
except OSError as e:
    print(f"File error: {e}")
finally:
    if 'file_obj' in locals():
        file_obj.close()
    elif 'fd' in locals():
        os.close(fd)

第一部分显示了如何处理无效的描述符。 第二部分演示了一种完整的安全方法,并在 finally 块中进行正确的清理。

始终确保正确关闭描述符和文件对象,即使发生错误也是如此。

使用管道

os.fdopen 通常与 os.pipe() 创建的管道一起使用。 此示例显示了使用管道描述符的进程间通信。

pipe_communication.py
import os

# Create pipe
read_fd, write_fd = os.pipe()

# Convert descriptors to file objects
reader = os.fdopen(read_fd, "r")
writer = os.fdopen(write_fd, "w")

# Write to pipe
writer.write("Message through pipe\n")
writer.flush()  # Ensure data is sent

# Read from pipe
message = reader.readline()
print(f"Received: {message.strip()}")

# Clean up
writer.close()
reader.close()

该示例创建了一个管道,将其描述符转换为文件对象,然后演示了简单的通信。 管道是单向的 - 一个用于读取,一个用于写入。

这种模式是 Python 中进程间通信的基础。

非阻塞文件操作

这个高级示例展示了如何在设置 O_NONBLOCK 标志后,使用 os.fdopen 从描述符创建非阻塞文件对象。

non_blocking.py
import os
import errno
import fcntl

# Create descriptor with non-blocking flag
fd = os.open("fifo", os.O_RDONLY | os.O_NONBLOCK)

# Set non-blocking if not set during open
# fcntl.fcntl(fd, fcntl.F_SETFL, os.O_NONBLOCK)

# Create non-blocking file object
f = os.fdopen(fd, "r")

try:
    data = f.read()
    print(f"Read: {data}")
except IOError as e:
    if e.errno == errno.EAGAIN:
        print("No data available (non-blocking)")
    else:
        print(f"Read error: {e}")

f.close()

该示例演示了如何处理非阻塞 I/O 操作,当没有数据立即可用时,读取可能会因 EAGAIN 而失败。

非阻塞模式对于无法承受阻塞 I/O 操作的事件驱动程序很有用。

安全注意事项

最佳实践

资料来源

作者

我的名字是 Jan Bodnar,我是一位充满激情的程序员,拥有丰富的编程经验。 我从 2007 年开始撰写编程文章。 迄今为止,我已经撰写了超过 1,400 篇文章和 8 本电子书。 我拥有超过十年的编程教学经验。

列出所有 Python 教程