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:

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

Functions

def default_endpoint() -> str

Get the URL of the Router.

Returns

str: Typically "https://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, active_team: Optional[int] = 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://tripleblind.app" if not otherwise specified.
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.
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

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://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 organization_id : uuid.UUID

The organization associated with the session's user.

Returns

UUID: The organization id.

var organization_name : uuid.UUID

The organization associated with the session's user.

Returns

UUID: The organization id.

var organization_owner : bool

Is the session's user an organization owner?

Returns

bool: Is the session's user an organization owner?

var team_id : int

The team id associated with the session's user.

Returns

int: A identifier for the user's team.

var team_name : str

The team name associated with the session's user.

Returns

int: The user's team name.

var user_email : str

The email of the user associated with the session's user.

Returns

str: The user's email.

var user_id : uuid.UUID

var user_name : str

The full name of the user associated with the session's user.

Returns

str: The user's full name.

var user_username : str

The username of the user associated with the session's user.

Returns

str: The user's username.

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, no_exception_on_500=False) -> 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 post request.
files : any: the files to send in the post request.
json : any: the json to send in the post request, avoiding the need to call json.dumps() and set header type
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 post request.
timeout : int: timeout value passed to session post method

Returns

res (ResponseWrapper): The response from post

def put(self, url: str, data=None, json=None, 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