1
0
favro-sync/favro_sync/favro_client.py

256 lines
8.1 KiB
Python
Raw Normal View History

2024-09-28 10:57:27 +00:00
"""Favro API Client.
Implements methods for interacting with the [Favro API](https://favro.com/developer/).
"""
2024-09-28 12:13:51 +00:00
import dataclasses
2024-09-26 16:49:28 +00:00
from collections.abc import Iterator
from logging import getLogger
2024-09-26 16:49:28 +00:00
2024-09-26 21:08:37 +00:00
import requests
from .favro_data_model import (
Card,
CardId,
2024-10-02 14:23:15 +00:00
Collection,
2024-10-01 14:15:03 +00:00
CustomFieldId,
CustomFieldInfo,
OrganizationId,
2024-10-02 15:17:14 +00:00
CollectionId,
SeqId,
TagId,
TagInfo,
UserId,
UserInfo,
)
2024-09-27 14:13:03 +00:00
from .favro_markdown import CardContents
2024-09-26 17:16:53 +00:00
2024-09-26 21:07:46 +00:00
logger = getLogger(__name__)
2024-09-26 16:49:28 +00:00
2024-09-26 14:43:30 +00:00
# Endpoints
URL_API_ROOT = 'https://favro.com/api/v1'
2024-10-02 14:23:15 +00:00
URL_GET_CARDS = URL_API_ROOT + '/cards'
2024-09-26 21:08:37 +00:00
URL_UPDATE_CARD = URL_API_ROOT + '/cards/{card_id}'
URL_GET_USER = URL_API_ROOT + '/users/{user_id}'
URL_GET_TAG = URL_API_ROOT + '/tags/{tag_id}'
2024-10-01 14:06:06 +00:00
URL_GET_CUSTOM_FIELD = URL_API_ROOT + '/customfields/{custom_field_id}'
2024-10-01 14:40:44 +00:00
URL_GET_TASKS = URL_API_ROOT + '/tasks'
2024-10-02 14:23:15 +00:00
URL_GET_COLLECTIONS = URL_API_ROOT + '/collections'
2024-09-27 09:06:06 +00:00
2024-10-02 08:32:14 +00:00
class CardCache:
2024-09-27 09:06:06 +00:00
def __init__(self):
self.cards = []
def add_card(self, card: Card) -> None:
self.remove(card.card_id)
self.cards.append(card)
def get_card_by_card_id(self, card_id: CardId) -> Card | None:
2024-09-27 14:13:03 +00:00
for card in reversed(self.cards):
2024-09-27 09:06:06 +00:00
if card.card_id == card_id:
return card
return None
2024-09-27 09:06:06 +00:00
def get_card_by_seq_id(self, seq_id: SeqId) -> Card | None:
2024-09-27 14:13:03 +00:00
for card in reversed(self.cards):
2024-09-27 09:06:06 +00:00
if card.seq_id == seq_id:
return card
return None
2024-09-27 09:06:06 +00:00
def remove(self, card_id: CardId) -> Card | None:
2024-09-27 14:13:03 +00:00
while card := self.get_card_by_card_id(card_id):
2024-09-27 09:06:06 +00:00
self.cards.remove(card)
2024-09-27 14:13:03 +00:00
return card
return None
2024-09-26 17:16:53 +00:00
2024-09-26 21:08:37 +00:00
class FavroClient:
2024-09-28 10:57:27 +00:00
"""Favro API Client.
Implements methods for interacting with the [Favro API](https://favro.com/developer/).
"""
2024-09-26 21:08:37 +00:00
def __init__(
self,
*,
favro_org_id: OrganizationId,
favro_username: str,
favro_password: str,
session: requests.Session | None = None,
read_only=True,
):
2024-10-01 14:18:46 +00:00
# Check arguments
2024-10-01 14:22:23 +00:00
if not isinstance(favro_org_id, OrganizationId):
2024-10-01 14:18:46 +00:00
msg = f'favro_org_id must be str, but was {type(favro_org_id)}'
raise TypeError(msg)
if not isinstance(favro_username, str):
msg = f'favro_username must be str, but was {type(favro_username)}'
raise TypeError(msg)
if not isinstance(favro_password, str):
msg = f'favro_password must be str, but was {type(favro_password)}'
raise TypeError(msg)
2024-09-26 17:16:53 +00:00
2024-09-26 14:43:30 +00:00
# Setup session
self.session = session or requests.Session()
2024-09-26 17:16:53 +00:00
self.session.auth = (favro_username, favro_password)
2024-09-26 21:08:37 +00:00
self.session.headers.update(
{
2024-09-26 17:16:53 +00:00
'organizationId': favro_org_id.raw_id,
2024-09-26 14:43:30 +00:00
'content-type': 'application/json',
2024-09-26 21:08:37 +00:00
},
)
self.read_only = read_only
2024-09-27 09:06:06 +00:00
self.cache = CardCache()
2024-09-26 14:43:30 +00:00
2024-09-26 17:16:53 +00:00
def check_logged_in(self) -> None:
next(self.get_todo_list_cards())
2024-09-26 16:49:28 +00:00
def get_todo_list_cards(self) -> Iterator[Card]:
2024-10-02 14:23:15 +00:00
for card in self.get_cards(todo_list=True):
self.cache.add_card(card)
yield card
2024-09-26 16:49:28 +00:00
2024-09-26 21:08:37 +00:00
def get_cards(
self,
*,
seq_id: SeqId | None = None,
2024-10-02 15:17:14 +00:00
collection_id: CollectionId | None = None,
todo_list=False,
2024-09-26 21:08:37 +00:00
) -> Iterator[Card]:
2024-10-02 15:17:14 +00:00
request = self._get_cards_request(seq_id=seq_id,todo_list=todo_list,collection_id=collection_id)
2024-10-02 14:23:15 +00:00
yield from self._get_paginated(request, Card.from_json)
2024-10-02 14:23:15 +00:00
def get_collections(self) -> Iterator[Collection]:
request = requests.Request('GET', URL_GET_COLLECTIONS)
yield from self._get_paginated(request, Collection.from_json)
2024-10-02 14:23:15 +00:00
def _get_cards_prepared_request(
self, **kwargs,
) -> requests.PreparedRequest:
request = self._get_cards_request(**kwargs)
return self.session.prepare_request(request)
2024-09-26 16:49:28 +00:00
2024-09-26 21:08:37 +00:00
def _get_cards_request(
self,
seq_id: SeqId | None = None,
2024-10-01 14:14:51 +00:00
todo_list: bool = False,
2024-10-02 15:17:14 +00:00
collection_id: CollectionId | None = None,
2024-10-02 14:23:15 +00:00
) -> requests.Request:
params = {'descriptionFormat': 'markdown'}
if seq_id is not None:
2024-09-26 21:08:37 +00:00
params['cardSequentialId'] = str(seq_id.raw_id)
if todo_list is True:
params['todoList'] = 'true'
2024-10-02 15:17:14 +00:00
if collection_id is not None:
params['collectionId'] = str(collection_id.raw_id)
2024-10-02 14:23:15 +00:00
return requests.Request('GET', URL_GET_CARDS, params=params)
2024-09-26 19:51:53 +00:00
def get_card(self, seq_id: SeqId) -> Card:
2024-09-27 09:06:06 +00:00
if card := self.cache.get_card_by_seq_id(seq_id):
return card
2024-09-26 19:51:53 +00:00
return next(self.get_cards(seq_id=seq_id))
2024-09-26 14:43:30 +00:00
2024-09-28 12:10:13 +00:00
def get_card_by_card_id(self, card_id: CardId) -> Card:
response = self.session.get(URL_UPDATE_CARD.format(card_id=card_id.raw_id))
return Card.from_json(response.json())
def get_user(self, user_id: UserId) -> UserInfo:
response = self.session.get(URL_GET_USER.format(user_id=user_id.raw_id))
return UserInfo.from_json(response.json())
def get_tag(self, tag_id: TagId) -> TagInfo:
response = self.session.get(URL_GET_TAG.format(tag_id=tag_id.raw_id))
return TagInfo.from_json(response.json())
2024-10-01 14:18:46 +00:00
def get_custom_field(self, custom_field_id: CustomFieldId) -> CustomFieldInfo:
2024-10-01 14:14:51 +00:00
response = self.session.get(
2024-10-01 14:15:03 +00:00
URL_GET_CUSTOM_FIELD.format(custom_field_id=custom_field_id.raw_id),
2024-10-01 14:14:51 +00:00
)
2024-10-01 14:06:06 +00:00
return CustomFieldInfo.from_json(response.json())
def _invalidate_cache(self, card_id: CardId) -> None:
2024-09-27 09:06:06 +00:00
card = self.cache.remove(card_id)
2024-09-27 14:13:03 +00:00
if card:
self.session.cache.delete(
2024-10-02 14:23:15 +00:00
requests=[self._get_cards_prepared_request(seq_id=card.seq_id)],
2024-09-27 14:13:03 +00:00
)
def update_card_contents_locally(
2024-09-28 12:13:51 +00:00
self,
card_id: CardId,
card_contents: CardContents,
2024-09-28 12:10:13 +00:00
) -> Card | None:
2024-09-27 14:13:03 +00:00
if card := self.cache.remove(card_id):
card = dataclasses.replace(
card,
detailed_description=card_contents.description,
name=card_contents.name,
2024-09-28 12:13:51 +00:00
tags=[], # TODO?
assignments=[], # TODO?
dependencies=[], # TODO
)
2024-09-27 14:13:03 +00:00
self.cache.add_card(card)
return card
def update_card_contents(
2024-09-28 12:13:51 +00:00
self,
card_id: CardId,
card_contents: CardContents,
2024-09-28 12:10:13 +00:00
) -> Card | None:
2024-09-26 16:49:28 +00:00
"""Returns updated Card."""
if self.read_only == 'silent':
2024-09-26 21:08:37 +00:00
logger.warning(
'FavroClient is silent read only: Discarding card description update',
2024-09-26 21:08:37 +00:00
)
return None # TODO
if self.read_only is True:
msg = 'FavroClient is read only'
raise Exception(msg)
2024-09-26 16:49:28 +00:00
json_body = {
2024-09-27 14:13:03 +00:00
'name': card_contents.name,
'detailedDescription': card_contents.description,
2024-09-26 21:08:37 +00:00
'descriptionFormat': 'markdown',
2024-10-02 08:31:42 +00:00
'archived': card_contents.is_archived,
2024-09-26 16:49:28 +00:00
}
2024-09-27 09:12:14 +00:00
url = URL_UPDATE_CARD.format(card_id=card_id.raw_id)
logger.warning('Sending request: %s', url)
2024-09-26 21:08:37 +00:00
response = self.session.put(
url,
json=json_body,
2024-09-26 21:08:37 +00:00
)
2024-09-26 16:49:28 +00:00
response.raise_for_status()
self._invalidate_cache(card_id)
2024-09-27 14:13:03 +00:00
return self.update_card_contents_locally(card_id, card_contents)
2024-10-02 14:23:15 +00:00
def _get_paginated(self, base_request: requests.Request, entity_from_json) -> Iterator:
page = 0
request_id = None
num_pages = 1
while page < num_pages:
# Determine params for get_cards
base_request.params['page'] = page
base_request.params['request_id'] = request_id
request = self.session.prepare_request(base_request)
# Run query
logger.warning('Sending request: %s', request.url)
response = self.session.send(request)
response.raise_for_status()
json = response.json()
for entity_json in json['entities']:
entity = entity_from_json(entity_json)
yield entity
del entity_json, entity
# Pagination bookkeeping
page = json['page'] + 1
request_id = json['requestId']
num_pages = json['pages']