add a FileDescriptor type

It serves two main purposes. For one thing, it allows telling bare
integers and file descriptors apart on a typing level similar to a
NewType. For another it adds common methods to a file descriptor and
enables closing it via a context manager.

1 files changed, 75 insertions, 0 deletions

diff --git a/linuxnamespaces/filedescriptor.py b/linuxnamespaces/filedescriptor.py
new file mode 100644
index 0000000..4395a54
--- /dev/null
+++ b/linuxnamespaces/filedescriptor.py
@@ -0,0 +1,75 @@
+# Copyright 2024 Helmut Grohne <helmut@subdivi.de>
+# SPDX-License-Identifier: GPL-3
+
+"""A type tag for integers that represent file descriptors."""
+
+import fcntl
+import os
+import typing
+
+
+class FileDescriptor(int):
+    """Type tag for integers that represent file descriptors. It also provides
+    a few very generic file descriptor methods.
+    """
+
+    def __enter__(self) -> "FileDescriptor":
+        """When used as a context manager, close the file descriptor on scope
+        exit.
+        """
+        return self
+
+    def __exit__(self, *args: typing.Any) -> None:
+        """When used as a context manager, close the file descriptor on scope
+        exit.
+        """
+        self.close()
+
+    def close(self) -> None:
+        """Close the file descriptor. Since int is immutable, the caller is
+        responsibe for not closing twice.
+        """
+        os.close(self)
+
+    def dup(self, inheritable: bool = True) -> "FileDescriptor":
+        """Return a duplicate of the file descriptor."""
+        if inheritable:
+            return FileDescriptor(os.dup(self))
+        return FileDescriptor(fcntl.fcntl(self, fcntl.F_DUPFD_CLOEXEC, 0))
+
+    def dup2(self, fd2: int, inheritable: bool = True) -> "FileDescriptor":
+        """Duplicate the file to the given file descriptor number."""
+        return FileDescriptor(os.dup2(self, fd2, inheritable))
+
+    def fileno(self) -> int:
+        """Return self such that it satisfies the HasFileno protocol."""
+        return self
+
+    def get_blocking(self) -> bool:
+        """Get the blocking mode of the file descriptor."""
+        return os.get_blocking(self)
+
+    def get_inheritable(self) -> bool:
+        """Get the close-on-exec flag of the file descriptor."""
+        return os.get_inheritable(self)
+
+    @classmethod
+    def pipe(
+        cls, blocking: bool = True, inheritable: bool = True
+    ) -> tuple["FileDescriptor", "FileDescriptor"]:
+        """Create a pipe with flags set atomically. This actually corresponds
+        to the pipe2 syscall, but skipping flags is equivalent to calling pipe.
+        """
+        rfd, wfd = os.pipe2(
+            (0 if blocking else os.O_NONBLOCK)
+            | (0 if inheritable else os.O_CLOEXEC),
+        )
+        return (cls(rfd), cls(wfd))
+
+    def set_blocking(self, blocking: bool) -> None:
+        """Set the blocking mode of the file descriptor."""
+        os.set_blocking(self, blocking)
+
+    def set_inheritable(self, inheritable: bool) -> None:
+        """Set the close-on-exec flag of the file descriptor."""
+        os.set_inheritable(self, inheritable)