How to Enforce Role-Based Data Access in AI Applications with Oracle Deep Data Security

same query shape
same SQL
different Deep Sec end user
different authorized result

WHERE audience = :role

WHERE assigned_officer = :end_user

WHERE audience IN ('general', :role)

if end_user == "linda":
    row["customer_ssn"] = None

CREATE END USER linda IDENTIFIED BY <password>;
CREATE END USER wendy IDENTIFIED BY <password>;

GRANT CREATE DATA ROLE TO deepsec;
GRANT CREATE DATA GRANT TO deepsec;
GRANT CREATE ANY DATA GRANT TO deepsec;
GRANT ADMINISTER ANY DATA GRANT TO deepsec;
GRANT GRANT ANY DATA ROLE TO deepsec;
GRANT CREATE END USER TO deepsec;
GRANT CREATE END USER SECURITY CONTEXT TO deepsec;
GRANT SET USE DATA GRANTS ONLY TO deepsec;

CREATE ROLE deal_direct_logon_role;
GRANT CREATE SESSION TO deal_direct_logon_role;

GRANT deal_direct_logon_role TO loan_officer_role;
GRANT deal_direct_logon_role TO underwriter_role;

SET USE DATA GRANTS ONLY ON loan_applications ENABLED;
SET USE DATA GRANTS ONLY ON loan_policies ENABLED;

git clone https://github.com/markxnelson/deep-data-security.git deal-deepsec
cd deal-deepsec

mkdir deal-deepsec
cd deal-deepsec

python -m venv .venv
source .venv/bin/activate

python -m venv .venv
.venv\Scripts\Activate.ps1

oracledb
python-dotenv
oci

pip install -r requirements.txt

Successfully installed oracledb ...
Successfully installed python-dotenv ...
Successfully installed oci ...

# Oracle AI Database 26ai on ADB-S connection.
ADB_USERNAME=DEEPSEC
ADB_PASSWORD=your-database-password

# Use an ADB wallet TNS alias such as yourdb_high,
# or a TCPS connect descriptor with host, port, and service_name values.
ADB_DSN=your-adb-tns-alias-or-connect-descriptor

# Directory containing the unzipped wallet files, including tnsnames.ora and ewallet.pem.
ADB_WALLET_LOCATION=/path/to/unzipped/adb-wallet

# Wallet password created when the wallet was downloaded.
# This is not the database user password.
ADB_WALLET_PASSPHRASE=your-wallet-password

# Deep Sec context mode must match your administrator-prepared identity setup.
# Use direct_logon for local Deep Sec users without IAM/Entra tokens.
# Use external_token for IAM/Entra-style identity strings.
# Use local_tuple when a service connection sets context for local end-user tuples.
DEEPSEC_CONTEXT_MODE=direct_logon

# Required only for service-user context modes.
# direct_logon does not use this value.
DEEPSEC_DATABASE_ACCESS_TOKEN=

# Direct-logon passwords for the local Deep Sec end users.
DEEPSEC_LINDA_KEY=your-linda-end-user-password
DEEPSEC_WENDY_KEY=your-wendy-end-user-password

# Protected object owner used when direct-logon users query the demo tables.
DEAL_OBJECT_OWNER=DEEPSEC

# OCI Generative AI settings for the final chatbot step.
OCI_CONFIG_FILE=~/.oci/config
OCI_PROFILE=DEFAULT
OCI_GENAI_ENDPOINT=https://inference.generativeai.us-chicago-1.oci.oraclecloud.com
OCI_GENAI_COMPARTMENT_ID=your-compartment-ocid
OCI_GENAI_MODEL_ID=your-generative-ai-model-id

cp .env.example .env

.env
.venv/
__pycache__/
*.pyc
.pytest_cache/
.DS_Store

python 01_verify_deepsec.py
python 02_create_schema.py
python 03_load_data.py
python 04_configure_deepsec.py
python 05_context_read_demo.py
python 07_vector_basics.py
python 08_secure_rag_retrieval.py
python 08b_keyword_vs_vector_check.py  # optional

python 09_deal_tools_demo.py
python 10_run_deal_sessions.py
python 11_oci_genai_chatbot.py  # optional, requires OCI GenAI configuration

import os
import sys
from pathlib import Path

import oracledb
from dotenv import load_dotenv


load_dotenv()


def required_env(name):
    value = os.getenv(name)
    if (
        not value
        or value.startswith("replace-")
        or value.startswith("your-")
        or value.startswith("/path/to/")
    ):
        raise RuntimeError(f"Set {name} in .env before running this script.")
    return value


