opendatalabs-query-sdk 0.1.1

A Python SDK for interacting with the Query API service

Query API SDK

A Python SDK for interacting with Vana's Query API service. This SDK allows you to easily query and transform your AI-generated data, including posts, tweets, and other content generated by your AI models.

Features

• 🐍 Full Python type hints
• 🔒 Built-in authentication
• 🔄 Async-style API
• 📊 Data transformation support
• 🔔 Webhook integration
• 📝 Comprehensive typing
• ⏱️ Polling utilities for long-running queries

Installation

pip install query-api-sdk
# or
poetry add query-api-sdk

Quick Start

from query_api_sdk import create_client, QueryClientConfig

client = create_client(QueryClientConfig(
    api_key='your-api-key',
    base_url='https://api.vana.ai/query'
))

# Submit a query and get results
def get_my_posts():
    try:
        query_id = client.submit_query({
            "query": "SELECT * FROM reddit_posts WHERE file_owner = 'my_user_id'"
        })
        
        results = client.wait_for_results(query_id)
        print(results)
    except Exception as error:
        print(f"Error: {str(error)}")

Available Data Schemas

The Query API provides access to various Vana-generated content types:

Reddit Posts

reddit_posts {
  file_owner: string     -- User ID of the post owner
  post_id: integer       -- Unique identifier for the post
  title: string         -- Post title
  content: string       -- Post content
}

Twitter Tweets

twitter_tweets {
  file_owner: string    -- User ID of the tweet owner
  tweet_id: integer     -- Unique identifier for the tweet
  text: string         -- Tweet content
}

Detailed Usage

Configuration

from query_api_sdk import create_client, QueryClientConfig

client = create_client(QueryClientConfig(
    api_key='your-api-key',
    base_url='https://api.vana.ai/query',
    timeout=30000  # Optional: default is 10000ms
))

Getting Available Schemas

schemas = client.get_schemas()
print(schemas)

Submitting Queries

Basic query:

query_id = client.submit_query({
    "query": "SELECT * FROM reddit_posts LIMIT 10"
})

With data transformation:

query_id = client.submit_query({
    "query": "SELECT * FROM reddit_posts",
    "transform": """
    def transform(rows):
        return [{**row, "word_count": len(row["content"].split())} for row in rows]
    """
})

With webhook notification:

query_id = client.submit_query({
    "query": "SELECT * FROM twitter_tweets",
    "webhook_url": "https://your-server.com/webhook"
})

Checking Query Status

status = client.get_query_status(query_id)
print(status["status"])  # 'queued' | 'processing' | 'completed' | 'failed'

Getting Results

With pagination:

results = client.get_query_results(
    query_id,
    limit=100,
    cursor="200"
)

Waiting for completion:

results = client.wait_for_results(
    query_id,
    timeout=300000,      # Optional: max time to wait (default: 5 minutes)
    poll_interval=1000   # Optional: time between status checks (default: 1 second)
)

Common Query Examples

Getting Recent Posts

query_id = client.submit_query({
    "query": """
        SELECT *
        FROM reddit_posts
        WHERE file_owner = 'your_user_id'
        ORDER BY post_id DESC
        LIMIT 10
    """
})

Analyzing Content Length

query_id = client.submit_query({
    "query": """
        SELECT *
        FROM reddit_posts
        WHERE file_owner = 'your_user_id'
    """,
    "transform": """
    def transform(rows):
        return [{
            **row,
            "content_length": len(row["content"]),
            "word_count": len(row["content"].split())
        } for row in rows]
    """
})

Combining Data Sources

query_id = client.submit_query({
    "query": """
        SELECT 
            'reddit' as source,
            title as content,
            post_id as id
        FROM reddit_posts
        WHERE file_owner = 'your_user_id'
        UNION ALL
        SELECT 
            'twitter' as source,
            text as content,
            tweet_id as id
        FROM twitter_tweets
        WHERE file_owner = 'your_user_id'
    """
})

Error Handling

The SDK uses a custom QueryAPIError class for error handling:

from query_api_sdk import QueryAPIError

try:
    results = client.get_query_results("invalid-id")
except QueryAPIError as error:
    print(f"API Error: {str(error)}")
    print(f"Status Code: {error.status_code}")
    print(f"Response: {error.response}")

Webhook Integration

When providing a webhook URL, your endpoint will receive POST requests with the following format:

{
    "query_id": str,
    "status": str,  # 'completed' | 'failed'
    "error": Optional[str]
}

Example webhook handler (Flask):

from flask import Flask, request

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    data = request.json
    query_id = data["query_id"]
    status = data["status"]
    error = data.get("error")
    
    if status == "completed":
        # Handle completion
        pass
    elif status == "failed":
        # Handle failure
        pass
    
    return "", 200

Rate Limiting

The Query API implements rate limiting to ensure fair usage. The SDK will automatically handle rate limit responses by raising a QueryAPIError with the appropriate status code and message.

Type Hints Support

The SDK is written with full Python type hints and provides comprehensive type definitions for all features. You can import types directly:

from query_api_sdk import (
    QueryStatusType,
    Schema,
    QueryRequest,
    QueryResults
)

Development

For development, clone the repository and install dependencies:

git clone https://github.com/vana-com/query-sdk-python.git
cd query-sdk-python
pip install -e ".[dev]"

Run tests:

pytest

License

MIT License - see LICENSE for details.