​
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_"}
    }]
)

​
Source Code

Github repo for the Definite SDK can be found here.