Skip to main content

​
Environment Variable Access

When running custom Python functions on Definite, the DEFINITE_API_KEY is automatically injected into your execution environment. You can access it using the standard Python environment variable pattern: 
import os

# Access your API key from the environment
DEFINITE_API_KEY = os.environ["DEFINITE_API_KEY"]

# Use with the Definite SDK
from definite_sdk.client import DefiniteClient
client = DefiniteClient(DEFINITE_API_KEY)
Key Points:
  • The API key is automatically provided - no need to manually set it
  • Access it via os.environ["DEFINITE_API_KEY"] for standard Python compatibility
  • Perfect for use with the Definite SDK or making direct API calls
  • Alternative: You can also access it directly as a global variable DEFINITE_API_KEY
Example with API calls: 
import os
import requests

api_key = os.environ["DEFINITE_API_KEY"]
headers = {"Authorization": f"Bearer {api_key}"}
response = requests.post("https://api.definite.app/v1/query", 
                        json={"sql": "SELECT * FROM users"}, 
                        headers=headers)
This approach follows standard Python environment variable conventions and works seamlessly with existing Python libraries and SDKs.

​
Installation

pip: 
pip install definite-sdk
poetry: 
poetry add definite-sdk

​
Key-Value Store

The SDK can store and update key-value objects that can be called by custom Python scripts hosted on Definite. An example use case might be saving/retrieving replication state for a custom data connector. Example: 
from definite_sdk.client import DefiniteClient
DEFINITE_API_KEY = 'YOUR_API_KEY'

# initialize the client
client =  DefiniteClient(DEFINITE_API_KEY)

# initialize or retrieve an existing key-value store
store = client.get_kv_store('test_store')

# add or update key-value pairs to the store
store['replication_key'] = 'created_at'
store['replication_state'] = '2024-05-20'

# commit the store changes
store.commit()
If you want to view the contents of a store 
from definite_sdk.client import DefiniteClient
DEFINITE_API_KEY = 'YOUR_API_KEY'

client =  DefiniteClient(DEFINITE_API_KEY)

# retrieve an existing store
store = client.get_kv_store('test_store')
print(store)
Which will output a dictionary object like: 
{'replication_state': '2024-05-20', 'replication_key': 'create_date'}

​
SQL Query Execution

Execute SQL queries against your connected database integrations. 
# Initialize the SQL client
sql_client = client.get_sql_client()

# Execute a SQL query
result = sql_client.execute("SELECT * FROM users LIMIT 10")
print(result)

# Execute a SQL query with a specific integration
result = sql_client.execute(
    "SELECT COUNT(*) FROM orders WHERE status = 'completed'",
    integration_id="my_database_integration"
)
print(result)

​
Cube Query Execution

Execute Cube queries for advanced analytics and data modeling. 
# Prepare a Cube query
cube_query = {
    "dimensions": [],
    "measures": ["sales.total_amount"],
    "timeDimensions": [{
        "dimension": "sales.date", 
        "granularity": "month"
    }],
    "limit": 1000
}

# Execute the Cube query
result = sql_client.execute_cube_query(
    cube_query, 
    integration_id="my_cube_integration"
)
print(result)

​
Secret Store

Securely store and retrieve secrets for your integrations. 
# Initialize the secret store
secret_store = client.get_secret_store()
# Or use the alias method
secret_store = client.secret_store()

# Set a secret
secret_store.set_secret("database_password", "my_secure_password")

# Get a secret
password = secret_store.get_secret("database_password")

# List all secrets
secrets = list(secret_store.list_secrets())

​
Integration Management

Manage your data integrations and connections. 
# Initialize the integration store
integration_store = client.get_integration_store()
# Or use the alias method
integration_store = client.integration_store()

# List all integrations
integrations = list(integration_store.list_integrations())

# Get a specific integration
integration = integration_store.get_integration("my_integration")

​
Messaging

Send messages through various channels using the messaging client. 
# Initialize the message client
message_client = client.get_message_client()
# Or use the alias method
message_client = client.message_client()