def first_env(*names):
    for name in names:
        value = os.getenv(name)
        if value and not value.startswith("replace-") and not value.startswith("your-"):
            return value
    raise RuntimeError(f"Set {' or '.join(names)} in .env before running this script.")


def optional_env(*names):
    for name in names:
        value = os.getenv(name)
        if value and not value.startswith("replace-") and not value.startswith("your-"):
            return value
    return None


def wallet_dsn(wallet_dir):
    tnsnames = Path(wallet_dir) / "tnsnames.ora"
    if not tnsnames.exists():
        raise RuntimeError("Set ADB_DSN or provide a wallet with tnsnames.ora.")
    for line in tnsnames.read_text(encoding="utf-8").splitlines():
        stripped = line.strip()
        if "=" in stripped and not stripped.startswith("#"):
            return stripped.split("=", 1)[0].strip()
    raise RuntimeError("No TNS aliases found in wallet tnsnames.ora.")


def connect():
    wallet_dir = first_env("ADB_WALLET_LOCATION", "DB_WALLET_DIR")
    dsn = optional_env("ADB_DSN", "DB_DSN") or wallet_dsn(wallet_dir)
    return oracledb.connect(
        user=first_env("ADB_USERNAME", "DB_USER"),
        password=first_env("ADB_PASSWORD", "DB_PASSWORD"),
        dsn=dsn,
        config_dir=wallet_dir,
        wallet_location=wallet_dir,
        wallet_password=optional_env(
            "ADB_WALLET_PASSPHRASE", "ADB_WALLET_PASSWORD", "DB_WALLET_PASSWORD"
        ),
    )


print("DEAL environment check")

with connect() as conn:
    cur = conn.cursor()

    cur.execute("select sys_context('USERENV', 'CURRENT_SCHEMA') from dual")
    print(f"Connected as: {cur.fetchone()[0]}")

    try:
        cur.execute("select banner_full from v$version where rownum = 1")
        version_text = cur.fetchone()[0]
    except oracledb.DatabaseError:
        cur.execute(
            """
            select version_full
            from product_component_version
            where product like 'Oracle Database%'
            fetch first 1 row only
            """
        )
        version_text = cur.fetchone()[0]

    print(f"Database version: {version_text}")

    mode = "Thin" if oracledb.is_thin_mode() else "Thick"
    print(f"python-oracledb mode: {mode}")

    if mode != "Thin":
        print("Deep Data Security requires python-oracledb Thin mode for this demo.")
        sys.exit(1)

    cur.execute(
        """
        select view_name
        from all_views
        where view_name in (
            'DBA_DATA_ROLES',
            'DBA_DATA_ROLE_GRANTS',
            'DBA_DATA_GRANTS',
            'ALL_DATA_GRANTS',
            'USER_DATA_GRANTS'
        )
        order by view_name
        """
    )
    visible_views = [row[0] for row in cur.fetchall()]

    if visible_views:
        print("Deep Sec metadata visible to this schema: " + ", ".join(visible_views))
    else:
        print("Deep Sec metadata views are not visible to this schema.")
        print("Continue only with an administrator-prepared Deep Sec environment.")

python 01_verify_deepsec.py

DEAL environment check
Connected as: DEEPSEC
Database version: Oracle AI Database 26ai Enterprise Edition Release 23.26.2.1.0
python-oracledb mode: Thin
Deep Sec metadata visible to this schema: ALL_DATA_GRANTS, USER_DATA_GRANTS

Deep Sec metadata views are not visible to this schema.
Continue only with an administrator-prepared Deep Sec environment.

import os
from pathlib import Path

import oracledb
from dotenv import load_dotenv


load_dotenv()


def required_env(name):
    value = os.getenv(name)
    if (
        not value
        or value.startswith("replace-")
        or value.startswith("your-")
        or value.startswith("/path/to/")
    ):
        raise RuntimeError(f"Set {name} in .env before running this script.")
    return value


def first_env(*names):
    for name in names:
        value = os.getenv(name)
        if value and not value.startswith("replace-") and not value.startswith("your-"):
            return value
    raise RuntimeError(f"Set {' or '.join(names)} in .env before running this script.")


def optional_env(*names):
    for name in names:
        value = os.getenv(name)
        if value and not value.startswith("replace-") and not value.startswith("your-"):
            return value
    return None


def wallet_dsn(wallet_dir):
    tnsnames = Path(wallet_dir) / "tnsnames.ora"
    if not tnsnames.exists():
        raise RuntimeError("Set ADB_DSN or provide a wallet with tnsnames.ora.")
    for line in tnsnames.read_text(encoding="utf-8").splitlines():
        stripped = line.strip()
        if "=" in stripped and not stripped.startswith("#"):
            return stripped.split("=", 1)[0].strip()
    raise RuntimeError("No TNS aliases found in wallet tnsnames.ora.")


