Embeddable TAM SDK Overview

Introduction

The Embeddable TAM provides a native library support for Python, Java and C# based applications to interact with Trimble Access Management by allowing users to query policy decisions and partial decisions.

Usage

The environment, application credentials, and the token URL are required to fully initialize the library. In the following example these values have been defined as environment variables.

  Environment:
          Variables:
            TAM_CLIENT_ID: __id__
            TAM_CLIENT_SECRET: __secret__
            TAM_TOKEN_URL: __tokenUrl__
            TAM_ENVIRONMENT: stage
            MONITORING_ENABLED: true
      TAM_PARTITION: <<partitionValue>> (Optional)
      LOG_LEVEL: <<DEBUG | INFO | WARN | ERROR>> (Optional)
            OFFLINE_MODE: <<TRUE|FALSE>> (Optional)  # When true, load policies from a local bundle instead of from the server.
            BUNDLE_PATH: <</path/to/bundle.tar.gz>> (Optional)  # Full path to the .tar.gz bundle (used when OFFLINE_MODE=true).

The library is initialized with the above values

# Load the shared library
try:
    lib = ctypes.CDLL('./lib/tamembedded.so')
except OSError as e:
    print(f"Error loading shared library: {e}")

# Define the argument types and the return type of the function
lib.Initialize.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p]
lib.Initialize.restype = None

# Initialize OPA
try:
    client_id = os.getenv("TAM_CLIENT_ID")
    client_secret = os.getenv("TAM_CLIENT_SECRET")
    token_url = os.getenv("TAM_TOKEN_URL")
    env = os.getenv("TAM_ENVIRONMENT")
  partition = os.getenv("TAM_PARTITION")
    lib.Initialize(env.encode('utf-8'), client_id.encode('utf-8'), client_secret.encode('utf-8'), token_url.encode('utf-8'), partition.encode('utf-8))
    logger.info("OPA initialized successfully.")
except Exception as e:
    logger.error(f"Error initializing OPA: {e}")
    raise

Once it has been initialized, the Decision (Evaluate) and Partial Decision (PartialEvaluate) functions may be called on the instance.

# Define the argument types and the return type of the function
lib.Evaluate.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p]
lib.Evaluate.restype = ctypes.c_char_p

def lambda_handler(event, context):
    try:
        result = lib.Evaluate(namespace, rule, input_json)
    except Exception as e:
        logger.error(f"Error in lambda_handler: {e}")

lib.PartialEvaluate.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p]
lib.PartialEvaluate.restype = ctypes.c_char_p

def lambda_handler(event, context):
    try:
        result = lib.PartialEvaluate(namespace, rule, input_json)
    except Exception as e:
        logger.error(f"Error in lambda_handler: {e}")

Examples

Examples of how to use the tam library can be found below.

import ctypes
import json
import time
import logging
import os

# Create a logger
logger = logging.getLogger(__name__)

# Set the log level to INFO
logger.setLevel(logging.INFO)

# Load the shared library
try:
    lib = ctypes.CDLL('./lib/tamembedded.so')
except OSError as e:
    print(f"Error loading shared library: {e}")

# Define the argument types and the return type of the function
lib.Evaluate.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p]
lib.Evaluate.restype = ctypes.c_char_p

lib.PartialEvaluate.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p]
lib.PartialEvaluate.restype = ctypes.c_char_p

# Define the argument types and the return type of the function
lib.Initialize.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p]
lib.Initialize.restype = None

# Initialize OPA
try:
    client_id = os.getenv("TAM_CLIENT_ID")
    client_secret = os.getenv("TAM_CLIENT_SECRET")
    token_url = os.getenv("TAM_TOKEN_URL")
    env = os.getenv("TAM_ENVIRONMENT")
  partition = os.getenv("TAM_PARTITION")
    lib.Initialize(env.encode('utf-8'), client_id.encode('utf-8'), client_secret.encode('utf-8'), token_url.encode('utf-8'), partition.encode('utf-8))
    logger.info("OPA initialized successfully.")
except Exception as e:
    logger.error(f"Error initializing OPA: {e}")
    raise

def lambda_handler(event, context):
    try:
        data = event if 'namespace' in event else json.loads(event['body'])
        namespace = data['namespace'].encode('utf-8')
        rule = data['rule'].encode('utf-8')
        input_json =  json.dumps(data['input']).encode('utf-8')

        # Start the timer
        start_time = time.time()

        # Call the Evaluate function
        #result = lib.Evaluate(namespace, rule, input_json)
        result = lib.PartialEvaluate(namespace, rule, input_json)

        # End the timer
        end_time = time.time()
        # Calculate the time taken in milliseconds
        time_taken = (end_time - start_time) * 1000
        logger.info(f'Time taken for decision: {time_taken} ms')

        # The result is a C string, convert it to a Python string
        result_str = ctypes.c_char_p(result).value.decode('utf-8')
        result_json = json.loads(result_str)

        if 'error' in result_json and result_json['error']:
            logger.error(f"Error: {result_json['error']}, DecisionId: {result_json.get('decisionId', 'N/A')}")
            return {
                'statusCode': 400,
                'body': json.dumps({
                'error': result_json['error'],
                'decisionId': result_json.get('decisionId', 'N/A')
                })
            }
        else:
            return {
                'statusCode': 200,
                'body': json.dumps({
                    'result':result_json['Result']
                })
            }
    except Exception as e:
        logger.error(f"Error in lambda_handler: {e}")
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }