import dataclasses import errno import re import stat from collections.abc import Iterator from logging import getLogger import fuse from .favro_client import FavroClient from .favro_data_model import Card, SeqId from .favro_markdown import CardContents, CardFileFormatter logger = getLogger(__name__) fuse.fuse_python_api = (0, 2) class FavroStat(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_IDENTIFIER_FORMAT = 'PAR-{seq_id}' CARD_FILENAME_FORMAT = CARD_IDENTIFIER_FORMAT + '.md' CARD_FILENAME_REGEX = r'^\/PAR\-(\d+)\.md$' OFFICIAL_URL = 'https://favro.com/organization/{org_id}?card=par-{seq_id}' @dataclasses.dataclass(frozen=True) class FileSystemItem: @staticmethod def from_path(path_str: str) -> 'FileSystemItem | None': if path_str == '/': return RootFileSystemItem() if m := re.match(CARD_FILENAME_REGEX, path_str): return CardFileSystemItem(SeqId(int(m.group(1)))) return None @dataclasses.dataclass(frozen=True) class RootFileSystemItem(FileSystemItem): pass @dataclasses.dataclass(frozen=True) class CardFileSystemItem(FileSystemItem): seq_id: SeqId class FavroFuse(fuse.Fuse): """Favro FileSystem in Userspace.""" def __init__( self, favro_client: FavroClient, formatter: CardFileFormatter, **kwargs, ): self.favro_client = favro_client self.formatter = formatter self.wiped_cards = set() super().__init__(**kwargs) def getattr(self, path: str) -> FavroStat | int: thing = FileSystemItem.from_path(path) st = FavroStat() if isinstance(thing, RootFileSystemItem): st.st_mode = stat.S_IFDIR | 0o755 st.st_nlink = 2 elif isinstance(thing, CardFileSystemItem): card = self.favro_client.get_card(thing.seq_id) st.st_mode = stat.S_IFREG | 0o666 st.st_nlink = 1 st.st_size = len(self._format_card_file(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]: logger.warning('readdir(path=%s, offset=%s)', path, offset) 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) -> int | None: thing = FileSystemItem.from_path(path) if not isinstance(thing, CardFileSystemItem): return -errno.ENOENT return None def read(self, path: str, size: int, offset: int) -> bytes | int: # Check that this is a card thing. thing = FileSystemItem.from_path(path) if not isinstance(thing, CardFileSystemItem): return -errno.ENOENT card = self.favro_client.get_card(thing.seq_id) contents_str = self._format_card_file(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 = FileSystemItem.from_path(path) if not isinstance(thing, CardFileSystemItem): return -errno.ENOENT card = self.favro_client.get_card(thing.seq_id) # Splice contents contents_str = self._format_card_file(card) contents = bytes(contents_str, 'utf8') contents = splice(contents, written_buffer, offset) contents_str = contents.decode('utf8') # Write to favro card_updated = self.formatter.parse_card_contents(contents_str) self.favro_client.update_card_contents(card.card_id, card_updated) self.wiped_cards.remove(thing.seq_id) # Return amount written return len(written_buffer) def truncate(self, path: str, new_size: int): # Check that this is a card thing. thing = FileSystemItem.from_path(path) if not isinstance(thing, CardFileSystemItem): return -errno.ENOENT card = self.favro_client.get_card(thing.seq_id) # Splice contents contents_str = self._format_card_file(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 card_updated = self.formatter.parse_card_contents(contents_str) self.favro_client.update_card_contents_locally(card.card_id, card_updated) self.wiped_cards.add(thing.seq_id) # Return amount written return 0 def _format_card_file(self, card: Card) -> str: if card.seq_id in self.wiped_cards: return '' tags = [self.favro_client.get_tag(tag_id).name for tag_id in card.tags] assignments = [ self.favro_client.get_user(assignment.user).name for assignment in card.assignments ] dependencies = [ CARD_FILENAME_FORMAT.format( seq_id=self.favro_client.get_card_by_card_id(dep.card_id).seq_id.raw_id, ) for dep in card.dependencies if dep.is_before ] card_contents = CardContents( CARD_IDENTIFIER_FORMAT.format(seq_id=card.seq_id.raw_id), card.name, card.detailed_description, tags, assignments, dependencies, url=OFFICIAL_URL.format( org_id=card.organization_id.raw_id, seq_id=card.seq_id.raw_id, ), todo_list_completed = card.todo_list_completed, ) return self.formatter.format_card_contents(card_contents) 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, formatter=CardFileFormatter(), version='%prog ' + fuse.__version__, usage=HELP, dash_s_do='setsingle', ) server.parse(errex=1) server.main()