1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
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
responsible 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)
|
