favro-sync/favro_sync/favro_client.py

from collections.abc import Iterator
from logging import getLogger

import requests

from .favro_data_model import Card, CardId, OrganizationId, SeqId

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())