Module `briar_wrapper.models.contacts`

Wrapper around Briar API's /contacts/ resource

Expand source code

# 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

Classes

class Contacts (api)

Initialize with Api instance api

Expand source code

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

Ancestors

Model

Methods

def add_pending(self, link: str, alias: str) ‑> NoneType

Adds pending contact to Briar with link URL and alias

Upstream documentation

Added in version: 0.0.3

Expand source code

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 delete(self, contact_id: int) ‑> NoneType

Deletes the contact with contact_id

Upstream documentation

Added in version: 0.0.4

Expand source code

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

Returns sorted list containing all contacts

Upstream documentation

Added in version: 0.0.3

Changed in version: 0.0.4

Expand source code

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

Returns briar:// link

Upstream documentation

Added in version: 0.0.3

Expand source code

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 handle_connections_callback(self, message: str) ‑> NoneType

Expand source code

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

def set_alias(self, contact_id: int, alias: str) ‑> NoneType

Sets the alias of a given user

Upstream documentation

Added in version: 0.0.5

Changed in version: 0.0.6

Expand source code

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 watch_connections(self, callback: Callable) ‑> List[int]

Calls callback whenever a contact's connection status changes

Upstream documentation

Added in version: 0.0.4

Expand source code

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