Module `briar_wrapper.api`

Central API wrapper handling login and registration

Expand source code

# Copyright (c) 2019 Nico Alt
# SPDX-License-Identifier: AGPL-3.0-only
# License-Filename: LICENSE.md
"""
Central API wrapper handling login and registration
"""

import os
from threading import Thread

from briar_wrapper.api_thread import ApiThread
from briar_wrapper.constants import BRIAR_AUTH_TOKEN, BRIAR_DB
from briar_wrapper.models.socket_listener import SocketListener


class Api:

    # pylint: disable=line-too-long
    auth_token = None
    """
    Briar's authentication token

    [Upstream documentation](https://code.briarproject.org/briar/briar/blob/master/briar-headless/README.md#how-to-use)
    """

    socket_listener = None
    """
    `briar_wrapper.api.Api`'s instance of
    `briar_wrapper.models.socket_listener.SocketListener`
    """

    _api_thread = None

    def __init__(self, headless_jar):
        """
        Initialize with path to Briar Headless JAR `headless_jar`
        """
        self._api_thread = ApiThread(self, headless_jar)

    @staticmethod
    def has_account():
        """
        Checks if `briar_wrapper.constants.BRIAR_DB` exists

        .. versionadded:: 0.0.3
        """
        return os.path.isfile(BRIAR_DB)

    def is_running(self):
        """
        Returns `True` if `briar_wrapper.api_thread.ApiThread` is running

        .. versionadded:: 0.0.3
        """
        return self._api_thread.is_running()

    def login(self, password, callback):
        """
        Login to Briar API with `password`.

        Calls `callback` once login process finished.

        .. versionadded:: 0.0.3
        """
        self._start_and_watch(callback)
        startup_thread = Thread(target=self._api_thread.login,
                                args=(password,), daemon=True)
        startup_thread.start()

    def register(self, credentials, callback):
        """
        Register at Briar API with 2-tuple `credentials`.

        Calls `callback` once registration process finished.

        .. versionadded:: 0.0.3
        """
        if len(credentials) != 2:
            raise Exception("Can't process credentials")
        self._start_and_watch(callback)
        startup_thread = Thread(target=self._api_thread.register,
                                args=(credentials,), daemon=True)
        startup_thread.start()

    def stop(self):
        """
        Stops API wrapper

        .. versionadded:: 0.0.3
        """
        self._api_thread.stop()

    def _start_and_watch(self, callback):
        self._api_thread.start()
        self._api_thread.watch(callback)

    def on_successful_startup(self, callback):
        """
        Called by `briar_wrapper.api_thread.ApiThread` if startup finished
        successfully.

        Should not be called from outside `briar_wrapper`.
        """
        self._load_auth_token()
        self.socket_listener = SocketListener(self)
        callback(True)

    def _load_auth_token(self):
        if not Api.has_account():
            raise Exception("Can't load authentication token")
        with open(BRIAR_AUTH_TOKEN, 'r') as file:
            self.auth_token = file.read()

Classes

class Api (headless_jar)

Initialize with path to Briar Headless JAR headless_jar

Expand source code

