"""Favro API Client. Implements methods for interacting with the [Favro API](https://favro.com/developer/). """ import dataclasses from collections.abc import Iterator from logging import getLogger import requests from .favro_data_model import ( Card, CardId, CustomFieldId, CustomFieldInfo, OrganizationId, SeqId, TagId, TagInfo, UserId, UserInfo, ) from .favro_markdown import CardContents logger = getLogger(__name__) # Endpoints URL_API_ROOT = 'https://favro.com/api/v1' URL_GET_ALL_CARDS = URL_API_ROOT + '/cards' 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}' URL_GET_CUSTOM_FIELD = URL_API_ROOT + '/customfields/{custom_field_id}' URL_GET_TASKS = URL_API_ROOT + '/tasks' class CardCache: 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: for card in reversed(self.cards): if card.card_id == card_id: return card return None def get_card_by_seq_id(self, seq_id: SeqId) -> Card | None: for card in reversed(self.cards): if card.seq_id == seq_id: return card return None def remove(self, card_id: CardId) -> Card | None: while card := self.get_card_by_card_id(card_id): self.cards.remove(card) return card return None class FavroClient: """Favro API Client. Implements methods for interacting with the [Favro API](https://favro.com/developer/). """ def __init__( self, *, favro_org_id: OrganizationId, favro_username: str, favro_password: str, session: requests.Session | None = None, read_only=True, ): # Check arguments if not isinstance(favro_org_id, OrganizationId): 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) # Setup session self.session = session or requests.Session() self.session.auth = (favro_username, favro_password) self.session.headers.update( { 'organizationId': favro_org_id.raw_id, 'content-type': 'application/json', }, ) self.read_only = read_only self.cache = CardCache() def check_logged_in(self) -> None: next(self.get_todo_list_cards()) def get_todo_list_cards(self) -> Iterator[Card]: yield from self.get_cards(todo_list=True) def get_cards( self, *, seq_id: SeqId | None = None, todo_list=False, ) -> Iterator[Card]: page = 0 request_id = None num_pages = 1 while page < num_pages: # Determine params for get_cards request = self._get_cards_request( seq_id, todo_list, page=page, request_id=request_id, ) # 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']: card = Card.from_json(entity_json) self.cache.add_card(card) yield card del entity_json # Pagination bookkeeping page = json['page'] + 1 request_id = json['requestId'] num_pages = json['pages'] def _get_cards_request( self, seq_id: SeqId | None = None, todo_list: bool = False, request_id: None | str = None, page: None | int = None, ) -> requests.PreparedRequest: params = {'descriptionFormat': 'markdown'} if seq_id is not None: params['cardSequentialId'] = str(seq_id.raw_id) if todo_list is True: params['todoList'] = 'true' if request_id: params['requestId'] = request_id if page: params['page'] = page request = requests.Request('GET', URL_GET_ALL_CARDS, params=params) return self.session.prepare_request(request) def get_card(self, seq_id: SeqId) -> Card: if card := self.cache.get_card_by_seq_id(seq_id): return card return next(self.get_cards(seq_id=seq_id)) 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()) def get_custom_field(self, custom_field_id: CustomFieldId) -> CustomFieldInfo: response = self.session.get( URL_GET_CUSTOM_FIELD.format(custom_field_id=custom_field_id.raw_id), ) return CustomFieldInfo.from_json(response.json()) def _invalidate_cache(self, card_id: CardId) -> None: card = self.cache.remove(card_id) if card: self.session.cache.delete( requests=[self._get_cards_request(seq_id=card.seq_id)], ) def update_card_contents_locally( self, card_id: CardId, card_contents: CardContents, ) -> Card | None: if card := self.cache.remove(card_id): card = dataclasses.replace( card, detailed_description=card_contents.description, name=card_contents.name, tags=[], # TODO? assignments=[], # TODO? dependencies=[], # TODO ) self.cache.add_card(card) return card def update_card_contents( self, card_id: CardId, card_contents: CardContents, ) -> Card | None: """Returns updated Card.""" if self.read_only == 'silent': logger.warning( 'FavroClient is silent read only: Discarding card description update', ) return None # TODO if self.read_only is True: msg = 'FavroClient is read only' raise Exception(msg) json_body = { 'name': card_contents.name, 'detailedDescription': card_contents.description, 'descriptionFormat': 'markdown', 'archived': card_contents.archived, } url = URL_UPDATE_CARD.format(card_id=card_id.raw_id) logger.warning('Sending request: %s', url) response = self.session.put( url, json=json_body, ) response.raise_for_status() self._invalidate_cache(card_id) return self.update_card_contents_locally(card_id, card_contents)