1
0

Ruff
Some checks failed
Test Python / Test (push) Failing after 22s

This commit is contained in:
Jon Michael Aanes 2024-09-26 23:08:37 +02:00
parent fe59477cfa
commit b2c32881df
Signed by: Jmaa
SSH Key Fingerprint: SHA256:Ab0GfHGCblESJx7JRE4fj4bFy/KRpeLhi41y4pF3sNA
5 changed files with 124 additions and 93 deletions

View File

@ -1,8 +1,8 @@
import secret_loader
import requests_cache
import tempfile
from .favro_client import FavroClient, SeqId, OrganizationId import requests_cache
import secret_loader
from .favro_client import FavroClient, OrganizationId
from .favro_fuse import start_favro_fuse from .favro_fuse import start_favro_fuse
@ -14,12 +14,15 @@ def main():
read_only = False read_only = False
#with tempfile.TemporaryDirectory(prefix='favro_sync_') as tmpdirname: # with tempfile.TemporaryDirectory(prefix='favro_sync_') as tmpdirname:
tmpdirname = './output' # TODO tmpdirname = './output' # TODO
if True: if True:
session = requests_cache.CachedSession(tmpdirname + '/http-cache.sqlite', expire_after=360) session = requests_cache.CachedSession(
tmpdirname + '/http-cache.sqlite', expire_after=360,
)
client = FavroClient(favro_org_id=OrganizationId(favro_org_id), client = FavroClient(
favro_org_id=OrganizationId(favro_org_id),
favro_username=favro_username, favro_username=favro_username,
favro_password=favro_password, favro_password=favro_password,
session=session, session=session,
@ -29,5 +32,6 @@ def main():
client.check_logged_in() client.check_logged_in()
start_favro_fuse(client) start_favro_fuse(client)
if __name__ == '__main__': if __name__ == '__main__':
main() main()

View File

@ -1,25 +1,28 @@
import requests
import dataclasses
import datetime
from typing import Any
import functools
from collections.abc import Iterator from collections.abc import Iterator
from logging import getLogger from logging import getLogger
from .favro_data_model import Card, CardId, SeqId, OrganizationId import requests
from .favro_data_model import Card, CardId, OrganizationId, SeqId
logger = getLogger(__name__) logger = getLogger(__name__)
# Endpoints # Endpoints
URL_API_ROOT = 'https://favro.com/api/v1' URL_API_ROOT = 'https://favro.com/api/v1'
URL_GET_ALL_CARDS = URL_API_ROOT+'/cards' URL_GET_ALL_CARDS = URL_API_ROOT + '/cards'
URL_UPDATE_CARD = URL_API_ROOT+'/cards/{card_id}' URL_UPDATE_CARD = URL_API_ROOT + '/cards/{card_id}'
class FavroClient: class FavroClient:
def __init__(
def __init__(self, *, favro_org_id: OrganizationId, favro_username: str, favro_password: str, self,
session: requests.Session | None = None, read_only=True): *,
favro_org_id: OrganizationId,
favro_username: str,
favro_password: str,
session: requests.Session | None = None,
read_only=True,
):
assert favro_org_id is not None assert favro_org_id is not None
assert favro_username is not None assert favro_username is not None
assert favro_password is not None assert favro_password is not None
@ -27,10 +30,12 @@ class FavroClient:
# Setup session # Setup session
self.session = session or requests.Session() self.session = session or requests.Session()
self.session.auth = (favro_username, favro_password) self.session.auth = (favro_username, favro_password)
self.session.headers.update({ self.session.headers.update(
{
'organizationId': favro_org_id.raw_id, 'organizationId': favro_org_id.raw_id,
'content-type': 'application/json', 'content-type': 'application/json',
}) },
)
self.read_only = read_only self.read_only = read_only
self.card_id_to_seq_id: dict[CardId, SeqId] = {} self.card_id_to_seq_id: dict[CardId, SeqId] = {}
@ -42,8 +47,9 @@ class FavroClient:
def get_todo_list_cards(self) -> Iterator[Card]: def get_todo_list_cards(self) -> Iterator[Card]:
yield from self.get_cards(todo_list=True) yield from self.get_cards(todo_list=True)
def get_cards(
def get_cards(self, *, seq_id: SeqId | None = None, todo_list=False) -> Iterator[Card]: self, *, seq_id: SeqId | None = None, todo_list=False,
) -> Iterator[Card]:
# Determine params for get_cards # Determine params for get_cards
request = self._get_cards_request(seq_id, todo_list) request = self._get_cards_request(seq_id, todo_list)
@ -60,14 +66,16 @@ class FavroClient:
yield card yield card
del entity_json del entity_json
def _get_cards_request(self, seq_id: SeqId | None = None, todo_list=False) -> requests.PreparedRequest: def _get_cards_request(
self, seq_id: SeqId | None = None, todo_list=False,
) -> requests.PreparedRequest:
params = {'descriptionFormat': 'markdown'} params = {'descriptionFormat': 'markdown'}
if seq_id is not None: if seq_id is not None:
params['cardSequentialId']= str(seq_id.raw_id) params['cardSequentialId'] = str(seq_id.raw_id)
if todo_list is True: if todo_list is True:
params['todoList'] = 'true' params['todoList'] = 'true'
request = requests.Request('GET', URL_GET_ALL_CARDS, params = params) request = requests.Request('GET', URL_GET_ALL_CARDS, params=params)
return self.session.prepare_request(request) return self.session.prepare_request(request)
def get_card(self, seq_id: SeqId) -> Card: def get_card(self, seq_id: SeqId) -> Card:
@ -76,20 +84,21 @@ class FavroClient:
def get_card_id(self, seq_id: SeqId) -> CardId: def get_card_id(self, seq_id: SeqId) -> CardId:
if card_id := self.seq_id_to_card_id[seq_id]: if card_id := self.seq_id_to_card_id[seq_id]:
return card_id return card_id
first_card = next(self.get_cards(seq_id = seq_id)) first_card = next(self.get_cards(seq_id=seq_id))
return first_card.card_id return first_card.card_id
def _invalidate_cache(self, card_id: CardId) -> None: def _invalidate_cache(self, card_id: CardId) -> None:
self.session.cache.delete(requests=[ self.session.cache.delete(
self._get_cards_request(seq_id=self.card_id_to_seq_id[card_id]) requests=[self._get_cards_request(seq_id=self.card_id_to_seq_id[card_id])],
]) )
def update_card_description(self, card_id: CardId, description: str) -> Card: def update_card_description(self, card_id: CardId, description: str) -> Card:
"""Returns updated Card.""" """Returns updated Card."""
if self.read_only == 'silent': if self.read_only == 'silent':
logger.warning('FavroClient is silent read only: Discarding card description update of length %d', logger.warning(
len(description)) 'FavroClient is silent read only: Discarding card description update of length %d',
len(description),
)
return None # TODO return None # TODO
elif self.read_only is True: elif self.read_only is True:
raise Exception('FavroClient is read only') raise Exception('FavroClient is read only')
@ -100,11 +109,11 @@ class FavroClient:
} }
logger.warning('Sending body: %s', json_body) logger.warning('Sending body: %s', json_body)
response = self.session.put(URL_UPDATE_CARD.format(card_id=card_id.raw_id), json=json_body) response = self.session.put(
URL_UPDATE_CARD.format(card_id=card_id.raw_id), json=json_body,
)
response.raise_for_status() response.raise_for_status()
logger.warning("Response: %s", response.json()) logger.warning('Response: %s', response.json())
self._invalidate_cache(card_id) self._invalidate_cache(card_id)
return Card.from_json(response.json()) return Card.from_json(response.json())