class Api:

    # pylint: disable=line-too-long
    auth_token = None
    """
    Briar's authentication token

    [Upstream documentation](https://code.briarproject.org/briar/briar/blob/master/briar-headless/README.md#how-to-use)
    """

    socket_listener = None
    """
    `briar_wrapper.api.Api`'s instance of
    `briar_wrapper.models.socket_listener.SocketListener`
    """

    _api_thread = None

    def __init__(self, headless_jar):
        """
        Initialize with path to Briar Headless JAR `headless_jar`
        """
        self._api_thread = ApiThread(self, headless_jar)

    @staticmethod
    def has_account():
        """
        Checks if `briar_wrapper.constants.BRIAR_DB` exists

        .. versionadded:: 0.0.3
        """
        return os.path.isfile(BRIAR_DB)

    def is_running(self):
        """
        Returns `True` if `briar_wrapper.api_thread.ApiThread` is running

        .. versionadded:: 0.0.3
        """
        return self._api_thread.is_running()

    def login(self, password, callback):
        """
        Login to Briar API with `password`.

        Calls `callback` once login process finished.

        .. versionadded:: 0.0.3
        """
        self._start_and_watch(callback)
        startup_thread = Thread(target=self._api_thread.login,
                                args=(password,), daemon=True)
        startup_thread.start()

    def register(self, credentials, callback):
        """
        Register at Briar API with 2-tuple `credentials`.

        Calls `callback` once registration process finished.

        .. versionadded:: 0.0.3
        """
        if len(credentials) != 2:
            raise Exception("Can't process credentials")
        self._start_and_watch(callback)
        startup_thread = Thread(target=self._api_thread.register,
                                args=(credentials,), daemon=True)
        startup_thread.start()

    def stop(self):
        """
        Stops API wrapper

        .. versionadded:: 0.0.3
        """
        self._api_thread.stop()

    def _start_and_watch(self, callback):
        self._api_thread.start()
        self._api_thread.watch(callback)

    def on_successful_startup(self, callback):
        """
        Called by `briar_wrapper.api_thread.ApiThread` if startup finished
        successfully.

        Should not be called from outside `briar_wrapper`.
        """
        self._load_auth_token()
        self.socket_listener = SocketListener(self)
        callback(True)

    def _load_auth_token(self):
        if not Api.has_account():
            raise Exception("Can't load authentication token")
        with open(BRIAR_AUTH_TOKEN, 'r') as file:
            self.auth_token = file.read()

Class variables

var auth_token: Briar's authentication token

Upstream documentation
var socket_listener: Api's instance of SocketListener

Static methods

def has_account()

Checks if BRIAR_DB exists

Added in version: 0.0.3

Expand source code

@staticmethod
def has_account():
    """
    Checks if `briar_wrapper.constants.BRIAR_DB` exists

    .. versionadded:: 0.0.3
    """
    return os.path.isfile(BRIAR_DB)

Methods

def is_running(self)

Returns True if ApiThread is running

Added in version: 0.0.3

Expand source code

def is_running(self):
    """
    Returns `True` if `briar_wrapper.api_thread.ApiThread` is running

    .. versionadded:: 0.0.3
    """
    return self._api_thread.is_running()

def login(self, password, callback)

Calls callback once login process finished.

Added in version: 0.0.3

Expand source code

def login(self, password, callback):
    """
    Login to Briar API with `password`.

    Calls `callback` once login process finished.

    .. versionadded:: 0.0.3
    """
    self._start_and_watch(callback)
    startup_thread = Thread(target=self._api_thread.login,
                            args=(password,), daemon=True)
    startup_thread.start()

def on_successful_startup(self, callback)

Called by ApiThread if startup finished successfully.

Should not be called from outside briar_wrapper.

Expand source code

def on_successful_startup(self, callback):
    """
    Called by `briar_wrapper.api_thread.ApiThread` if startup finished
    successfully.

    Should not be called from outside `briar_wrapper`.
    """
    self._load_auth_token()
    self.socket_listener = SocketListener(self)
    callback(True)

def register(self, credentials, callback)

Calls callback once registration process finished.

Added in version: 0.0.3

Expand source code

def register(self, credentials, callback):
    """
    Register at Briar API with 2-tuple `credentials`.

    Calls `callback` once registration process finished.

    .. versionadded:: 0.0.3
    """
    if len(credentials) != 2:
        raise Exception("Can't process credentials")
    self._start_and_watch(callback)
    startup_thread = Thread(target=self._api_thread.register,
                            args=(credentials,), daemon=True)
    startup_thread.start()

def stop(self)

Stops API wrapper

Added in version: 0.0.3

Expand source code

def stop(self):
    """
    Stops API wrapper

    .. versionadded:: 0.0.3
    """
    self._api_thread.stop()