# Send a Slack message using the unified interface
result = message_client.send_message(
    channel="slack",
    integration_id="your_slack_integration_id",
    to="C0920MVPWFN",  # Slack channel ID
    content="Hello from Definite SDK! 👋"
)

# Send a Slack message with blocks and threading
result = message_client.send_message(
    channel="slack",
    integration_id="your_slack_integration_id",
    to="C0920MVPWFN",
    content="Fallback text",
    blocks=[{
        "type": "section",
        "text": {"type": "mrkdwn", "text": "*Important Update*"}
    }],
    thread_ts="1234567890.123456"  # Reply in thread
)

# Or use the convenience method for Slack
result = message_client.send_slack_message(
    integration_id="your_slack_integration_id",
    channel_id="C0920MVPWFN",
    text="Quick message using the convenience method!",
    blocks=[{
        "type": "section",
        "text": {"type": "mrkdwn", "text": "Message with *rich* _formatting_"}
    }]
)

​
Drive Client

Upload arbitrary bytes to your team’s Definite Drive and get back a gs:// path that’s usable in SQL. The file goes client → GCS directly via a signed URL (the backend only mints the URL), so there’s no local GCS auth, no Cloud Run timeout, and no 32 MB body limit. 
from definite_sdk import DefiniteClient

client = DefiniteClient("YOUR_API_KEY")
drive = client.get_drive_client()

# Persistent — stays in Drive until you delete it
r = drive.write_file(b"id,name\n1,alice\n", path="imports/seed.csv")
print(r.gcs_path)    # gs://def-cube-models-.../{team_id}/drive/imports/seed.csv
print(r.drive_path)  # /home/user/drive/imports/seed.csv   (gcsfuse in sandboxes)

# Temporary — auto-deleted after ttl_days (1..30) via a GCS lifecycle rule
r = drive.write_temporary_file(parquet_bytes, name="events.parquet", ttl_days=7)
print(r.expires_at)  # ISO-8601 deletion timestamp
write_file and write_temporary_file accept bytes, str, a local path (os.PathLike), or any readable file-like object. Large files stream via requests — no chunking logic needed.

​
Load local data into DuckLake

Combine the Drive client with the SQL client to load a local file — an API response, a DataFrame, a generated parquet — into the lake. The ingest SQL runs on the Definite backend against the team’s DuckLake; no local DuckDB required. 
import pandas as pd
from definite_sdk import DefiniteClient

client = DefiniteClient("YOUR_API_KEY")
drive = client.get_drive_client()
sql = client.get_sql_client()

df = pd.DataFrame({"id": [1, 2, 3], "name": ["alice", "bob", "carol"]})

# Materialize to parquet bytes, upload to a temp path, and ingest.
r = drive.write_temporary_file(df.to_parquet(index=False), name="users.parquet", ttl_days=7)

sql.execute(
    f"CREATE OR REPLACE TABLE LAKE.MY_SCHEMA.USERS AS "
    f"SELECT * FROM read_parquet('{r.gcs_path}')"
)

# Query the new table
result = sql.execute("SELECT * FROM LAKE.MY_SCHEMA.USERS")
for row in result["data"]:
    print(row)

​
Legacy: attach_ducklake()

For teams provisioned before April 2026 that still have HMAC keys on their DuckLake integration, client.attach_ducklake() returns SQL that attaches the lake to a local DuckDB connection. This is deprecated and emits a DeprecationWarning — new code should use get_drive_client() + get_sql_client() as shown above. 
import duckdb
from definite_sdk import DefiniteClient, UnsupportedDuckLakeAttachError

client = DefiniteClient("YOUR_API_KEY")

try:
    conn = duckdb.connect()
    conn.execute(client.attach_ducklake())  # deprecated
    df = conn.sql("SELECT * FROM lake.my_schema.users").df()
except UnsupportedDuckLakeAttachError:
    # Teams provisioned after April 2026 don't ship laptop-usable credentials.
    # Use the Drive + SQL pattern above instead.
    raise

​
Source Code

Github repo for the Definite SDK can be found here.