View File

@ -1,32 +1,33 @@
import requests
import dataclasses import dataclasses
import datetime import datetime
from typing import Any from typing import Any
import functools
from collections.abc import Iterator
from logging import getLogger
@dataclasses.dataclass(frozen=True) @dataclasses.dataclass(frozen=True)
class SeqId: class SeqId:
raw_id: int raw_id: int
@dataclasses.dataclass(frozen=True) @dataclasses.dataclass(frozen=True)
class CardId: class CardId:
raw_id: str raw_id: str
@dataclasses.dataclass(frozen=True) @dataclasses.dataclass(frozen=True)
class CommonId: class CommonId:
raw_id: str raw_id: str
@dataclasses.dataclass(frozen=True) @dataclasses.dataclass(frozen=True)
class UserId: class UserId:
raw_id: str raw_id: str
@dataclasses.dataclass(frozen=True) @dataclasses.dataclass(frozen=True)
class OrganizationId: class OrganizationId:
raw_id: str raw_id: str
@dataclasses.dataclass(frozen=True) @dataclasses.dataclass(frozen=True)
class Card: class Card:
card_id: CardId card_id: CardId
@ -44,7 +45,7 @@ class Card:
detailed_description: str | None detailed_description: str | None
''' TODO, fieds: """ TODO, fieds:
'position': -399 'position': -399
'listPosition': -399 'listPosition': -399
@ -57,24 +58,24 @@ class Card:
'timeOnBoard': None 'timeOnBoard': None
'timeOnColumns': None 'timeOnColumns': None
'favroAttachments': [] 'favroAttachments': []
''' """
@staticmethod @staticmethod
def from_json(json: dict[str, Any]) -> 'Card': def from_json(json: dict[str, Any]) -> 'Card':
return Card( return Card(
card_id = CardId(json['cardId']), card_id=CardId(json['cardId']),
seq_id = SeqId(json['sequentialId']), seq_id=SeqId(json['sequentialId']),
common_id = CommonId(json['cardCommonId']), common_id=CommonId(json['cardCommonId']),
detailed_description = json.get('detailedDescription'), detailed_description=json.get('detailedDescription'),
is_archived = json['archived'], is_archived=json['archived'],
organization_id = OrganizationId(json['organizationId']), organization_id=OrganizationId(json['organizationId']),
name = json['name'], name=json['name'],
todo_list_user_id = UserId(json['todoListUserId' ]) if 'todoListUserId' in json else None, todo_list_user_id=UserId(json['todoListUserId'])
todo_list_completed = json.get('todoListCompleted'), if 'todoListUserId' in json
dependencies = json['dependencies'], else None,
tags = json['tags'], todo_list_completed=json.get('todoListCompleted'),
creator_user_id = UserId(json['createdByUserId']), dependencies=json['dependencies'],
creation_date = datetime.datetime.fromisoformat(json['createdAt']), tags=json['tags'],
creator_user_id=UserId(json['createdByUserId']),
creation_date=datetime.datetime.fromisoformat(json['createdAt']),
) )

View File

@ -1,17 +1,20 @@
import os, stat, errno, fuse
from pathlib import Path
import re
import dataclasses import dataclasses
import errno
import re
import stat
from collections.abc import Iterator from collections.abc import Iterator
from .favro_data_model import CardId, SeqId, Card import fuse
from .favro_client import FavroClient from .favro_client import FavroClient
from .favro_data_model import Card, SeqId
fuse.fuse_python_api = (0, 2) fuse.fuse_python_api = (0, 2)
hello_path = '/hello' hello_path = '/hello'
hello_str = b'Hello World!\n' hello_str = b'Hello World!\n'
class MyStat(fuse.Stat): class MyStat(fuse.Stat):
def __init__(self): def __init__(self):
self.st_mode = 0 self.st_mode = 0
@ -25,12 +28,13 @@ class MyStat(fuse.Stat):
self.st_mtime = 0 self.st_mtime = 0
self.st_ctime = 0 self.st_ctime = 0
CARD_FILENAME_FORMAT = 'PAR-{seq_id}.md' CARD_FILENAME_FORMAT = 'PAR-{seq_id}.md'
CARD_FILENAME_REGEX = r'^\/PAR\-(\d+)\.md$' CARD_FILENAME_REGEX = r'^\/PAR\-(\d+)\.md$'
@dataclasses.dataclass(frozen=True) @dataclasses.dataclass(frozen=True)
class Thing: class Thing:
@staticmethod @staticmethod
def from_path(path_str: str) -> 'Thing | None': def from_path(path_str: str) -> 'Thing | None':
if path_str == '/': if path_str == '/':
@ -39,25 +43,28 @@ class Thing:
return CardThing(SeqId(int(m.group(1)))) return CardThing(SeqId(int(m.group(1))))
return None return None
@dataclasses.dataclass(frozen=True) @dataclasses.dataclass(frozen=True)
class RootThing(Thing): class RootThing(Thing):
pass pass
@dataclasses.dataclass(frozen=True) @dataclasses.dataclass(frozen=True)
class CardThing(Thing): class CardThing(Thing):
seq_id: SeqId seq_id: SeqId
def card_to_contents(card: Card) -> str: def card_to_contents(card: Card) -> str:
ls = [] ls = []
#ls.append('# ') # ls.append('# ')
#ls.append(card.name) # ls.append(card.name)
#ls.append('\n\n') # ls.append('\n\n')
ls.append(card.detailed_description or '') ls.append(card.detailed_description or '')
return ''.join(ls) return ''.join(ls)
class FavroFuse(fuse.Fuse): class FavroFuse(fuse.Fuse):
'''Favro Filesystem in Userspace. """Favro Filesystem in Userspace."""
'''
def __init__(self, favro_client: FavroClient, **kwargs): def __init__(self, favro_client: FavroClient, **kwargs):
self.favro_client = favro_client self.favro_client = favro_client
@ -93,8 +100,8 @@ class FavroFuse(fuse.Fuse):
thing = Thing.from_path(path) thing = Thing.from_path(path)
if not isinstance(thing, CardThing): if not isinstance(thing, CardThing):
return -errno.ENOENT return -errno.ENOENT
#accmode = os.O_RDONLY | os.O_WRONLY | os.O_RDWR # accmode = os.O_RDONLY | os.O_WRONLY | os.O_RDWR
#if (flags & accmode) != os.O_RDONLY: return -errno.EACCES # if (flags & accmode) != os.O_RDONLY: return -errno.EACCES
return None return None
def read(self, path: str, size: int, offset: int) -> bytes | int: def read(self, path: str, size: int, offset: int) -> bytes | int:
@ -112,7 +119,7 @@ class FavroFuse(fuse.Fuse):
if offset < slen: if offset < slen:
if offset + size > slen: if offset + size > slen:
size = slen - offset size = slen - offset
buf = contents[offset:offset+size] buf = contents[offset : offset + size]
else: else:
buf = b'' buf = b''
return buf return buf
@ -159,20 +166,31 @@ class FavroFuse(fuse.Fuse):
# Return amount written # Return amount written
return 0 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=""" 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 Userspace hello example
""" + fuse.Fuse.fusage """
+ fuse.Fuse.fusage
)
def start_favro_fuse(favro_client: FavroClient): def start_favro_fuse(favro_client: FavroClient):
# TODO: # TODO:
server = FavroFuse( server = FavroFuse(
favro_client = favro_client, favro_client=favro_client,
version='%prog ' + fuse.__version__, version='%prog ' + fuse.__version__,
usage=HELP, usage=HELP,
dash_s_do='setsingle') dash_s_do='setsingle',
)
server.parse(errex=1) server.parse(errex=1)
server.main() server.main()

View File

@ -38,8 +38,7 @@ REQUIREMENTS_MAIN = [
'requests', 'requests',
'secret_loader @ git+https://gitfub.space/Jmaa/secret_loader', 'secret_loader @ git+https://gitfub.space/Jmaa/secret_loader',
] ]
REQUIREMENTS_TEST = [ REQUIREMENTS_TEST = []
]
setup( setup(