1
0

Working on FUSE

This commit is contained in:
Jon Michael Aanes 2024-09-26 18:49:28 +02:00
parent 109789fe78
commit 8c3c0eda45
Signed by: Jmaa
SSH Key Fingerprint: SHA256:Ab0GfHGCblESJx7JRE4fj4bFy/KRpeLhi41y4pF3sNA
4 changed files with 149 additions and 42 deletions

View File

@ -3,5 +3,7 @@
Filesystem in User Space for Favro.
Synchronize your local notes and your Favro.
"""
Uses the [Favro API](https://favro.com/developer/). Rate limiting depends upon
your organization's payment plan.
"""

View File

@ -1,35 +1,22 @@
import secret_loader
from .favro_client import FavroClient
from .favro_client import FavroClient, SeqId
from .favro_fuse import start_favro_fuse
# Authentication
description = '''
I cannot remember what this card involved
# Tasks
- [ ] Why doesn't task work?
- [ ] Task 2
# Quotes
I love this card.
'''
def main(card_id: CardId):
def main():
secrets = secret_loader.SecretLoader()
FAVRO_ORG_ID = secrets.load_or_fail('FAVRO_ORGANIZATION_ID')
FAVRO_USERNAME = secrets.load_or_fail('FAVRO_USERNAME')
FAVRO_PASSWORD = secrets.load_or_fail('FAVRO_PASSWORD')
favro_org_id = secrets.load_or_fail('FAVRO_ORGANIZATION_ID')
favro_username = secrets.load_or_fail('FAVRO_USERNAME')
favro_password = secrets.load_or_fail('FAVRO_PASSWORD')
client = FavroClient(FAVRO_ORG_ID, FAVRO_USERNAME, FAVRO_PASSWORD)
card_id = client.get_card_id(SeqId(4714))
client.update_card_description(card_id, description)
client = FavroClient(favro_org_id=favro_org_id, favro_username=favro_username, favro_password=favro_password)
#card_id = client.get_card_id(SeqId(4714))
#description = 'TEST'
#client.update_card_description(card_id, description)
start_favro_fuse(client)
if __name__ == '__main__':
main()

View File

@ -2,6 +2,9 @@ import requests
import dataclasses
from typing import Any
import functools
from collections.abc import Iterator
PREFIX = 'PAR-' # TODO: Make configurable
# Types
@ -13,6 +16,23 @@ class SeqId:
class CardId:
raw_id: str
@dataclasses.dataclass(frozen=True)
class Card:
card_id: CardId
seq_id: SeqId
seq_id_with_prefix: str
detailed_description: str
@staticmethod
def from_json(json: dict[str, Any]) -> 'Card':
print(json)
return Card(
card_id = CardId(json['cardId']),
seq_id = SeqId(json['cardSequentialId']),
seq_id_with_prefix = PREFIX + json['cardSequentialId'],
detailed_description = json['detailedDescription'],
)
# Endpoints
URL_API_ROOT = 'https://favro.com/api/v1'
URL_GET_ALL_CARDS = URL_API_ROOT+'/cards'
@ -20,7 +40,7 @@ URL_UPDATE_CARD = URL_API_ROOT+'/cards/{card_id}'
class FavroClient:
def __init__(*, favro_org_id: str, favro_username: str, favro_password: str,
def __init__(self, *, favro_org_id: str, favro_username: str, favro_password: str,
session: requests.Session | None = None):
# Setup session
self.session = session or requests.Session()
@ -30,29 +50,44 @@ class FavroClient:
'content-type': 'application/json',
})
def get_todo_list_cards(self) -> Iterator[Card]:
yield from self.get_cards(todo_list=True)
def get_card_json(seqid: SeqId) -> dict[str, Any]:
# TODO: Return structured?
params = {'cardSequentialId': seqid.raw_id}
def get_cards(self, *, seqid: SeqId | None = None, todo_list=True) -> Iterator[Card]:
# Determine params for get_cards
params = {}
if seqid:
params['cardSequentialId']= seqid.raw_id
if todo_list:
params['todoList'] = True
# Run query
response = self.session.get(URL_GET_ALL_CARDS, params = params)
response.raise_for_status()
json = response.json()
assert json['pages'] == 1
assert len(json['entities']) == 1
return json['entities'][0]
# TODO: Pageination
for entity_json in json['entities']:
yield Card.from_json(entity_json)
del entity_json
def get_card(self, seqid: SeqId) -> Card:
return next(self.get_cards(seqid=seqid))
@functools.cache
def get_card_id(seqid: SeqId) -> CardId:
json = get_card_json(seqid)
def get_card_id(self, seqid: SeqId) -> CardId:
json = self.get_card_json(seqid)
return CardId(json['cardId'])
def update_card_description(card_id: CardId, description: str):
json_body = {
'detailedDescription': description,
'descriptionFormat': 'markdown',
}
def update_card_description(self, card_id: CardId, description: str) -> Card:
"""Returns updated Card."""
json_body = {
'detailedDescription': description,
'descriptionFormat': 'markdown',
}
response = SESSION.put(URL_UPDATE_CARD.format(card_id=card_id.raw_id), json=json_body)
response.raise_for_status()
# TODO: Return updated?
response = self.session.put(URL_UPDATE_CARD.format(card_id=card_id.raw_id), json=json_body)
response.raise_for_status()
return Card.from_json(response.json())

83
favro_sync/favro_fuse.py Normal file
View File

@ -0,0 +1,83 @@
import os, stat, errno, fuse
from pathlib import Path
from collections.abc import Iterator
from .favro_client import FavroClient
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
class FavroFuse(fuse.Fuse):
def __init__(self, favro_client: FavroClient, **kwargs):
self.favro_client = favro_client
super().__init__(**kwargs)
def getattr(self, path: str) -> MyStat | int:
st = MyStat()
if path == '/':
st.st_mode = stat.S_IFDIR | 0o755
st.st_nlink = 2
elif path == hello_path:
st.st_mode = stat.S_IFREG | 0o444
st.st_nlink = 1
st.st_size = len(hello_str)
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.seqid_with_prefix)
def open(self, path: str, flags) -> int:
if path != hello_path:
return -errno.ENOENT
accmode = os.O_RDONLY | os.O_WRONLY | os.O_RDWR
if (flags & accmode) != os.O_RDONLY:
return -errno.EACCES
def read(self, path: str, size: int, offset: int) -> bytes | int:
if path != hello_path:
return -errno.ENOENT
slen = len(hello_str)
if offset < slen:
if offset + size > slen:
size = slen - offset
buf = hello_str[offset:offset+size]
else:
buf = b''
return buf
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()