import requests import dataclasses import datetime from typing import Any import functools from collections.abc import Iterator from logging import getLogger from .favro_data_model import Card, CardId, SeqId, OrganizationId 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}' class FavroClient: def __init__(self, *, 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_username is not None assert favro_password is not None # 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.card_id_to_seq_id: dict[CardId, SeqId] = {} self.seq_id_to_card_id: dict[SeqId, CardId] = {} 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]: # Determine params for get_cards request = self._get_cards_request(seq_id, todo_list) # Run query response = self.session.send(request) response.raise_for_status() json = response.json() # TODO: Pageination for entity_json in json['entities']: card = Card.from_json(entity_json) self.card_id_to_seq_id[card.card_id] = card.seq_id self.seq_id_to_card_id[card.seq_id] = card.card_id yield card del entity_json def _get_cards_request(self, seq_id: SeqId | None = None, todo_list=False) -> 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' request = requests.Request('GET', URL_GET_ALL_CARDS, params = params) return self.session.prepare_request(request) def get_card(self, seq_id: SeqId) -> Card: return next(self.get_cards(seq_id=seq_id)) def get_card_id(self, seq_id: SeqId) -> CardId: if card_id := self.seq_id_to_card_id[seq_id]: return card_id first_card = next(self.get_cards(seq_id = seq_id)) return first_card.card_id def _invalidate_cache(self, card_id: CardId) -> None: self.session.cache.delete(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: """Returns updated Card.""" if self.read_only == 'silent': logger.warning('FavroClient is silent read only: Discarding card description update of length %d', len(description)) return None # TODO elif self.read_only is True: raise Exception('FavroClient is read only') json_body = { 'detailedDescription': description, 'descriptionFormat': 'markdown', } 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.raise_for_status() logger.warning("Response: %s", response.json()) self._invalidate_cache(card_id) return Card.from_json(response.json())