favro-sync/favro_sync/favro_client.py

"""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}'


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, str):
            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)