Monitoring custom integrations by logging to the Integrations dashboard

# Set up LUSID and Horizon API
import pandas as pd
import logging
logging.basicConfig(level = logging.INFO)

import lusid as lu
import lusid.api as la
import lusid.models as lm
from lusidjam import RefreshingToken
import lusid.extensions as le
from finbourne_sdk_utils.lpt.lpt import to_date

lusid_config_loaders=[
    le.ArgsConfigurationLoader(api_url="https://<your-domain>.lusid.com/api" ,access_token = RefreshingToken(), app_name = "LusidJupyterNotebook"),
    le.EnvironmentVariablesConfigurationLoader()]
lusid_api_factory = le.SyncApiClientFactory(config_loaders=lusid_config_loaders)

# Set pandas display options
pd.set_option("display.max_columns", None)
pd.set_option("display.max_rows", None)
pd.options.display.float_format = "{:,.2f}".format

import finbourne_horizon as horizon
from finbourne_horizon.models.external_log_insertion_request import ExternalLogInsertionRequest
from finbourne_horizon.models.external_log_record import ExternalLogRecord
from finbourne_horizon.exceptions import ApiException
from finbourne_horizon.extensions.configuration_options import ConfigurationOptions
from finbourne_horizon.models import *
from pprint import pprint

horizon_config_loaders=[
    horizon.ArgsConfigurationLoader(api_url = "https://<your-domain>.lusid.com/horizon", access_token = RefreshingToken(), app_name = "LusidJupyterNotebook"),
    horizon.EnvironmentVariablesConfigurationLoader()
    ]
horizon_api_factory = horizon.SyncApiClientFactory(config_loaders=horizon_config_loaders)

integrations_api = horizon_api_factory.build(horizon.IntegrationsApi)
logs_api = horizon_api_factory.build(horizon.LogsApi)
runs_api = horizon_api_factory.build(horizon.RunsApi)

# Obtain the Instruments API
instruments_api=lusid_api_factory.build(la.InstrumentsApi)

# Read assets into pandas dataframe from securities file and show it
assets_df = pd.read_csv("data/assets.csv", keep_default_na = False)
display(assets_df)

One-time setup of integration instance:

# ONE-TIME SETUP: Create integration instance (run this cell only once, then comment it out)
triggers = [Trigger(type = "", cronExpression="", time_zone="")]

try:
     create_instance_request = CreateInstanceRequest(
         name="Instrument Upsert Integration",
         description="Logs for instrument upsert operations from Jupyter notebook",
         integration_type="external-client-application",
         triggers = triggers,
         details = "",
         enabled=True
     )
     
     instance_response = integrations_api.create_instance(create_instance_request=create_instance_request)
     instance_id = instance_response.id
     print(f"Created integration instance with ID: {instance_id}")
     print("IMPORTANT: Save this instance_id and use it in the next cell")
except horizon.ApiException as e:
     print(f"Error creating instance: {e}")

Kick off a run of the instance and log or except the results:

# After running the above once, paste your instance_id here:
instance_id = "b13f175e-c50a-443d-ba0d-de26b5ad8f0b"  # Replace with your actual instance ID

print(f"Using instance ID: {instance_id}")

# Create a dictionary of instrument definitions from asset data
definitions = {}

# Iterate over each row in the assets dataframe
for index, asset in assets_df.iterrows():
    
    # Map identifier columns to case-sensitive LUSID identifier names       
    identifiers = {
        # Unique identifiers
        "Figi": lm.InstrumentIdValue(value = asset["figi"]),
        "ClientInternal": lm.InstrumentIdValue(value = asset["internal_id"]),
        # Non-unique identifiers
        "Isin": lm.InstrumentIdValue(value = asset["isin"]),
        "Ticker": lm.InstrumentIdValue(value = asset["ticker"])  
    }
                                       
    # Model equities
    if asset["security_type"] == "equity":
        # Create definitions
        definitions[asset["instrument_name"]] = lm.InstrumentDefinition(
            name = asset["instrument_name"],
            identifiers = identifiers,
            definition = lm.Equity(
                instrument_type = "Equity",
                dom_ccy = asset["currency"],
                identifiers = {}
            )
        )
    # Model bonds
    elif asset["security_type"] == "govt bond":
        definitions[asset["instrument_name"]] = lm.InstrumentDefinition(
            name = asset["instrument_name"],
            identifiers = identifiers,
            definition = lm.Bond(
                instrument_type = "Bond",
                start_date = "2021-01-01T00:00:00Z",
                maturity_date = asset["maturity_date"],
                dom_ccy = asset["currency"],
                flow_conventions = lm.FlowConventions(
                    currency = asset["currency"],
                    payment_frequency = "6M",
                    day_count_convention = "ActualActual",
                    roll_convention = "NoAdjustment",
                    payment_calendars = [],
                    reset_calendars = [],
                    settle_days = 0,
                    reset_days = 0
                ),
                principal = 1,
                coupon_rate = float(asset["coupon"])
            )
        )

