Python

This page outlines the steps to use the Python connector cdata-connect-ai with Connect AI. The PEP 249-compliant connector enables querying and writing real-time data via standard SQL through the Connect AI REST API.

See the cdata-connect-ai PyPI page for more information.

Prerequisites

Before you can configure and use Python with Connect AI, you must have the following:

• Python >=3.10
• requests >= 2.28.0
• ijson >= 3.1.0
• A data source in your Connect AI account. See Sources for more information.
• A Personal Access Token (PAT). Generate a PAT on the Settings page. Copy this down, as it acts as your password during authentication.

Install the Connector

pip install cdata-connect-ai

For pandas integration: install the connector with the full extra to enable pandas support.

pip install "cdata-connect-ai[full]"

Connect to Connect AI

The following is an example of Python code connecting to the Connect AI API and performing a query on the Account table in a Salesforce connection. Provide your Connect AI username and password (the password is the PAT you created in the prerequisites).

import cdata_connect_ai

conn = cdata_connect_ai.connect(
    username="[e-mail]",
    password="<your_personal_access_token>",
)

with conn.cursor() as cur:
    cur.execute("SELECT * FROM [Salesforce1].[Salesforce].[Account]")
    rows = cur.fetchall()

for row in rows:
    print(row)

conn.close()

Configuration File

Use a PyHOCON configuration file to keep credentials out of code:

# config.conf
cdata_api_db {
  username = "[e-mail]"
  password = "<your_personal_access_token>"
}

The code now refers to the configuration file and does not contain credentials:

import cdata_connect_ai

conn = cdata_connect_ai.connect(config_path="config.conf")

Connect to a Workspace

To connect to a Connect AI workspace, add a workspace parameter as follows:

conn = cdata_connect_ai.connect(
    username="[e-mail]",
    password="<your_personal_access_token>",
    workspace="production",
)

Examples

Parameterized Queries

with conn.cursor() as cur:
    cur.execute(
        "SELECT * FROM [DB].[public].[users] WHERE city = %(city)s LIMIT %(limit)s",
        {"city": "New York", "limit": 10},
    )
    rows = cur.fetchall()

Batch Operations

with conn.cursor() as cur:
    cur.executemany(
        "INSERT INTO [DB].[public].[cities] (city, id) VALUES (@city, @id)",
        [
            {"@city": {"dataType": 5, "value": "New York"}, "@id": {"dataType": 8, "value": 1}},
            {"@city": {"dataType": 5, "value": "London"},   "@id": {"dataType": 8, "value": 2}},
        ],
    )

See Data Types for dataType values and descriptions.

Stored Procedures

with conn.cursor() as cur:
    cur.callproc("[DB].[public].[my_procedure]", ("arg1", "arg2"))
    rows = cur.fetchall()

Pandas Integration

import cdata_connect_ai
import pandas as pd

conn = cdata_connect_ai.connect(
    username="[e-mail]",
    password="<your_personal_access_token>",
)

with conn.cursor() as cur:
    cur.execute("SELECT Id, Name, Industry, AnnualRevenue FROM [Salesforce1].[Salesforce].[Account]")
    columns = [desc[0] for desc in cur.description]
    rows = cur.fetchall()

df = pd.DataFrame(rows, columns=columns)
print(df.head())
print(df.describe())

conn.close()

Connection Options

The following table contains the connection options available for the connector.

Parameter	Description	Default
username	Authentication username	-
password	Personal access token	-
config_path	Path to PyHOCON config file	-
workspace	Connect AI workspace name	-
timeout	HTTP request timeout (seconds)	90
max_retries	Retries on transient 5xx errors	3
retry_delay	Base delay between retries (seconds)	1.0