Module tripleblind.session

A Session connects the script to the TripleBlind Router. In a typical script, the user token is retrieved from tripleblind.yaml during a call to tb.initialize(), creating the default session for all operations.

It is also possible to create multiple Sessions, allowing a single script to connect as different users on the Router.

The user authentication token can be found in the TripleBlind web interface:

  1. Log in at https://eval.tripleblind.app
  2. Click on the My Account menu item
  3. Scroll to "Authentication Token" and click the Show button

Functions

def default_endpoint() -> str

Get the URL of the Router.

Returns

str
Typically "https://eval.tripleblind.app"
def get_default_session() -> Session

Retrieves the Session used in operations which don't specify a Session.

Raises

Exception
Raised if tb.initialize() has not been called yet.

Returns

Session
The connection to the Router as a specific user.
def initialize(api_token: Optional[str] = None, endpoint: Optional[str] = None, ca_bundle: Optional[str] = None, example: Optional[bool] = False) -> Session

Establishes the default connection session

The default config values are set in the configuration file "tripleblind.yaml", either in the directory tree of the top-level script being run or pointed to by the environment variable TB_API_CONFIG.

Args

api_token : str
The access token for the user. You can obtain this from the TripleBlind website via the My Account page. The default is to use the sdkuser_token value from your tripleblind.yaml configuration file.
endpoint : str, optional
The URL to connect with. Defaults to endpoint from the endpoint value in the tripleblind.yaml configuration file or "https://eval.tripleblind.app" if not otherwise specified.
ca_bundle : str, optional
Advanced setting generally not necessary. Can also be set using ca_bundle in the config YAML.
example : bool, optional
When set to True, special steps are taken to make sure the example runs cleanly. Example data is verified and created if necessary, and the current directory is set to the directory of the example script.

Returns

Session
The default session

Raises

Exception
Thrown if already initialized
def parse_API_response(res: ResponseWrapper) -> dict

Safe parser for API responses

Args

res : ResponseWrapper
A response from a get/put/etc.

Raises

ValueError
if result is not JSON
TripleblindTokenError
the given user token failed to validate

Returns

dict
The JSON response object
def perform_req(session, url, method, headers, json=None, data=None, tries=3, timeout=10) -> ResponseWrapper
def router_proxy_request(session, router_request: RouterProxyRequest, org_info, tries=3, timeout=10) -> ResponseWrapper
def run(command: str, subdir: str)

Launch the command as a subprocess

Args

command : str
Command to execute
subdir : str
Subdirectory to use.
def tb_version(example_dir: pathlib.Path)

Use tb.py cli tool to run tb –version.

def test_cache()

Classes

class RequestWrapper (method: str, headers: Any, url: str, data: Any = None, json: Dict[str, Any] = None)

RequestWrapper(method: str, headers: Any, url: str, data: Any = None, json: Dict[str, Any] = None)

Class variables

var data : Any
var headers : Any
var json : Dict[str, Any]
var method : str
var url : str
class ResponseWrapper (headers: Any, status_code: int, ok: bool, reason: str, data: bytes, request: RequestWrapper)

ResponseWrapper(headers: Any, status_code: int, ok: bool, reason: str, data: bytes, request: tripleblind.session.RequestWrapper)

Class variables

var data : bytes
var headers : Any
var ok : bool
var reason : str
var request : RequestWrapper
var status_code : int
class RouterProxyRequest (method: str, router_url: str, data: Optional[Any] = None, json: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, Any]] = None)

RouterProxyRequest(method: str, router_url: str, data: Optional[Any] = None, json: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, Any]] = None)

Class variables

var data : Optional[Any]
var headers : Optional[Dict[str, Any]]
var json : Optional[Dict[str, Any]]
var method : str
var router_url : str

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) -> ~A
def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) -> ~A
def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) -> dataclasses_json.mm.SchemaF[~A]

Methods

def to_dict(self, encode_json=False) -> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]
def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) -> str
class Session (api_token: str, endpoint: str = None, ca_bundle: Union[pathlib.Path, str, ForwardRef(None)] = None, active_team: int = None, from_default: bool = False)

A connection to the Router as a specific user.

Define the connection details

Args

