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
var socket_listener
-
Api
's instance ofSocketListener
Static methods
def has_account()
-
Checks if
BRIAR_DB
existsAdded 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
ifApiThread
is runningAdded 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)
-
Login to Briar API with
password
.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)
-
Register at Briar API with 2-tuple
credentials
.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()