我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用stat.S_ISFIFO。
def copyfile(src, dst): """Copy data from src to dst""" if _samefile(src, dst): raise Error("`%s` and `%s` are the same file" % (src, dst)) for fn in [src, dst]: try: st = os.stat(fn) except OSError: # File most likely does not exist pass else: # XXX What about other special files? (sockets, devices...) if stat.S_ISFIFO(st.st_mode): raise SpecialFileError("`%s` is a named pipe" % fn) with open(src, 'rb') as fsrc: with open(dst, 'wb') as fdst: copyfileobj(fsrc, fdst)
def __hashEntry(self, prefix, entry, s): if stat.S_ISREG(s.st_mode): digest = self.__index.check(prefix, entry, s, hashFile) elif stat.S_ISDIR(s.st_mode): digest = self.__hashDir(prefix, entry) elif stat.S_ISLNK(s.st_mode): digest = self.__index.check(prefix, entry, s, DirHasher.__hashLink) elif stat.S_ISBLK(s.st_mode) or stat.S_ISCHR(s.st_mode): digest = struct.pack("<L", s.st_rdev) elif stat.S_ISFIFO(s.st_mode): digest = b'' else: digest = b'' logging.getLogger(__name__).warning("Unknown file: %s", entry) return digest
def _setup_input_pipes(input_pipes): """ Given a mapping of input pipes, return a tuple with 2 elements. The first is a list of file descriptors to pass to ``select`` as writeable descriptors. The second is a dictionary mapping paths to existing named pipes to their adapters. """ wds = [] fifos = {} for pipe, adapter in six.viewitems(input_pipes): if isinstance(pipe, int): # This is assumed to be an open system-level file descriptor wds.append(pipe) else: if not os.path.exists(pipe): raise Exception('Input pipe does not exist: %s' % pipe) if not stat.S_ISFIFO(os.stat(pipe).st_mode): raise Exception('Input pipe must be a fifo object: %s' % pipe) fifos[pipe] = adapter return wds, fifos
def __init__(self, loop, pipe, protocol, waiter=None, extra=None): super().__init__(extra) self._extra['pipe'] = pipe self._loop = loop self._pipe = pipe self._fileno = pipe.fileno() mode = os.fstat(self._fileno).st_mode if not (stat.S_ISFIFO(mode) or stat.S_ISSOCK(mode) or stat.S_ISCHR(mode)): raise ValueError("Pipe transport is for pipes/sockets only.") _set_nonblocking(self._fileno) self._protocol = protocol self._closing = False self._loop.call_soon(self._protocol.connection_made, self) # only start reading when connection_made() has been called self._loop.call_soon(self._loop.add_reader, self._fileno, self._read_ready) if waiter is not None: # only wake up the waiter when connection_made() has been called self._loop.call_soon(waiter._set_result_unless_cancelled, None)
def is_special_file(path): """ This function checks to see if a special file. It checks if the file is a character special device, block special device, FIFO, or socket. """ mode = os.stat(path).st_mode # Character special device. if stat.S_ISCHR(mode): return True # Block special device if stat.S_ISBLK(mode): return True # FIFO. if stat.S_ISFIFO(mode): return True # Socket. if stat.S_ISSOCK(mode): return True return False
