The Python SDK wraps the AiQL API for Python 3.9 and later.

​
Prerequisites

  • Python 3.9 or later
  • An AiQL API key from the Dashboard

​
Install

pip install aiql

​
Authenticate

Create a client with your API key. Read the key from an environment variable instead of hardcoding it. 
import os
from aiql import AiQL

client = AiQL(api_key=os.environ["AIQL_API_KEY"])
To target a different environment, set the base URL: 
client = AiQL(
    api_key=os.environ["AIQL_API_KEY"],
    base_url="http://127.0.0.1:8000",
)

​
Workspaces

workspace = client.workspaces.create(name="Analytics")

workspaces = client.workspaces.list()

workspace = client.workspaces.get(workspace.id)

workspace = client.workspaces.update(workspace.id, name="Analytics EU")

client.workspaces.delete(workspace.id)

​
Ingestions

ingestion = client.ingestions.create(
    workspace_id=workspace.id,
    source="s3://my-bucket/events.jsonl",
)

ingestion = client.ingestions.get(ingestion.id)

client.ingestions.cancel(ingestion.id)

client.ingestions.retry(ingestion.id)

​
Queries

query = client.queries.create(
    workspace_id=workspace.id,
    text="count events grouped by day",
)

result = client.queries.get(query.id)

​
Tools

converted = client.tools.convert(data=raw_records, to="parquet")

chunks = client.tools.split(data=document, max_tokens=512)

shaped = client.tools.shape(data=records, schema=target_schema)

fields = client.tools.extract(data=document, fields=["name", "email"])

​
Error handling

The SDK raises AiQLError for failed requests. Inspect status_code and message to handle specific cases. 
from aiql import AiQL, AiQLError

client = AiQL(api_key=os.environ["AIQL_API_KEY"])

try:
    client.queries.get("missing-id")
except AiQLError as error:
    print(error.status_code, error.message)