Commit 5e7855e0 authored by Alexander Pushkov's avatar Alexander Pushkov Committed by Alexander Pushkov

initial commit

parents
/mnt/
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
env/
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*,cover
.hypothesis/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# pyenv
.python-version
# celery beat schedule file
celerybeat-schedule
# SageMath parsed files
*.sage.py
# dotenv
.env
# virtualenv
.venv
venv/
ENV/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
Defuse
=======
A thin abstraction layer on top of `FUSE`_ and `fusepy`_. It is a work in progress.
.. _FUSE: https://fuse.sourceforge.net/
.. _fusepy: https://github.com/terencehonles/fusepy
.. code-block:: Python
from defuse.fs import DefuseFS
from defuse.nodes import Directory, FileProxy
root = Directory({
"proxied.rst": FileProxy(open("./README.rst", "rb"))
})
fs = DefuseFS("demofs", root)
if len(sys.argv) != 2:
print("usage: %s <mountpoint>" % sys.argv[0])
sys.exit(1)
fs.run(sys.argv[1], foreground=True)
import sys
import logging
from .fs import DefuseFS
from .nodes import Directory, FileProxy
root = Directory({
"proxied.rst": FileProxy(open("./README.rst", "rb"))
})
fs = DefuseFS("demofs", root)
if len(sys.argv) != 2:
print("usage: %s <mountpoint>" % sys.argv[0])
sys.exit(1)
logging.basicConfig(level=logging.DEBUG)
fs.run(sys.argv[1], foreground=True)
#!/usr/bin/env python
from __future__ import print_function, absolute_import, division
import logging
from collections import defaultdict
from errno import EROFS, ENOTSUP, EIO, ENOENT
from stat import S_IFDIR, S_IFLNK, S_IFREG
from sys import argv, exit
from os.path import split as split_path
from time import time
from fuse import FUSE, FuseOSError, Operations, LoggingMixIn
if not hasattr(__builtins__, "bytes"):
bytes = str
def or_raise(val):
if isinstance(val, Exception):
raise val
else:
return val
class DefuseFS:
"""
A simple `fuse.Operations`-like object that supports method delegation
to its contents.
"""
log = logging.getLogger("defusefs")
defaults = {
"access": 0,
"chmod": FuseOSError(EROFS),
"chown": FuseOSError(EROFS),
"create": FuseOSError(EROFS),
"destroy": None,
"flush": 0,
"fsync": 0,
"fsyncdir": 0,
"getattr": FuseOSError(ENOENT),
"getxattr": FuseOSError(ENOTSUP),
"init": None,
"link": FuseOSError(EROFS),
"mkdir": FuseOSError(EROFS),
"mknod": FuseOSError(EROFS),
"open": 0,
"opendir": 0,
"read": FuseOSError(EIO),
"readdir": [".", ".."],
"readlink": FuseOSError(ENOENT),
"release": 0,
"releasedir": 0,
"removexattr": FuseOSError(ENOTSUP),
"rename": FuseOSError(EROFS),
"rmdir": FuseOSError(EROFS),
"setxattr": FuseOSError(ENOTSUP),
"statfs": {},
"symlink": FuseOSError(EROFS),
"truncate": FuseOSError(EROFS),
"unlink": FuseOSError(EROFS),
"utimens": 0,
"write": FuseOSError(EROFS),
}
def __init__(self, name, root):
self.name = name
self.root = root
def __getattr__(self, name):
# HACK!
if name in self.defaults:
return lambda *a, **kw: self.defaults(name)
def _get_file(self, path):
segments = path.strip("/").split("/")
cur_node = self.root
for segment in segments:
if segment == "":
continue
try:
cur_node = cur_node[segment]
except KeyError:
raise FuseOSError(ENOENT)
return cur_node
def __call__(self, op, path, *args):
self.log.debug("-> %s %s %s", op, path, repr(args))
try:
if op not in self.defaults:
# Invalid operation
raise FuseOSError(EFAULT)
if op in ["link", "symlink"]:
# TODO
raise FuseOSError(EROFS)
elif op in ["create", "mkdir", "mknod"]:
_path, name = split_path(path)
node = self._get_file(_path)
if hasattr(node, op):
ret = getattr(node, op)(name, *args)
self.log.debug("<- %s %s %r", op, path, ret)
return ret
else:
self.log.debug("<- %s %s (unsupported)", op, path)
return or_raise(self.defaults[op])
else:
node = self._get_file(path)
if hasattr(node, op):
ret = getattr(node, op)(*args)
self.log.debug("<- %s %s %r", op, path, ret)
return ret
else:
self.log.debug("<- %s %s (unsupported)", op, path)
return or_raise(self.defaults[op])
except FuseOSError as e:
self.log.debug("<! %s %s %r", op, path, e)
raise
def run(self, mountpoint, **kwargs):
kwargs.setdefault("fsname", self.name)
# kwargs.setdefault("debug", True)
print("heeeeyyyy")
return FUSE(self, mountpoint, **kwargs)
from stat import S_IFDIR, S_IFREG
class Directory(dict):
_perms = 0o555
def getattr(self, fh=None):
return {
"st_mode": (S_IFDIR | self._perms),
}
def readdir(self, fh=None):
yield "."
yield ".."
yield from self.keys()
class FileProxy:
_perms = 0o444
def __init__(self, fp, size=2**20):
self.fp = fp
self.size = size
def getattr(self, fh=None):
return {
"st_mode": (S_IFREG | self._perms),
"st_size": self.size,
}
def read(self, size, offset, fh=None):
if self.fp.seekable():
self.fp.seek(offset)
return self.fp.read(size)
import os
import sys
from setuptools import setup
setup(
name="defuse",
version="0.0.0",
author="Ale",
author_email="hi@ale.rocks",
description="A thin abstraction layer on top of FUSE",
url="https://gitlab.com/iamale/defuse",
packages=["defuse"],
install_requires=[
"fusepy==2.0.4",
],
)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment