OrbexDB Connect Python SDK

Connect Python applications and Genorbex notebooks directly to OrbexDB. The SDK uses the OrbexDB native query API and has no third-party dependencies.

Installation

python -m pip install orbexdb-connect

Create an OrbexDB connection/API key in the OrbexDB connection settings, then configure the SDK:

export ORBEXDB_API_KEY="your-64-character-api-key"
export ORBEXDB_DATABASE_ID="your-database-id"

Managed Genorbex notebooks can use the injected runtime token instead:

from orbexdb_connect import connect

db = connect(database_id="your-database-id")

For an external Python application:

from orbexdb_connect import connect

db = connect(
    database_id="your-database-id",
    api_key="your-api-key",
    schema="PUBLIC",
)

Table operations

db.create_table(
    "customers",
    {
        "id": "NUMBER NOT NULL",
        "name": "VARCHAR(100)",
        "email": "VARCHAR(255)",
        "active": "BOOLEAN",
    },
)

db.insert("customers", {
    "id": 1,
    "name": "Ada",
    "email": "ada@example.com",
    "active": True,
})

db.insert("customers", [
    {"id": 2, "name": "Grace", "email": "grace@example.com", "active": True},
    {"id": 3, "name": "Linus", "email": "linus@example.com", "active": False},
])

db.update("customers", {"active": False}, where={"id": 2})
db.delete("customers", where={"id": 3})

db.merge(
    "customers",
    [
        {"id": 1, "name": "Ada Lovelace", "email": "ada@example.com", "active": True},
        {"id": 4, "name": "Margaret", "email": "margaret@example.com", "active": True},
    ],
    key_columns=["id"],
)

result = db.query("SELECT id, name FROM customers ORDER BY id")
for row in result:
    print(row["id"], row["name"])

db.truncate_table("customers")
db.drop_table("customers")

update and delete require a where mapping by default. To intentionally affect every row, pass allow_all=True.

Metadata

schemas = db.list_schemas()
tables = db.list_tables()
columns = db.describe_table("customers")

Raw SQL

Any SQL operation supported by OrbexDB can be executed directly:

result = db.execute("SELECT CURRENT_TIMESTAMP AS now")
print(result.first())

Configuration can also be supplied using ORBEXDB_TOKEN, ORBEXDB_USERNAME and ORBEXDB_PASSWORD, ORBEXDB_BASE_URL, and ORBEXDB_DATABASE_ID.