我们从Python开源项目中,提取了以下30个代码示例,用于说明如何使用stat.ST_DEV。
def emit(self, record): """ Emit a record. First check if the underlying file has changed, and if it has, close the old stream and reopen the file to get the current stream. """ if not os.path.exists(self.baseFilename): stat = None changed = 1 else: stat = os.stat(self.baseFilename) changed = (stat[ST_DEV] != self.dev) or (stat[ST_INO] != self.ino) if changed and self.stream is not None: self.stream.flush() self.stream.close() self.stream = self._open() if stat is None: stat = os.stat(self.baseFilename) self.dev, self.ino = stat[ST_DEV], stat[ST_INO] logging.FileHandler.emit(self, record)
def __init__(self, filename, mode='a', encoding=None, delay=0): logging.FileHandler.__init__(self, filename, mode, encoding, delay) if not os.path.exists(self.baseFilename): self.dev, self.ino = -1, -1 else: stat = os.stat(self.baseFilename) self.dev, self.ino = stat[ST_DEV], stat[ST_INO]
def _statstream(self): if self.stream: sres = os.fstat(self.stream.fileno()) self.dev, self.ino = sres[ST_DEV], sres[ST_INO]
def emit(self, record): """ Emit a record. First check if the underlying file has changed, and if it has, close the old stream and reopen the file to get the current stream. """ # Reduce the chance of race conditions by stat'ing by path only # once and then fstat'ing our new fd if we opened a new log stream. # See issue #14632: Thanks to John Mulligan for the problem report # and patch. try: # stat the file by path, checking for existence sres = os.stat(self.baseFilename) except OSError as err: if err.errno == errno.ENOENT: sres = None else: raise # compare file system stat with that of our stream file handle if not sres or sres[ST_DEV] != self.dev or sres[ST_INO] != self.ino: if self.stream is not None: # we have an open file handle, clean it up self.stream.flush() self.stream.close() self.stream = None # See Issue #21742: _open () might fail. # open a new file handle and get new stat info from that fd self.stream = self._open() self._statstream() logging.FileHandler.emit(self, record)
def fetch_js(): libs = { 'd3.v3.min.js': 'https://d3js.org/d3.v3.min.js', 'd3.v4.min.js': 'https://d3js.org/d3.v4.min.js', 'svg-crowbar.js': 'https://nytimes.github.io/svg-crowbar/svg-crowbar.js', } for js, url in libs.iteritems(): target = os.path.sep.join(('LemonGraph', 'data', js)) source = os.path.sep.join(('deps', 'js', js)) dotsource = os.path.sep.join(('deps', 'js', '.%s' % js)) try: os.mkdir(os.path.sep.join(('deps', 'js'))) except OSError: pass try: s1 = os.stat(source) except OSError: with open(dotsource, 'wb') as fh: for chunk in do_curl(url): fh.write(chunk) s1 = os.fstat(fh.fileno()) fh.close() os.link(dotsource, source) os.unlink(dotsource) try: s2 = os.stat(target) except OSError: print >> sys.stderr, "Hard linking: %s -> %s" % (source, target) os.link(source, target) continue for i in (stat.ST_INO, stat.ST_DEV): if s1[i] != s2[i]: os.unlink(target) print >> sys.stderr, "Hard linking: %s -> %s" % (source, target) os.link(source, target) break
def emit(self, record): """ Emit a record. First check if the underlying file has changed, and if it has, close the old stream and reopen the file to get the current stream. """ # Reduce the chance of race conditions by stat'ing by path only # once and then fstat'ing our new fd if we opened a new log stream. # See issue #14632: Thanks to John Mulligan for the problem report # and patch. try: # stat the file by path, checking for existence sres = os.stat(self.baseFilename) except OSError as err: if err.errno == errno.ENOENT: sres = None else: raise # compare file system stat with that of our stream file handle if not sres or sres[ST_DEV] != self.dev or sres[ST_INO] != self.ino: if self.stream is not None: # we have an open file handle, clean it up self.stream.flush() self.stream.close() # open a new file handle and get new stat info from that fd self.stream = self._open() self._statstream() logging.FileHandler.emit(self, record)
def emit(self, record): """ Emit a record. First check if the underlying file has changed, and if it has, close the old stream and reopen the file to get the current stream. """ # Reduce the chance of race conditions by stat'ing by path only # once and then fstat'ing our new fd if we opened a new log stream. # See issue #14632: Thanks to John Mulligan for the problem report # and patch. try: # stat the file by path, checking for existence sres = os.stat(self.baseFilename) except FileNotFoundError: sres = None # compare file system stat with that of our stream file handle if not sres or sres[ST_DEV] != self.dev or sres[ST_INO] != self.ino: if self.stream is not None: # we have an open file handle, clean it up self.stream.flush() self.stream.close() self.stream = None # See Issue #21742: _open () might fail. # open a new file handle and get new stat info from that fd self.stream = self._open() self._statstream() logging.FileHandler.emit(self, record)
def emit(self, record): """ Emit a record. First check if the underlying file has changed, and if it has, close the old stream and reopen the file to get the current stream. """ # Reduce the chance of race conditions by stat'ing by path only # once and then fstat'ing our new fd if we opened a new log stream. # See issue #14632: Thanks to John Mulligan for the problem report # and patch. try: # stat the file by path, checking for existence sres = os.stat(self.baseFilename) except OSError as e: if e.errno == errno.ENOENT: sres = None else: raise # compare file system stat with that of our stream file handle if not sres or sres[ST_DEV] != self.dev or sres[ST_INO] != self.ino: # Fixed by xp 2017 Apr 03: # os.fstat still gets OSError(errno=2), although it operates # directly on fd instead of path. The same for stream.flush(). # Thus we keep on trying this close/open/stat loop until no # OSError raises. for ii in range(16): try: if self.stream is not None: # we have an open file handle, clean it up self.stream.flush() self.stream.close() # See Issue #21742: _open () might fail. self.stream = None # open a new file handle and get new stat info from # that fd self.stream = self._open() self._statstream() break except OSError as e: if e.errno == errno.ENOENT: continue else: raise logging.FileHandler.emit(self, record)
def make_file_dict_python(filename): """Create the data dictionary using a Python call to os.lstat We do this on Windows since Python's implementation is much better than the one in cmodule.c Eventually, we will move to using this on all platforms since CPUs have gotten much faster than they were when it was necessary to write cmodule.c """ try: statblock = os.lstat(filename) except os.error: return {'type':None} data = {} mode = statblock[stat.ST_MODE] if stat.S_ISREG(mode): type_ = 'reg' elif stat.S_ISDIR(mode): type_ = 'dir' elif stat.S_ISCHR(mode): type_ = 'dev' s = statblock.st_rdev data['devnums'] = ('c',) + (s >> 8, s & 0xff) elif stat.S_ISBLK(mode): type_ = 'dev' s = statblock.st_rdev data['devnums'] = ('b',) + (s >> 8, s & 0xff) elif stat.S_ISFIFO(mode): type_ = 'fifo' elif stat.S_ISLNK(mode): type_ = 'sym' data['linkname'] = os.readlink(filename) elif stat.S_ISSOCK(mode): type_ = 'sock' else: raise C.UnknownFileError(filename) data['type'] = type_ data['size'] = statblock[stat.ST_SIZE] data['perms'] = stat.S_IMODE(mode) data['uid'] = statblock[stat.ST_UID] data['gid'] = statblock[stat.ST_GID] data['inode'] = statblock[stat.ST_INO] data['devloc'] = statblock[stat.ST_DEV] data['nlink'] = statblock[stat.ST_NLINK] if os.name == 'nt': global type if type(filename) == unicode: attribs = win32file.GetFileAttributesW(filename) else: attribs = win32file.GetFileAttributes(filename) if attribs & winnt.FILE_ATTRIBUTE_REPARSE_POINT: data['type'] = 'sym' data['linkname'] = None if not (type_ == 'sym' or type_ == 'dev'): # mtimes on symlinks and dev files don't work consistently data['mtime'] = long(statblock[stat.ST_MTIME]) data['atime'] = long(statblock[stat.ST_ATIME]) data['ctime'] = long(statblock[stat.ST_CTIME]) return data