Using the Solution Database on FICO Platform

All Xpress components (Insight, Workbench, and Executor) deployed on FICO® Platform in the same solution, at the same lifecycle stage, will share the solution database.

insight.get_solution_database can be used to access the solution database.

requirements.txt

# Pip module dependencies
--pre
xpress==8.14.2
xpressinsight==1.7.*
requests==2.28.1
numpy==1.23.4
debugpy==1.6.3
mysql-connector-python==8.0.33

application.py

import os
import requests
import subprocess
import time
import xpressinsight as xi
import mysql.connector
from mysql.connector import errorcode


@xi.ExecMode(name="ConnectToMySql", descr="Connect to MySql", threads=1, clear_input=True)
def call_connect_mysql(self):
    teamnames = [["Wibbles"], ["Wobbles"]]
    soldb = self.insight.get_solution_database()

    try:
        cnx = mysql.connector.connect(
            user=soldb.user,
            password=soldb.password,
            host=soldb.host,
            port=soldb.port,
            database=soldb.database
        )
        print("Connected to database")

        cursor = cnx.cursor()

        cursor.execute("SHOW STATUS LIKE 'ssl_cipher'")
        print("SSL Ciphers: ", cursor.fetchone())

        cursor.execute("drop table if exists MoselTest")
        cursor.execute("create table MoselTest (team VARCHAR(300))")
        cursor.executemany("INSERT into MoselTest (team) values(%s)", teamnames)
        cnx.commit()
        print("Created table MoselTest")
        print("rows inserted =", cursor.rowcount)

        cursor.execute("SELECT team FROM MoselTest ORDER BY team ASC")
        teams = ",".join(map( lambda row: row[0], cursor.fetchall()))
        print(f"teams [{teams}]")

        cursor.close()
    except mysql.connector.Error as err:
        if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
            print("Something is wrong with your username or password")
        elif err.errno == errorcode.ER_BAD_DB_ERROR:
            print("Database does not exist")
        else:
            print(err)
    finally:
        if cnx.is_connected():
            cnx.close()
            print('connection closed')