# Start an integration run to get a run ID
try:
    run_response = integrations_api.execute_instance(instance_id=instance_id)
    run_id = run_response.execution_id
    print(f"Started integration run with ID: {run_id}")
except horizon.ApiException as e:
    print(f"Error starting integration run: {e}")
    raise

try:
    # Upsert instruments to a custom scope in LUSID
    upsert_instruments_response = instruments_api.upsert_instruments(
        request_body = definitions, 
        scope = "external-client-application",
    )
    
    # Transform API response to a dataframe
    upsert_instruments_response_df = lusid_response_to_data_frame(
        list(upsert_instruments_response.values.values())
    )
    
    # Display the results
    display(upsert_instruments_response_df[["name", "lusidInstrumentId"]])
    
    # Log successful upserts - one log entry per instrument
    logs_to_insert = []
    
    for index, row in upsert_instruments_response_df.iterrows():
        message = f"Successfully upserted instrument: {row['name']}"
        log_record = ExternalLogRecord(
            logid=index + 1,
            loglevel="Resource",
            logstatus="ResourceLoaded",
            sourcerecordtype="Instrument",
            sourceprimaryidtype="InstrumentName",
            sourceprimaryidvalue=row["name"],
            targetprimaryidtype="LusidInstrumentId",
            targetprimaryidvalue=row["lusidInstrumentId"],
            targetrecordtype="LUSID",
            targetrecordaction="Upsert",
            message=message[:1024],
            timestamp="2025-01-22 23:12:13.1234"
        )
        logs_to_insert.append(log_record)
    
    # Insert logs to Horizon
    external_log_insertion_request = ExternalLogInsertionRequest(logs=logs_to_insert)
    
    logs_response = logs_api.insert_external_logs(
        instanceid=instance_id,
        runid=run_id,
        external_log_insertion_request=external_log_insertion_request
    )
    
    print(f"Successfully logged {len(logs_to_insert)} instrument upserts to Horizon")
    
    # Complete the integration run successfully
    try:
        runs_api.stop_instance_execution(instance_id=instance_id, run_id=run_id)
        print(f"Stopped integration run {run_id}")
    except horizon.ApiException as e:
        print(f"Warning: Could not stop integration run: {e}")
    
except lu.ApiException as e:
    # Log the failure
    print(f"Failed to upsert instruments: {e}")
    
    error_message = f"Failed to upsert instruments to LUSID: {str(e)}"
    error_log = ExternalLogRecord(
        logid=1,
        loglevel="Resource",
        logstatus="ResourceFailed",
        sourcerecordtype="InstrumentUpsert",
        message=error_message[:1024],
        timestamp="2025-01-22 23:12:13.1234"
    )
    
    external_log_insertion_request = ExternalLogInsertionRequest(logs=[error_log])
    
    try:
        logs_response = logs_api.insert_external_logs(
            instanceid=instance_id,
            runid=run_id,
            external_log_insertion_request=external_log_insertion_request
        )
        print("Failure logged to Horizon")
        
        # Stop the integration run
        runs_api.stop_instance_execution(instance_id=instance_id, run_id=run_id)
        print(f"Stopped integration run {run_id}")
        
    except horizon.ApiException as log_error:
        print(f"Failed to log error to Horizon: {log_error}")
    
    # Re-raise the original exception
    raise

except Exception as e:
    # Catch any other unexpected errors
    print(f"Unexpected error: {e}")
    
    error_message = f"Unexpected error during instrument upsert: {str(e)}"
    error_log = ExternalLogRecord(
        logid=1,
        loglevel="Payload",
        logstatus="PayloadFailed",
        sourcerecordtype="InstrumentUpsert",
        message=error_message[:1024],
        timestamp="2025-01-22 23:12:13.1234"
    )
    
    external_log_insertion_request = ExternalLogInsertionRequest(logs=[error_log])
    
    try:
        logs_response = logs_api.insert_external_logs(
            instanceid=instance_id,
            runid=run_id,
            external_log_insertion_request=external_log_insertion_request
        )
        print("Failure logged to Horizon")
        
        # Stop the integration run
        runs_api.stop_instance_execution(instance_id=instance_id, run_id=run_id)
        print(f"Stopped integration run {run_id}")
        
    except horizon.ApiException as log_error:
        print(f"Failed to log error to Horizon: {log_error}")
    
    raise