Some OCI Integration Essentials

Show and Tell
1

Hello everyone,

Seems to have been ages since I dipped in. Just wanted to quickly share some very fast ways to integrate your Anvil projects with OCI - which may open up some opportunities for you given the wide spread of end users across health, public sector, and finance.

All you need to do is import the following python libraries in your app:

oracledb
oci

First - Functions:

OCI Functions work like AWS Lambdas. To integrate in the easiest possible way, all you need to do is tap the endpoint from your Anvil application.

def get_sensor_data():
  url = anvil.secrets.get_secret('endpoint')
  try:
    time.sleep(1)
    response = requests.get(url)
    response.raise_for_status() #raise error if not 200
    return response.json()
  except requests.RequestException as e:
    print("Error calling sensor function")
    return None

Next - Databases:

Over 50% of the world’s corporate data is in an Oracle database somewhere, so it seems useful to be able to tap an OCI hosted Autonomous database from an Anvil app, if you are aiming to integrate.

To create a connection, upload the database mTLS wallet zip to data files in your anvil app and create a few secrets:

connection=oracledb.connect(
     config_dir=[f'''{anvil.secrets.get_secret('directory')}'''], #Save the path to your directory as a secret or drop it in plain if you like.
     user=anvil.secrets.get_secret('db_user'),
     password=anvil.secrets.get_secret('db_pw'),
     dsn=anvil.secrets.get_secret('dsn'),
     wallet_location=[f'''{anvil.secrets.get_secret('directory')}'''],
     wallet_password=anvil.secrets.get_secret('walletpw'))

To drop a table:

def drop_table():
  # Drop a table
  with connection.cursor() as cursor:
    cursor.execute("""
    begin
      execute immediate 'drop table sensor_logs';
      exception when others then if sqlcode <> -942 then raise; end if;
    end;""")

To create a table:

def create_table():
  # Create a table
  with connection.cursor() as cursor:
  cursor.execute("""
        create table sensor_logs (
            id number generated always as identity,
            sensor varchar2(255),
            movement varchar2(10),
            sound varchar2(10),
            geo_lat number(9,6),
            geo_lon number(9,6),
            sensor_time timestamp,
            urn varchar2(255),
            creation_ts timestamp with time zone default current_timestamp,
            primary key (id))""")

To write some data to the DB:

def handle_low_priority(data):
  sensor = data['sensor']
  movement = data['movement']
  sound = data['sound']
  geo = data['geo']
  if geo:
    lat,lon = map(float,geo.split(","))
  else:
    lat,lon = None,None
  sensor_time = parse(data['timestamp'])
  urn = data['URN']
  try:
    with connection.cursor() as cursor:
      cursor.execute(
        "insert into sensor_logs (sensor, movement, sound, geo_lat, geo_lon, 
        sensor_time, urn) values (:1, :2, :3, :4, :5, :6, :7, :8)",
        (sensor,movement,sound,lat,lon,sensor_time,urn)
      )
      connection.commit()
  finally:
    connection.close()
    print("Sensor added")

Lastly - AI Agents:

Firstly, grab the info you need from your OCI tenancy (you need an API key and some details about the agent endpoint and its region). Set your config.

config = {
    "user": anvil.secrets.get_secret('user'),
    "key_file": data_files['file'],
    "fingerprint": anvil.secrets.get_secret('fingerprint'),
    "tenancy": anvil.secrets.get_secret('tenancy'),
    "region": anvil.secrets.get_secret('region')
}
service_ep = anvil.secrets.get_secret('sp')
agent_ep_id = anvil.secrets.get_secret('ep')

Then it’s just a simple bit of code to integrate a fast AI Agent and its knowledge base into your app.

# Response Generator
@anvil.server.callable
def response_generator(textinput):
    # Initialise service client with default config file
    generative_ai_agent_runtime_client = oci.generative_ai_agent_runtime.GenerativeAiAgentRuntimeClient(config,service_endpoint=service_ep)
 
    # Create Session
    create_session_response = generative_ai_agent_runtime_client.create_session(
        create_session_details=oci.generative_ai_agent_runtime.models.CreateSessionDetails(
            display_name="USER_Session",
            description="User Session"),
        agent_endpoint_id=agent_ep_id)
 
    sess_id = create_session_response.data.id
 
    response = generative_ai_agent_runtime_client.chat(
        agent_endpoint_id=agent_ep_id,
        chat_details=oci.generative_ai_agent_runtime.models.ChatDetails(
            user_message=textinput,
            session_id=sess_id))
 
    response = response.data.message.content.text
    return response

Anyway, enjoy and Happy Easter!

2 Likes