contacts.py

# Copyright (c) 2019 Nico Alt
# SPDX-License-Identifier: AGPL-3.0-only
# License-Filename: LICENSE.md
"""
Wrapper around Briar API's _/contacts/_ resource
"""

from operator import itemgetter
from typing import Callable, List
from urllib.parse import urljoin

from requests import delete as _delete
from requests import get as _get
from requests import post as _post
from requests import put as _put

from briar_wrapper.constants import BASE_HTTP_URL
from briar_wrapper.model import Model


class Contacts(Model):

    _API_ENDPOINT = "contacts/"
    _CONNECTION_EVENTS = ("ContactConnectedEvent", "ContactDisconnectedEvent")

    _connections_callback = None

    def add_pending(self, link: str, alias: str) -> None:
        # pylint: disable=line-too-long
        """

        Adds pending contact to Briar with `link` URL and `alias`

        [Upstream documentation](https://code.briarproject.org/briar/briar/blob/master/briar-headless/README.md#adding-a-contact)

        .. versionadded:: 0.0.3
        """
        url = urljoin(BASE_HTTP_URL, self._API_ENDPOINT + "add/pending/")
        _post(url, headers=self._headers, json={"link": link, "alias": alias})

    def set_alias(self, contact_id: int, alias: str) -> None:
        # pylint: disable=line-too-long
        """
        Sets the alias of a given user

        [Upstream documentation](https://code.briarproject.org/briar/briar/-/blob/master/briar-headless/README.md#changing-alias-of-a-contact)

        .. versionadded:: 0.0.5
        .. versionchanged:: 0.0.6
        """
        url = urljoin(BASE_HTTP_URL, self._API_ENDPOINT + f"{str(contact_id)}/alias")
        _put(url, headers=self._headers, json={"alias": alias})

    def delete(self, contact_id: int) -> None:
        # pylint: disable=line-too-long
        """

        Deletes the contact with `contact_id`

        [Upstream documentation](https://code.briarproject.org/briar/briar/blob/master/briar-headless/README.md#removing-a-contact)

        .. versionadded:: 0.0.4
        """
        url = urljoin(BASE_HTTP_URL, self._API_ENDPOINT + str(contact_id))
        _delete(url, headers=self._headers)

    def get(self) -> list:
        # pylint: disable=line-too-long
        """
        Returns sorted list containing all contacts

        [Upstream documentation](https://code.briarproject.org/briar/briar/blob/master/briar-headless/README.md#listing-all-contacts)

        .. versionadded:: 0.0.3
        .. versionchanged:: 0.0.4
        """
        url = urljoin(BASE_HTTP_URL, self._API_ENDPOINT)
        request = _get(url, headers=self._headers)
        contacts = request.json()
        contacts = Contacts._sort_contact_list(contacts)
        return contacts

    def get_link(self) -> str:
        # pylint: disable=line-too-long
        """
        Returns _briar://_ link

        [Upstream documentation](https://code.briarproject.org/briar/briar/blob/master/briar-headless/README.md#adding-a-contact)

        .. versionadded:: 0.0.3
        """
        url = urljoin(BASE_HTTP_URL, self._API_ENDPOINT + "add/link/")
        request = _get(url, headers=self._headers).json()
        return request['link']

    def watch_connections(self, callback: Callable) -> List[int]:
        # pylint: disable=line-too-long
        """
        Calls `callback` whenever a contact's connection status changes

        [Upstream documentation](https://code.briarproject.org/briar/briar/blob/master/briar-headless/README.md#a-contact-connected-or-disconnected)

        .. versionadded:: 0.0.4
        """
        self._connections_callback = callback
        signal_ids = list()
        event_callback = self.handle_connections_callback
        for event in self._CONNECTION_EVENTS:
            signal_id = self._api.socket_listener.connect(event,
                                                          event_callback)
            signal_ids.append(signal_id)
        return signal_ids

    def handle_connections_callback(self, message: str) -> None:
        contact_id = message["data"]["contactId"]
        if message["name"] == "ContactConnectedEvent":
            self._connections_callback(contact_id, True)
        elif message["name"] == "ContactDisconnectedEvent":
            self._connections_callback(contact_id, False)
        else:
            raise Exception(f"Wrong event in callback: {message['name']}")

    @staticmethod
    def _sort_contact_list(contacts: list) -> list:
        contacts.sort(key=itemgetter("lastChatActivity"),
                      reverse=True)
        return contacts