Definite SDK

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)

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)

pip install definite-sdk

poetry add definite-sdk

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()

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)

{'replication_state': '2024-05-20', 'replication_key': 'create_date'}

# 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)

# 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)

# 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())

# 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")

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

import duckdb
from definite_sdk import DefiniteClient

# Initialize the client
client = DefiniteClient("YOUR_API_KEY")

# Connect to DuckDB and attach DuckLake
conn = duckdb.connect()
conn.execute(client.attach_ducklake())

# Now you can use DuckLake tables
conn.execute("CREATE SCHEMA IF NOT EXISTS lake.my_schema;")
conn.execute("CREATE OR REPLACE TABLE lake.my_schema.users AS SELECT * FROM df")

# Query your DuckLake data
result = conn.sql("SELECT * FROM lake.my_schema.users").df()

# Attach with custom alias
conn.execute(client.attach_ducklake(alias="warehouse"))

# Use the custom alias
conn.execute("SELECT * FROM warehouse.my_schema.users")