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 | None: 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 return None 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) str_value = to_custom_field_value(field_assignment, field_def) if str_value is not None: custom_fields[field_def.name] = str_value del field_assignment, str_value return custom_fields def to_card_contents(card: Card, favro_client: FavroClient) -> CardContents: 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() def __str__(self): return '/' @dataclasses.dataclass(frozen=True) class CollectionFileSystemItem(FileSystemItem): collection_name: str def __post_init__(self): assert '/' not in self.collection_name @staticmethod def from_path_segment(segment: str) -> 'CollectionFileSystemItem': return CollectionFileSystemItem(segment) def __str__(self): return self.collection_name @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 __str__(self): return CARD_FILENAME_FORMAT.format(seq_id=self.seq_id.raw_id) 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, CollectionFileSystemItem, 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 | CollectionFileSystemItem): 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]: file_system_item = path_to_file_system_item(path, self.path_components) yield fuse.Direntry('.') yield fuse.Direntry('..') if isinstance(file_system_item, RootFileSystemItem): for collection in self.favro_client.get_collections(): collection_name = collection.name.replace('/', '') yield fuse.Direntry(str(CollectionFileSystemItem(collection_name))) del collection elif isinstance(file_system_item, CollectionFileSystemItem): # TODO: move into own function for collection in self.favro_client.get_collections(): if collection.name.replace('/', '') == file_system_item.collection_name: collection_id = collection.collection_id del collection for card in self.favro_client.get_cards(collection_id=collection_id): yield fuse.Direntry(str(CardFileSystemItem(card.seq_id))) del card 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()