DataLayer Virtual FileSystem

gistfile1.txt

#!/usr/bin/env python3

import requests
import fuse
from fuse import FUSE, FuseOSError, Operations

class ChiaDataLayerFileSystem(Operations):
    def __init__(self, rpc_base_url):
        self.rpc_base_url = rpc_base_url

    def get_subscriptions(self):
        url = f"{self.rpc_base_url}/subscriptions"
        headers = {"Content-Type": "application/json"}

        response = requests.post(url, headers=headers, verify=False)
        if response.status_code == 200 and response.json().get("success"):
            return response.json()["store_ids"]
        else:
            raise FuseOSError(fuse.errno.ECOMM)

    def get_keys(self, store_id):
        url = f"{self.rpc_base_url}/get_keys"
        headers = {"Content-Type": "application/json"}
        body = {"id": store_id}

        response = requests.post(url, json=body, headers=headers, verify=False)
        if response.status_code == 200 and response.json().get("success"):
            return [bytes.fromhex(key[2:]).decode('utf-8') for key in response.json()["keys"]]
        else:
            raise FuseOSError(fuse.errno.ECOMM)

    def readdir(self, path, fh):
        if path == "/":
            try:
                store_ids = self.get_subscriptions()
                yield from ['.', '..'] + store_ids
            except FuseOSError:
                pass
        else:
            store_id = path.strip("/")
            try:
                files = self.get_keys(store_id)
                yield from ['.', '..'] + files
            except FuseOSError:
                pass

    def open(self, path, flags):
        store_id, file_path = path.strip("/").split('/', 1)
        files = self.get_keys(store_id)
        if file_path in files:
            return 0  # Dummy file handle
        else:
            raise FuseOSError(fuse.errno.ENOENT)  # No such file

    def read(self, path, size, offset, fh):
        store_id, file_path = path.strip("/").split('/', 1)
        url = f"{self.rpc_base_url}/get_value"
        headers = {"Content-Type": "application/json"}
        body = {
            "id": store_id,
            "key": file_path.encode().hex()
        }

        response = requests.post(url, json=body, headers=headers, verify=False)
        if response.status_code == 200 and response.json().get("success"):
            file_content = bytes.fromhex(response.json()["value"]).decode('utf-8')
            return file_content[offset:offset + size]
        else:
            raise FuseOSError(fuse.errno.EIO)
            
     def getattr(self, path, fh=None):
        if path == '/':
            st_mode = (stat.S_IFDIR | 0o755)
            st_nlink = 2
        else:
            st_mode = (stat.S_IFREG | 0o444)
            st_nlink = 1
            st_size = 1024  # Arbitrary size; modify as needed
        return {
            'st_atime': 0,
            'st_ctime': 0,
            'st_gid': os.getgid(),
            'st_mode': st_mode,
            'st_mtime': 0,
            'st_nlink': st_nlink,
            'st_size': st_size,
            'st_uid': os.getuid()
        }

    # Add other necessary methods like getattr, etc...

if __name__ == '__main__':
    rpc_base_url = "https://localhost:8562"
    fuse = FUSE(ChiaDataLayerFileSystem(rpc_base_url), '/mnt/chia', foreground=True, nothreads=True)