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, CustomFieldInfo, CustomFieldItemId, CustomField from .favro_markdown import CardContents, CardFileFormatter ################################################################################ # Constants logger = getLogger(__name__) fuse.fuse_python_api = (0, 2) OFFICIAL_URL = 'https://favro.com/organization/{org_id}?card=par-{seq_id}' CARD_IDENTIFIER_FORMAT = 'PAR-{seq_id}' CARD_FILENAME_FORMAT = CARD_IDENTIFIER_FORMAT + '.md' CARD_FILENAME_REGEX = r'^PAR\-(\d+)\.md$' ################################################################################ # Formatting def to_custom_field_value(custom_field: CustomField, field_def: CustomFieldInfo) -> str: value: CustomFieldItemId | list[CustomFieldItemId] = custom_field.value if field_def.type in {'Single select','Multiple select'}: items = [field_def.get_field_item(item_id) for item_id in value] items = [i for i in items if i] return items[0].name if field_def.type in {'Color'}: return custom_field.color assert False, 'Unknown type: ' + field_def.type def to_custom_fields(card: Card, favro_client: FavroClient) -> dict[str,str]: custom_fields = {} for field_assignment in card.custom_fields: field_def = favro_client.get_custom_field(field_assignment.custom_field_id) custom_fields[field_def.name] = to_custom_field_value(field_assignment, field_def) del field_assignment return custom_fields def to_card_contents(card: Card, favro_client: FavroClient) -> str: tags = [favro_client.get_tag(tag_id).name for tag_id in card.tags] assignments = [ favro_client.get_user(assignment.user).name for assignment in card.assignments ] dependencies = [ CARD_FILENAME_FORMAT.format( seq_id=favro_client.get_card_by_card_id(dep.card_id).seq_id.raw_id, ) for dep in card.dependencies if dep.is_before ] return 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, is_archived=card.is_archived, start_date=card.start_date, due_date=card.due_date, custom_fields=to_custom_fields(card, favro_client), ) ################################################################################ # FUSE 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 @dataclasses.dataclass(frozen=True) class FileSystemItem: pass @dataclasses.dataclass(frozen=True) class RootFileSystemItem(FileSystemItem): @staticmethod def from_path_segment(segment: str) -> 'RootFileSystemItem': return RootFileSystemItem() @dataclasses.dataclass(frozen=True) class OrganizationFileSystemItem(FileSystemItem): name: str @staticmethod def from_path_segment(segment: str) -> 'OrganizationFileSystemItem': return OrganizationFileSystemItem(segment) @dataclasses.dataclass(frozen=True) class CardFileSystemItem(FileSystemItem): seq_id: SeqId @staticmethod def from_path_segment(segment: str) -> 'CardFileSystemItem | None': if m := re.match(CARD_FILENAME_REGEX, segment): return CardFileSystemItem(SeqId(int(m.group(1)))) return None def path_to_file_system_item(path_str: str, path_components: list[type[FileSystemItem]]) -> FileSystemItem | None: path = re.findall(r'[^/]+', path_str) component = path_components[len(path)] return component.from_path_segment(path[-1] if path else None) 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() #self.path_components = [RootFileSystemItem, OrganizationFileSystemItem, CardFileSystemItem] self.path_components = [RootFileSystemItem, CardFileSystemItem] super().__init__(**kwargs) def getattr(self, path: str) -> FavroStat | int: file_system_item = path_to_file_system_item(path, self.path_components) st = FavroStat() if isinstance(file_system_item, RootFileSystemItem): st.st_mode = stat.S_IFDIR | 0o755 st.st_nlink = 2 elif isinstance(file_system_item, CardFileSystemItem): card = self.favro_client.get_card(file_system_item.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: file_system_item = path_to_file_system_item(path, self.path_components) if not isinstance(file_system_item, CardFileSystemItem): return -errno.ENOENT return None def read(self, path: str, size: int, offset: int) -> bytes | int: # Check that this is a card file_system_item. file_system_item = path_to_file_system_item(path, self.path_components) if not isinstance(file_system_item, CardFileSystemItem): return -errno.ENOENT card = self.favro_client.get_card(file_system_item.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 file_system_item. file_system_item = path_to_file_system_item(path, self.path_components) if not isinstance(file_system_item, CardFileSystemItem): return -errno.ENOENT card = self.favro_client.get_card(file_system_item.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(file_system_item.seq_id) # Return amount written return len(written_buffer) def truncate(self, path: str, new_size: int): # Check that this is a card file_system_item. file_system_item = path_to_file_system_item(path, self.path_components) if not isinstance(file_system_item, CardFileSystemItem): return -errno.ENOENT card = self.favro_client.get_card(file_system_item.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(file_system_item.seq_id) # Return amount written return 0 def _format_card_file(self, card: Card) -> str: if card.seq_id in self.wiped_cards: return '' card_contents = to_card_contents(card, self.favro_client) 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()