def connect():
    wallet_dir = first_env("ADB_WALLET_LOCATION", "DB_WALLET_DIR")
    dsn = optional_env("ADB_DSN", "DB_DSN") or wallet_dsn(wallet_dir)
    return oracledb.connect(
        user=first_env("ADB_USERNAME", "DB_USER"),
        password=first_env("ADB_PASSWORD", "DB_PASSWORD"),
        dsn=dsn,
        config_dir=wallet_dir,
        wallet_location=wallet_dir,
        wallet_password=optional_env(
            "ADB_WALLET_PASSPHRASE", "ADB_WALLET_PASSWORD", "DB_WALLET_PASSWORD"
        ),
    )

import oracledb

from deal_db import connect


DDL = [
    """
    create table loan_applications (
        id number primary key,
        customer_name varchar2(100) not null,
        loan_amount number(12,2) not null,
        purpose varchar2(100) not null,
        status varchar2(40) not null,
        officer_notes varchar2(4000),
        customer_ssn varchar2(20),
        customer_income number(12,2),
        customer_credit_score number(4),
        underwriting_decision varchar2(40),
        risk_score number(5,2),
        underwriting_notes varchar2(4000),
        assigned_officer varchar2(40) not null,
        in_underwriting_queue char(1) check (in_underwriting_queue in ('Y','N'))
    )
    """,
    """
    create table loan_policies (
        id number primary key,
        title varchar2(200) not null,
        body varchar2(4000) not null,
        audience varchar2(40) not null,
        embedding vector(3, float32)
    )
    """,
]


def drop_if_exists(cur, table_name):
    try:
        cur.execute(f"drop table {table_name} purge")
    except oracledb.DatabaseError as exc:
        error = exc.args[0]
        if error.code != 942:
            raise


with connect() as conn:
    cur = conn.cursor()

    drop_if_exists(cur, "loan_policies")
    drop_if_exists(cur, "loan_applications")
    print("Dropped existing DEAL demo tables if they existed.")

    for statement in DDL:
        cur.execute(statement)

    print("Created loan_applications.")
    print("Created loan_policies with VECTOR(3, FLOAT32) embedding column.")
    print("Schema setup complete.")

python 02_create_schema.py

Dropped existing DEAL demo tables if they existed.
Created loan_applications.
Created loan_policies with VECTOR(3, FLOAT32) embedding column.
Schema setup complete.

from array import array

from deal_db import connect


loans = [
    (101, "Avery Stone", 320000, "Home purchase", "RECEIVED",
     "Missing final pay stub.", "111-22-3333", 128000, 720,
     "PENDING_REVIEW", 41, "Not reviewed yet.", "LINDA", "N"),
    (102, "Noah Rivers", 485000, "Home purchase", "UNDER_REVIEW",
     "Customer uploaded tax documents.", "222-33-4444", 151000, 695,
     "PENDING_REVIEW", 72, "Watch revolving debt.", "LINDA", "Y"),
    (103, "Maya Chen", 210000, "Refinance", "UNDER_REVIEW",
     "Transferred from branch intake.", "333-44-5555", 98000, 681,
     "PENDING_REVIEW", 68, "Check income variability.", "raj", "Y"),
    (104, "Grace Hill", 150000, "Home equity", "NEEDS_DOCS",
     "Awaiting insurance statement.", "444-55-6666", 87000, 735,
     "NOT_STARTED", 35, "Not in queue.", "amir", "N"),
    (105, "Owen Park", 640000, "Jumbo mortgage", "UNDER_REVIEW",
     "Customer requested expedited review.", "555-66-7777", 220000, 705,
     "PENDING_REVIEW", 77, "Large loan amount.", "LINDA", "Y"),
    (106, "Sofia Reyes", 390000, "Investment property", "ESCALATED",
     "Escalated by branch manager.", "666-77-8888", 132000, 660,
     "PENDING_REVIEW", 83, "Exception review needed.", "raj", "Y"),
]