api_token : str
User token (from web interface)
endpoint : str, optional
URL to contact, e.g. "https://eval.tripleblind.app"
ca_bundle : Union[Path, str, None], optional
Certificate authority bundle
active_team : int, optional
The team within the organization which is used for API call context. Default is the team set in your configuration file.
from_default : bool, optional
When true, missing values will be based on the default connection

Raises

Exception
Failed to load either the .crt or .key file

Class variables

var endpoint : str

Target URL of the Router.

var token : str

The Router API access token for the user.

Instance variables

var active_team : str

The id of the team within the organization used for API call context.

var org_id : int

The organization id associated with the session's user.

Returns

int
A identifier for the user's organization.
var org_name : str

The organization name associated with the session's user.

Returns

int
The user's organization name.
var user_id : uuid.UUID

Methods

def delete(self, url: str, data=None, headers=None, is_router_url=True, tries: int = 3, timeout: int = 10) -> ResponseWrapper

Wraps aiohttp.ClientSession delete method and adds retry logic and debugging

Args

url : str
the api endpoint to call.
data : any
the data to send in the delete request.
headers : any
the headers to send in the delete request.
is_router_url
(bool): indicates if the supplied url is a router URL. True by default.
tries : int
number of tries to attempt the delete request.
timeout : int
timeout value passed to session delete method

Returns

res (ResponseWrapper): The response from delete

def get(self, url: str, headers=None, is_router_url=True, tries: int = 3, timeout: int = 10) -> ResponseWrapper

Wraps aiohttp.ClientSession get method and adds retry logic and debugging

Args

url
(str): URL of the get endpoint.
headers
(any, optional): Headers to pass in the get.
is_router_url
(bool, optional): Is the supplied url is a router? True by default.
tries : int, optional
number of tries to attempt the get request, default = 3.
timeout : int, optional
Timeout passed to aiohttp.ClientSession get method, default is 10 seconds.

Returns

res (ResponseWrapper): The response from get

def get_access_point_version(self) -> str

Retrieve the version string from the Access Point

Raises

TripleblindNoResponse
description
TripleblindAPIError
description
TripleblindTokenError
description
TripleblindPermissionError
description
TripleblindNoResponse
description
TripleblindNoResponse
description

Returns

str
The version string (e.g. "1.45.1") or "UNKNOWN" if unable to connect to the AP.
def get_as_json(self, url: str, headers=None, is_router_url=True, tries: int = 3, timeout: int = 10) -> dict

Wraps get() method, returning the result parsed as a JSON structure

Args

url : str
URL of the get endpoint.
headers : any, optional
Headers to pass in the get.
is_router_url
(bool, optional): Is the supplied url is a router? True by default.
tries : int, optional
number of tries to attempt the get request, default = 3.
timeout : int, optional
Timeout passed to aiohttp.ClientSession get method, default is 10 seconds.

Returns

dict
The response loaded as if it were JSON
def patch(self, url: str, data=None, headers=None, is_router_url=True, tries: int = 3, timeout: int = 10) -> ResponseWrapper

Wraps aiohttp.ClientSession patch method and adds retry logic and debugging

Args

url
(str): the url of the patch endpoint.
headers
(any): the headers to pass in the patch.
is_router_url
(bool): indicates if the supplied url is a router URL. True by default.
tries : int
number of tries to attempt the patch request.
timeout : int
timeout value passed to aiohttp.ClientSession patch method

Returns

res (ResponseWrapper): The response from patch

def post(self, url: str, data=None, files=None, json=None, headers=None, is_router_url=True, tries=3, timeout=10) -> ResponseWrapper

Wraps aiohttp.ClientSession post method and adds retry logic and debugging

Args

url : str
the api endpoint to call.
data : any
the data to send in the delete request.
headers : any
the headers to send in the delete request.
is_router_url
(bool): indicates if the supplied url is a router URL. True by default.
tries : int
number of tries to attempt the delete request.
timeout : int
timeout value passed to session delete method

Returns

res (ResponseWrapper): The response from delete

def put(self, url: str, headers=None, is_router_url=True, tries: int = 3, timeout: int = 10) -> ResponseWrapper

Wraps aiohttp.ClientSession put method and adds retry logic and debugging

Args

url
(str): the url of the put endpoint.
headers
(any): the headers to pass in the put.
is_router_url
(bool): indicates if the supplied url is a router URL. True by default.
tries : int
number of tries to attempt the put request.
timeout : int
timeout value passed to the put method

Returns

res (ResponseWrapper): The response from put