favro-sync/favro_sync/favro_fuse.py

import os, stat, errno, fuse
from pathlib import Path
import re
import dataclasses
from collections.abc import Iterator

from .favro_client import FavroClient, CardId, SeqId, Card

fuse.fuse_python_api = (0, 2)

hello_path = '/hello'
hello_str = b'Hello World!\n'

class MyStat(fuse.Stat):
    def __init__(self):
        self.st_mode = 0
        self.st_ino = 0
        self.st_dev = 0
        self.st_nlink = 0
        self.st_uid = 0
        self.st_gid = 0
        self.st_size = 0
        self.st_atime = 0
        self.st_mtime = 0
        self.st_ctime = 0

CARD_FILENAME_FORMAT = 'PAR-{seq_id}.md'
CARD_FILENAME_REGEX = r'^\/PAR\-(\d+)\.md$'

@dataclasses.dataclass(frozen=True)
class Thing:

    @staticmethod
    def from_path(path_str: str) -> 'Thing | None':
        if path_str == '/':
            return RootThing()
        if m := re.match(CARD_FILENAME_REGEX, path_str):
            return CardThing(SeqId(int(m.group(1))))
        return None

@dataclasses.dataclass(frozen=True)
class RootThing(Thing):
    pass

@dataclasses.dataclass(frozen=True)
class CardThing(Thing):
    seq_id: SeqId

def card_to_contents(card: Card) -> str:
    ls = []
    #ls.append('# ')
    #ls.append(card.name)
    #ls.append('\n\n')
    ls.append(card.detailed_description or '')
    return ''.join(ls)

class FavroFuse(fuse.Fuse):
    '''Favro Filesystem in Userspace.
    '''

    def __init__(self, favro_client: FavroClient, **kwargs):
        self.favro_client = favro_client
        super().__init__(**kwargs)

    def getattr(self, path: str) -> MyStat | int:
        thing = Thing.from_path(path)

        st = MyStat()
        if isinstance(thing, RootThing):
            st.st_mode = stat.S_IFDIR | 0o755
            st.st_nlink = 2
        elif isinstance(thing, CardThing):
            card = self.favro_client.get_card(thing.seq_id)

            st.st_mode = stat.S_IFREG | 0o666
            st.st_nlink = 1
            st.st_size = len(card_to_contents(card))
            st.st_ctime = int(card.creation_date.timestamp())
            st.st_mtime = st.st_ctime # TODO
        else:
            return -errno.ENOENT
        return st

    def readdir(self, path: str, offset: int) -> Iterator[fuse.Direntry]:
        yield fuse.Direntry('.')
        yield fuse.Direntry('..')

        for card in self.favro_client.get_todo_list_cards():
            yield fuse.Direntry(CARD_FILENAME_FORMAT.format(seq_id=card.seq_id.raw_id))

    def open(self, path: str, flags) -> int | None:
        thing = Thing.from_path(path)
        if not isinstance(thing, CardThing):
            return -errno.ENOENT
        #accmode = os.O_RDONLY | os.O_WRONLY | os.O_RDWR
        #if (flags & accmode) != os.O_RDONLY: return -errno.EACCES
        return None

    def read(self, path: str, size: int, offset: int) -> bytes | int:
        # Check that this is a card thing.
        thing = Thing.from_path(path)
        if not isinstance(thing, CardThing):
            return -errno.ENOENT

        card = self.favro_client.get_card(thing.seq_id)

        contents_str = card_to_contents(card)
        contents = bytes(contents_str, 'utf8')

        slen = len(contents)
        if offset < slen:
            if offset + size > slen:
                size = slen - offset
            buf = contents[offset:offset+size]
        else:
            buf = b''
        return buf

    def write(self, path: str, written_buffer: bytes, offset: int) -> int:
        # Check that this is a card thing.
        thing = Thing.from_path(path)
        if not isinstance(thing, CardThing):
            return -errno.ENOENT

        card = self.favro_client.get_card(thing.seq_id)

        # Splice contents
        contents_str = card_to_contents(card)
        contents = bytes(contents_str, 'utf8')
        contents = splice(contents, written_buffer, offset)
        contents_str = contents.decode('utf8')

        # Write to favro
        self.favro_client.update_card_description(card.card_id, contents_str)

        # Return amount written
        return len(written_buffer)

    def truncate(self, path: str, new_size: int):
        # Check that this is a card thing.
        thing = Thing.from_path(path)
        if not isinstance(thing, CardThing):
            return -errno.ENOENT

        card = self.favro_client.get_card(thing.seq_id)

        # Splice contents
        contents_str = card_to_contents(card)
        contents = bytes(contents_str, 'utf8')
        old_size = len(contents)
        contents = contents[0:new_size] + b' ' * (old_size - new_size)
        assert len(contents) == old_size
        contents_str = contents.decode('utf8')

        # Write to favro
        self.favro_client.update_card_description(card.card_id, contents_str)

        # Return amount written
        return 0

def splice(original_buffer: bytes, input_buffer: bytes, offset: int) -> bytes:
    return original_buffer[0:offset-1] + input_buffer + original_buffer[offset+len(input_buffer)+1:len(original_buffer)]

HELP="""
Userspace hello example

""" + fuse.Fuse.fusage

def start_favro_fuse(favro_client: FavroClient):
    # TODO:
    server = FavroFuse(
            favro_client = favro_client,
            version='%prog ' + fuse.__version__,
                     usage=HELP,
                     dash_s_do='setsingle')
    server.parse(errex=1)
    server.main()