policies = [
    (1, "General lending eligibility",
     "Baseline eligibility requirements for consumer lending applications.",
     "general", array("f", [0.75, 0.20, 0.05])),
    (2, "Income verification basics",
     "Required documents for income verification and employment review.",
     "general", array("f", [0.65, 0.30, 0.05])),
    (3, "Document retention requirements",
     "Retention periods for lending documents and customer communications.",
     "general", array("f", [0.30, 0.20, 0.50])),
    (4, "Loan officer workflow checklist",
     "Loan officer workflow for intake, status updates, and customer follow-up.",
     "loan_officer", array("f", [0.85, 0.10, 0.05])),
    (5, "Officer notes standards",
     "Standards for writing clear customer-facing officer notes.",
     "loan_officer", array("f", [0.80, 0.05, 0.15])),
    (6, "Credit risk escalation policy",
     "When credit risk signals require underwriting escalation.",
     "underwriter", array("f", [0.05, 0.90, 0.05])),
    (7, "Debt-to-income exception review",
     "Underwriting guidance for debt-to-income ratio exceptions.",
     "underwriter", array("f", [0.10, 0.85, 0.05])),
    (8, "Collateral review guidance",
     "Collateral review requirements for underwriting decisions.",
     "underwriter", array("f", [0.05, 0.75, 0.20])),
]


with connect() as conn:
    cur = conn.cursor()

    cur.executemany(
        """
        insert into loan_applications (
            id, customer_name, loan_amount, purpose, status, officer_notes,
            customer_ssn, customer_income, customer_credit_score,
            underwriting_decision, risk_score, underwriting_notes,
            assigned_officer, in_underwriting_queue
        ) values (:1, :2, :3, :4, :5, :6, :7, :8, :9, :10, :11, :12, :13, :14)
        """,
        loans,
    )

    cur.executemany(
        """
        insert into loan_policies (id, title, body, audience, embedding)
        values (:1, :2, :3, :4, :5)
        """,
        policies,
    )

    conn.commit()

    print(f"Inserted {len(loans)} loan applications.")
    print(f"Inserted {len(policies)} loan policy documents.")
    print("Synthetic data load complete.")

python 03_load_data.py

Inserted 6 loan applications.
Inserted 8 loan policy documents.
Synthetic data load complete.

from deal_db import connect


with connect() as conn:
    cur = conn.cursor()

    print("Using prepared Deep Sec end users: LINDA, WENDY")

    cur.execute("create data role loan_officer_role")
    print("Created data role: LOAN_OFFICER_ROLE")

    cur.execute("create data role underwriter_role")
    print("Created data role: UNDERWRITER_ROLE")

    cur.execute("grant data role loan_officer_role to linda")
    print("Granted LOAN_OFFICER_ROLE to LINDA.")

    cur.execute("grant data role underwriter_role to wendy")
    print("Granted UNDERWRITER_ROLE to WENDY.")

    grants = [
        """
        create or replace data grant deal_loan_officer_read as
        select (
            all columns except customer_ssn, customer_income,
            customer_credit_score, underwriting_decision,
            risk_score, underwriting_notes
        )
        on loan_applications
        where assigned_officer = ORA_END_USER_CONTEXT.username
        to loan_officer_role
        """,
        """
        create or replace data grant deal_underwriter_read as
        select on loan_applications
        where in_underwriting_queue = 'Y'
        to underwriter_role
        """,
        """
        create or replace data grant deal_policy_general_to_officer as
        select on loan_policies
        where audience = 'general'
        to loan_officer_role
        """,
        """
        create or replace data grant deal_policy_officer as
        select on loan_policies
        where audience = 'loan_officer'
        to loan_officer_role
        """,
        """
        create or replace data grant deal_policy_general_to_underwriter as
        select on loan_policies
        where audience = 'general'
        to underwriter_role
        """,
        """
        create or replace data grant deal_policy_underwriter as
        select on loan_policies
        where audience = 'underwriter'
        to underwriter_role
        """,
    ]

    for grant in grants:
        cur.execute(grant)

    conn.commit()

    print("Created loan application read grants.")
    print("Created loan policy read grants.")
    print("Configured Deep Data Security for DEAL.")

python 04_configure_deepsec.py

Using prepared Deep Sec end users: LINDA, WENDY
Created data role: LOAN_OFFICER_ROLE
Created data role: UNDERWRITER_ROLE
Granted LOAN_OFFICER_ROLE to LINDA.
Granted UNDERWRITER_ROLE to WENDY.
Created loan application read grants.
Created loan policy read grants.
Configured Deep Data Security for DEAL.

GRANT deal_direct_logon_role TO loan_officer_role;
GRANT deal_direct_logon_role TO underwriter_role;

SET USE DATA GRANTS ONLY ON loan_applications ENABLED;
SET USE DATA GRANTS ONLY ON loan_policies ENABLED;

def direct_logon_connect(end_user):
    passwords = {
        "linda": required_env("DEEPSEC_LINDA_KEY"),
        "wendy": required_env("DEEPSEC_WENDY_KEY"),
    }
    if end_user not in passwords:
        raise ValueError(f"Unknown DEAL end user: {end_user}")

    wallet_dir = first_env("ADB_WALLET_LOCATION", "DB_WALLET_DIR")
    dsn = optional_env("ADB_DSN") or optional_env("DB_DSN") or wallet_dsn(wallet_dir)
    return oracledb.connect(
        user=end_user.upper(),
        password=passwords[end_user],
        dsn=dsn,
        config_dir=wallet_dir,
        wallet_location=wallet_dir,
        wallet_password=optional_env(
            "ADB_WALLET_PASSPHRASE", "ADB_WALLET_PASSWORD", "DB_WALLET_PASSWORD"
        ),
    )


def run_for_user(end_user, work):
    if (optional_env("DEEPSEC_CONTEXT_MODE") or "direct_logon") == "direct_logon":
        with direct_logon_connect(end_user) as conn:
            return work(conn)
    raise RuntimeError("This tutorial validates DEEPSEC_CONTEXT_MODE=direct_logon.")


def object_name(name):
    owner = optional_env("DEAL_OBJECT_OWNER") or "DEEPSEC"
    return f"{owner}.{name}"

from deal_db import object_name, run_for_user


SQL = """
select *
from {loan_applications}
order by id
""".format(loan_applications=object_name("loan_applications"))

FIELDS_TO_SHOW = [
    "customer_ssn",
    "customer_credit_score",
    "underwriting_decision",
    "risk_score",
]


def display_value(value):
    return "NULL" if value is None else value


def rows_for(end_user):
    def work(conn):
        cur = conn.cursor()
        cur.execute(SQL)
        columns = [col[0].lower() for col in cur.description]
        return [dict(zip(columns, row)) for row in cur.fetchall()]

    return run_for_user(end_user, work)


def show(end_user):
    rows = rows_for(end_user)
    ids = [str(row["id"]) for row in rows]

    print(f"\nAs {end_user}:")
    print(f"  Rows returned: {len(rows)}")
    print(f"  Visible application ids: {', '.join(ids)}")

    sample = rows[0] if rows else {}
    print("  Selected sensitive and underwriting fields:")
    for field in FIELDS_TO_SHOW:
        print(f"    {field}: {display_value(sample.get(field))}")


print("Read and column enforcement demo")
show("linda")
show("wendy")

python 05_context_read_demo.py

Read and column enforcement demo

As linda:
  Rows returned: 3
  Visible application ids: 101, 102, 105
  Selected sensitive and underwriting fields:
    customer_ssn: NULL
    customer_credit_score: NULL
    underwriting_decision: NULL
    risk_score: NULL

As wendy:
  Rows returned: 4
  Visible application ids: 102, 103, 105, 106
  Selected sensitive and underwriting fields:
    customer_ssn: 222-33-4444
    customer_credit_score: 695
    underwriting_decision: PENDING_REVIEW
    risk_score: 72

SELECT *
FROM loan_applications
ORDER BY id

from array import array

from deal_db import object_name, run_for_user


query_vector = array("f", [0.9, 0.1, 0.0])

sql = """
select title,
       vector_distance(embedding, :query_vector, COSINE) as distance
from {loan_policies}
order by distance
fetch first 3 rows only
""".format(loan_policies=object_name("loan_policies"))

print("Vector warm-up with manual 3-dimensional vectors")
print("Vector warm-up user context: linda")
print(f"Query vector: {list(query_vector)}")

def work(conn):
    cur = conn.cursor()
    cur.execute(sql, query_vector=query_vector)

    print("\nNearest policy vectors:")
    for index, (title, distance) in enumerate(cur, start=1):
        print(f"{index}. {title:<35} distance: {distance:.6f}")


run_for_user("linda", work)

python 07_vector_basics.py

Vector warm-up with manual 3-dimensional vectors
Vector warm-up user context: linda
Query vector: [0.8999999761581421, 0.10000000149011612, 0.0]

Nearest policy vectors:
1. Loan officer workflow checklist     distance: 0.001723
2. General lending eligibility         distance: 0.013266
3. Officer notes standards             distance: 0.018206

unstable cash flow or credit risk

from array import array

from deal_db import object_name, run_for_user


query_text = "unstable cash flow or credit risk"
query_vector = array("f", [0.05, 0.9, 0.05])

sql = """
select id, title, audience,
       vector_distance(embedding, :query_vector, COSINE) as distance
from {loan_policies}
order by distance
fetch first 3 rows only
""".format(loan_policies=object_name("loan_policies"))


def search_as(end_user):
    def work(conn):
        cur = conn.cursor()
        cur.execute(sql, query_vector=query_vector)
        return cur.fetchall()

    return run_for_user(end_user, work)


print("Deep Sec-scoped vector retrieval demo")
print(f"Policy search query: {query_text}")
print("Output below is scoped by the active Deep Sec end user.")

for end_user in ["linda", "wendy"]:
    print(f"\nAs {end_user}:")
    for index, row in enumerate(search_as(end_user), start=1):
        print(f"  {index}. {row[1]:<35} distance: {row[3]:.6f}")

python 08_secure_rag_retrieval.py

Deep Sec-scoped vector retrieval demo
Policy search query: unstable cash flow or credit risk
Output below is scoped by the active Deep Sec end user.

As linda:
  1. Income verification basics          distance: 0.529221
  2. Document retention requirements     distance: 0.604677
  3. General lending eligibility         distance: 0.686696

As wendy:
  1. Credit risk escalation policy      distance: 0.000000
  2. Debt-to-income exception review    distance: 0.001895
  3. Collateral review guidance         distance: 0.020925

from array import array

from deal_db import object_name, run_for_user


query_vector = array("f", [0.05, 0.9, 0.05])

vector_sql = """
select title
from {loan_policies}
order by vector_distance(embedding, :query_vector, COSINE)
fetch first 3 rows only
""".format(loan_policies=object_name("loan_policies"))

keyword_sql = """
select title
from {loan_policies}
where lower(body) like :term
order by id
fetch first 3 rows only
""".format(loan_policies=object_name("loan_policies"))


def run_as(end_user):
    def work(conn):
        cur = conn.cursor()

        cur.execute(vector_sql, query_vector=query_vector)
        vector_titles = [row[0] for row in cur.fetchall()]

        cur.execute(keyword_sql, term="%credit risk%")
        keyword_titles = [row[0] for row in cur.fetchall()]

        return vector_titles, keyword_titles

    return run_for_user(end_user, work)


print("Optional vector-style vs keyword retrieval check")

for end_user in ["linda", "wendy"]:
    vector_titles, keyword_titles = run_as(end_user)
    print(f"\nAs {end_user}:")
    print(f"  Vector-style titles: {', '.join(vector_titles)}")
    print(
        "  Keyword titles for 'credit risk': "
        + (", ".join(keyword_titles) if keyword_titles else "none")
    )

python 08b_keyword_vs_vector_check.py

Optional vector-style vs keyword retrieval check

As linda:
  Vector-style titles: Income verification basics, Document retention requirements, General lending eligibility
  Keyword titles for 'credit risk': none

As wendy:
  Vector-style titles: Credit risk escalation policy, Debt-to-income exception review, Collateral review guidance
  Keyword titles for 'credit risk': Credit risk escalation policy

from array import array

from deal_db import object_name, run_for_user


def _rows_as_dicts(cur):
    columns = [col[0].lower() for col in cur.description]
    return [dict(zip(columns, row)) for row in cur.fetchall()]


def _run_for_user(end_user, work):
    return run_for_user(end_user, work)


def _query_vector(query):
    text = query.lower()
    if "risk" in text or "credit" in text or "cash flow" in text:
        return array("f", [0.05, 0.9, 0.05])
    return array("f", [0.9, 0.1, 0.0])

def get_loan_applications(end_user):
    def work(conn):
        cur = conn.cursor()
        cur.execute(
            f"""
            select *
            from {object_name("loan_applications")}
            order by id
            """
        )
        return _rows_as_dicts(cur)

    return _run_for_user(end_user, work)


def get_application_detail(end_user, app_id):
    def work(conn):
        cur = conn.cursor()
        cur.execute(
            f"""
            select *
            from {object_name("loan_applications")}
            where id = :id
            """,
            id=app_id,
        )
        rows = _rows_as_dicts(cur)
        return rows[0] if rows else None

    return _run_for_user(end_user, work)

def search_policies(end_user, query):
    query_vector = _query_vector(query)

    def work(conn):
        cur = conn.cursor()
        cur.execute(
            f"""
            select id, title, body, audience,
                   vector_distance(embedding, :query_vector, COSINE) as distance
            from {object_name("loan_policies")}
            order by distance
            fetch first 3 rows only
            """,
            query_vector=query_vector,
        )
        return _rows_as_dicts(cur)

    return _run_for_user(end_user, work)

from deal_tools import (
    get_loan_applications,
    search_policies,
)


print("Tool demo: linda")
linda_loans = get_loan_applications("linda")
linda_policies = search_policies("linda", "credit risk")
print(f"  get_loan_applications returned {len(linda_loans)} rows.")
print(f"  search_policies returned {len(linda_policies)} policy documents.")

print("\nTool demo: wendy")
wendy_loans = get_loan_applications("wendy")
wendy_policies = search_policies("wendy", "credit risk")
print(f"  get_loan_applications returned {len(wendy_loans)} rows.")
print(f"  search_policies returned {len(wendy_policies)} policy documents.")

python 09_deal_tools_demo.py

Tool demo: linda
  get_loan_applications returned 3 rows.
  search_policies returned 3 policy documents.

Tool demo: wendy
  get_loan_applications returned 4 rows.
  search_policies returned 3 policy documents.

from deal_tools import (
    get_application_detail,
    get_loan_applications,
    search_policies,
)


def display_value(value):
    return "NULL" if value is None else value


def titles(rows):
    return ", ".join(row["title"] for row in rows)


def linda_session():
    print("========================")
    print("DEAL session: linda")
    print("========================")

    loans = get_loan_applications("linda")
    ids = [str(row["id"]) for row in loans]
    detail = get_application_detail("linda", 102)
    policies = search_policies("linda", "credit risk")

    print(f"Visible applications: {len(loans)}")
    print(f"Application ids: {', '.join(ids)}")
    print(f"Restricted risk_score: {display_value(detail.get('risk_score'))}")
    print(f"Policy results: {titles(policies)}")


def wendy_session():
    print("\n========================")
    print("DEAL session: wendy")
    print("========================")

    loans = get_loan_applications("wendy")
    ids = [str(row["id"]) for row in loans]
    detail = get_application_detail("wendy", 102)
    policies = search_policies("wendy", "credit risk")

    print(f"Visible applications: {len(loans)}")
    print(f"Application ids: {', '.join(ids)}")
    print(f"Underwriting risk_score: {display_value(detail.get('risk_score'))}")
    print(f"Policy results: {titles(policies)}")

def bypass_check():
    print("\n========================")
    print("Bypass check")
    print("========================")

    linda_rows = get_loan_applications("linda")
    wendy_rows = get_loan_applications("wendy")
    linda_docs = search_policies("linda", "credit risk")
    wendy_docs = search_policies("wendy", "credit risk")

    print(f"Broad loan query returned {len(linda_rows)} Linda-scoped rows for linda.")
    print(f"Broad loan query returned {len(wendy_rows)} Wendy-scoped rows for wendy.")
    print(f"Linda policy titles: {titles(linda_docs)}")
    print(f"Wendy policy titles: {titles(wendy_docs)}")
    print("No application-side row filter, redaction, or audience filter was used.")


linda_session()
wendy_session()
bypass_check()

python 10_run_deal_sessions.py

========================
DEAL session: linda
========================
Visible applications: 3
Application ids: 101, 102, 105
Restricted risk_score: NULL
Policy results: Income verification basics, Document retention requirements, General lending eligibility

========================
DEAL session: wendy
========================
Visible applications: 4
Application ids: 102, 103, 105, 106
Underwriting risk_score: 72
Policy results: Credit risk escalation policy, Debt-to-income exception review, Collateral review guidance

========================
Bypass check
========================
Broad loan query returned 3 Linda-scoped rows for linda.
Broad loan query returned 4 Wendy-scoped rows for wendy.
Linda policy titles: Income verification basics, Document retention requirements, General lending eligibility
Wendy policy titles: Credit risk escalation policy, Debt-to-income exception review, Collateral review guidance
No application-side row filter, redaction, or audience filter was used.

import json
import os
import re
from pathlib import Path

import oci
from dotenv import load_dotenv
from oci.generative_ai_inference import GenerativeAiInferenceClient
from oci.generative_ai_inference.models import (
    ChatDetails,
    GenericChatRequest,
    Message,
    OnDemandServingMode,
    UserMessage,
)

from deal_tools import (
    get_application_detail,
    get_loan_applications,
    search_policies,
)


load_dotenv()


TOOLS = {
    "get_loan_applications": get_loan_applications,
    "get_application_detail": get_application_detail,
    "search_policies": search_policies,
}


def required_env(name):
    value = os.getenv(name)
    if value:
        value = value.strip().strip('"').strip("'")
    if not value or value.startswith("your-"):
        raise RuntimeError(f"Set {name} in .env before running this script.")
    return value


def oci_client():
    config_file = Path(os.getenv("OCI_CONFIG_FILE", "~/.oci/config")).expanduser()
    profile = os.getenv("OCI_PROFILE", "DEFAULT")
    config = oci.config.from_file(str(config_file), profile)
    return GenerativeAiInferenceClient(
        config=config,
        service_endpoint=required_env("OCI_GENAI_ENDPOINT"),
    )


def message_text(chat_response):
    data = chat_response.data
    if hasattr(data, "chat_response") and hasattr(data.chat_response, "choices"):
        choice = data.chat_response.choices[0]
        return choice.message.content[0].text
    return json.dumps(oci.util.to_dict(data))


def chat_once(client, prompt):
    request = GenericChatRequest(
        api_format=GenericChatRequest.API_FORMAT_GENERIC,
        messages=[
            UserMessage(
                role=Message.ROLE_USER,
                content=[{"type": "TEXT", "text": prompt}],
            )
        ],
        max_tokens=500,
        temperature=0,
    )
    details = ChatDetails(
        compartment_id=required_env("OCI_GENAI_COMPARTMENT_ID"),
        serving_mode=OnDemandServingMode(model_id=required_env("OCI_GENAI_MODEL_ID")),
        chat_request=request,
    )
    return message_text(client.chat(details))


def choose_action(client, end_user, question):
    prompt = f"""
You are DEAL, a lending assistant running as end user {end_user}.
Choose exactly one tool call for the user's request.
Return only JSON with keys tool and arguments.

Available tools:
- get_loan_applications: {{}}
- get_application_detail: {{"app_id": number}}
- search_policies: {{"query": string}}

User request: {question}
"""
    raw = chat_once(client, prompt).strip()
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        fenced = re.search(r"```

(?:json)?\s*(\{.*?\})\s*

```", raw, re.DOTALL)
        if fenced:
            return json.loads(fenced.group(1))
        start = raw.find("{")
        end = raw.rfind("}")
        if start >= 0 and end > start:
            return json.loads(raw[start : end + 1])
        raise


def run_tool(end_user, action):
    tool_name = action["tool"]
    arguments = action.get("arguments", {})
    if tool_name not in TOOLS:
        raise ValueError(f"Unknown tool requested by model: {tool_name}")
    return TOOLS[tool_name](end_user, **arguments)


def answer_with_result(client, end_user, question, action, result):
    prompt = f"""
You are DEAL, a lending assistant running as end user {end_user}.
The database has already enforced Oracle Deep Data Security.
Answer the user briefly from the tool result only.
If the tool result is a list of loan applications, count the list and list the id values exactly.
If the tool result is a list of policies, list only the returned policy titles.
Do not say that no rows or policies were found unless the tool result is an empty list.

User request: {question}
Tool action: {json.dumps(action)}
Tool result: {json.dumps(result, default=str)}
"""
    return chat_once(client, prompt)


def run_session(client, end_user, questions):
    print(f"\n========================")
    print(f"OCI GenAI DEAL session: {end_user}")
    print(f"========================")
    for question in questions:
        action = choose_action(client, end_user, question)
        result = run_tool(end_user, action)
        answer = answer_with_result(client, end_user, question, action, result)
        print(f"\nUser: {question}")
        print(f"Tool: {action['tool']} {action.get('arguments', {})}")
        print(f"DEAL: {answer}")


client = oci_client()

run_session(
    client,
    "linda",
    [
        "Which loan applications can I see?",
        "Find policy guidance about credit risk.",
    ],
)

run_session(
    client,
    "wendy",
    [
        "Which applications are in my underwriting queue?",
        "Find policy guidance about credit risk.",
    ],
)

python 11_oci_genai_chatbot.py

========================
OCI GenAI DEAL session: linda
========================

User: Which loan applications can I see?
Tool: get_loan_applications {}
DEAL: You can see 3 loan applications: 101, 102, 105.

User: Find policy guidance about credit risk.
Tool: search_policies {'query': 'credit risk'}
DEAL: Income verification basics, Document retention requirements, General lending eligibility

========================
OCI GenAI DEAL session: wendy
========================

User: Which applications are in my underwriting queue?
Tool: get_loan_applications {}
DEAL: There are 4 applications in your underwriting queue: 102, 103, 105, 106.

User: Find policy guidance about credit risk.
Tool: search_policies {'query': 'credit risk'}
DEAL: Credit risk escalation policy
Debt-to-income exception review
Collateral review guidance

oracledb.init_oracle_client()

conn = pool.acquire()
try:
    context = context_for(end_user)
    conn.set_end_user_security_context(context)
    # Run user-scoped database work.
finally:
    conn.clear_end_user_security_context()
    conn.close()