Oracle NoSQL Database Python SDK

5.4.3 · active · verified Thu Apr 16

Borneo is the official Python SDK for Oracle NoSQL Database, version 5.4.3. It provides interfaces for developing Python applications that connect to the Oracle NoSQL Database Cloud Service, on-premise Oracle NoSQL Database, and the Oracle NoSQL Cloud Simulator. The library is actively maintained with regular feature releases and bug fixes, typically on a quarterly or bi-annual cadence.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart connects to Oracle NoSQL Database (Cloud Service, Simulator, or On-Premise) using OCI credentials provided via environment variables. It demonstrates how to create a table, put a record, retrieve a record by its primary key, and finally drop the table, cleaning up resources. For cloud connections, ensure the `oci` package is installed and your OCI API key configuration (tenancy_id, user_id, fingerprint, private_key_path, optional passphrase, and compartment_id) is correctly set in environment variables or hardcoded.

import os
import time
from borneo import (
    NoSQLHandle,
    NoSQLHandleConfig,
    Regions,
    TableLimits,
    TableRequest,
    PutRequest,
    GetRequest,
    DeleteRequest
)
from borneo.iam import SignatureProvider
from borneo.exception import NoSQLException, InvalidAuthorizationException

# --- Configure OCI Credentials and Region ---
# Replace with your actual OCI configuration or set environment variables
# OCI_REGION can be found at https://docs.oracle.com/en-us/iaas/Content/General/Concepts/regions.htm
# OCI_TENANCY, OCI_USER, OCI_FINGERPRINT are from your OCI API Key setup
# OCI_PRIVATE_KEY_PATH points to your private key file (e.g., ~/.oci/oci_api_key.pem)
# OCI_PRIVATE_KEY_PASSPHRASE is optional if your key is passphrase-protected

region = os.environ.get('OCI_REGION', 'us-ashburn-1') # e.g., Regions.US_ASHBURN_1
tenancy_id = os.environ.get('OCI_TENANCY_ID', '')
user_id = os.environ.get('OCI_USER_ID', '')
fingerprint = os.environ.get('OCI_FINGERPRINT', '')
private_key_path = os.environ.get('OCI_PRIVATE_KEY_PATH', os.path.expanduser('~/.oci/oci_api_key.pem'))
private_key_passphrase = os.environ.get('OCI_PRIVATE_KEY_PASSPHRASE', '')

# Configure table details
table_name = "myUserTable"
compartment_id = os.environ.get('OCI_COMPARTMENT_ID', tenancy_id) # Use tenancy_id if compartment_id is not specifically set


def get_nosql_handle():
    try:
        provider = SignatureProvider(
            tenant_id=tenancy_id,
            user_id=user_id,
            fingerprint=fingerprint,
            private_key_file=private_key_path,
            private_key_passphrase=private_key_passphrase
        )
        config = NoSQLHandleConfig(region, provider)
        config.set_default_compartment(compartment_id)
        return NoSQLHandle(config)
    except InvalidAuthorizationException as e:
        print(f"Authentication failed: {e}")
        print("Please ensure OCI_REGION, OCI_TENANCY_ID, OCI_USER_ID, OCI_FINGERPRINT, "
              "and OCI_PRIVATE_KEY_PATH environment variables are correctly set "
              "or hardcoded with valid OCI credentials.")
        return None

def create_and_wait_for_table(handle):
    table_limits = TableLimits(10, 10, 1)
    statement = f'CREATE TABLE IF NOT EXISTS {table_name} (id INTEGER, name STRING, age INTEGER, PRIMARY KEY (id))'
    request = TableRequest().set_statement(statement).set_table_limits(table_limits)

    print(f"Creating table {table_name}...")
    try:
        table_result = handle.do_table_request(request, 20000, 1000) # timeout_ms, polling_interval_ms
        print(f"Table {table_name} created and is in state: {table_result.get_state()}")
    except NoSQLException as e:
        print(f"Error creating table: {e}")
        # Check if the table already exists, which might be acceptable
        if "Table already exists" not in str(e):
            raise

def put_and_get_data(handle):
    # Put a row
    put_request = PutRequest().set_table_name(table_name).set_value({'id': 1, 'name': 'Alice', 'age': 30})
    put_result = handle.put(put_request)
    print(f"Put row: {put_result.get_version()}")

    # Get the row
    get_request = GetRequest().set_table_name(table_name).set_key({'id': 1})
    get_result = handle.get(get_request)
    if get_result.get_value():
        print(f"Get row: {get_result.get_value()}")
    else:
        print("Row not found.")

def delete_table(handle):
    statement = f'DROP TABLE IF EXISTS {table_name}'
    request = TableRequest().set_statement(statement)

    print(f"Dropping table {table_name}...")
    try:
        table_result = handle.do_table_request(request, 20000, 1000)
        print(f"Table {table_name} dropped and is in state: {table_result.get_state()}")
    except NoSQLException as e:
        print(f"Error dropping table: {e}")

if __name__ == '__main__':
    handle = None
    try:
        handle = get_nosql_handle()
        if not handle:
            exit(1)

        create_and_wait_for_table(handle)
        put_and_get_data(handle)
    finally:
        if handle:
            delete_table(handle) # Clean up the table
            handle.close()
            print("NoSQLHandle closed.")

view raw JSON →