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
instanceapi
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
Methods
def add_pending(self, link: str, alias: str) ‑> NoneType
-
Adds pending contact to Briar with
link
URL andalias
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
-
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
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
-
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
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 changesAdded 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