Skip to content

Instantly share code, notes, and snippets.

Created October 28, 2014 15:49
Show Gist options
  • Save williballenthin/d3bf8d7a1867aed16d30 to your computer and use it in GitHub Desktop.
Save williballenthin/d3bf8d7a1867aed16d30 to your computer and use it in GitHub Desktop.
FUSE module that exposes a copy of a source directory, but only entries whose `ctime` has changed in the past 10 minutes. Requires `fusepy`.
#!/usr/bin/env python
from __future__ import with_statement
import datetime
from errno import EACCES
from os.path import realpath
from sys import argv, exit
from threading import Lock
import os
from fuse import FUSE, FuseOSError, Operations, LoggingMixIn
class Loopback(LoggingMixIn, Operations):
def __init__(self, root):
self.root = realpath(root)
self.rwlock = Lock()
def __call__(self, op, path, *args):
return super(Loopback, self).__call__(op, self.root + path, *args)
def access(self, path, mode):
if not os.access(path, mode):
raise FuseOSError(EACCES)
chmod = os.chmod
chown = os.chown
def create(self, path, mode):
return, os.O_WRONLY | os.O_CREAT, mode)
def flush(self, path, fh):
return os.fsync(fh)
def fsync(self, path, datasync, fh):
return os.fsync(fh)
def getattr(self, path, fh=None):
st = os.lstat(path)
return dict((key, getattr(st, key)) for key in ('st_atime', 'st_ctime',
'st_gid', 'st_mode', 'st_mtime', 'st_nlink', 'st_size', 'st_uid'))
getxattr = None
def link(self, target, source):
return, target)
listxattr = None
mkdir = os.mkdir
mknod = os.mknod
open =
def read(self, path, size, offset, fh):
with self.rwlock:
os.lseek(fh, offset, 0)
return, size)
def readdir(self, path, fh):
ret = [".", ".."]
for p in os.listdir(path):
ctime = datetime.datetime.fromtimestamp(os.path.getctime(os.path.join(path, p)))
if ctime > - datetime.timedelta(minutes=10):
return ret
readlink = os.readlink
def release(self, path, fh):
return os.close(fh)
def rename(self, old, new):
return os.rename(old, self.root + new)
rmdir = os.rmdir
def statfs(self, path):
stv = os.statvfs(path)
return dict((key, getattr(stv, key)) for key in ('f_bavail', 'f_bfree',
'f_blocks', 'f_bsize', 'f_favail', 'f_ffree', 'f_files', 'f_flag',
'f_frsize', 'f_namemax'))
def symlink(self, target, source):
return os.symlink(source, target)
def truncate(self, path, length, fh=None):
with open(path, 'r+') as f:
unlink = os.unlink
utimens = os.utime
def write(self, path, data, offset, fh):
with self.rwlock:
os.lseek(fh, offset, 0)
return os.write(fh, data)
if __name__ == '__main__':
if len(argv) != 3:
print('usage: %s <root> <mountpoint>' % argv[0])
fuse = FUSE(Loopback(argv[1]), argv[2], foreground=True)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment