Skip to main content

Python Integration

A complete guide to integrating the Kanva API with your Python application.

Installation​

Install the required packages:

pip install requests

Basic Client​

Here's a simple client class to interact with the Kanva API:

import requests
from typing import Dict, Any, Optional

class KanvaClient:
    def __init__(self, api_key: str, project_id: str, base_url: str = "https://kanva.human-driven.ai/api/v1"):
        self.api_key = api_key
        self.project_id = project_id
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            "X-API-Key": api_key,
            "Content-Type": "application/json"
        })

    def predict_sync(self, input_data: Dict[str, list], timeout: int = 60) -> Dict[str, Any]:
        """Make a synchronous prediction."""
        response = self.session.post(
            f"{self.base_url}/projects/{self.project_id}/predict",
            json={"input": input_data},
            params={"sync": "true", "timeout": str(timeout)}
        )
        response.raise_for_status()
        return response.json()

    def predict_async(self, input_data: Dict[str, list]) -> str:
        """Start an async prediction, returns job ID."""
        response = self.session.post(
            f"{self.base_url}/projects/{self.project_id}/predict",
            json={"input": input_data}
        )
        response.raise_for_status()
        return response.json()["data"]["jobId"]

    def get_job_status(self, job_id: str) -> Dict[str, Any]:
        """Check the status of an async job."""
        response = self.session.get(
            f"{self.base_url}/projects/{self.project_id}/jobs/{job_id}"
        )
        response.raise_for_status()
        return response.json()

Usage Examples​

Synchronous Prediction​

client = KanvaClient(
    api_key="your_api_key",
    project_id="your_project_id"
)

result = client.predict_sync({
    "date": ["2024-01-15"],
    "category": ["A"],
    "value": [123.45]
})

print(f"Prediction: {result['data']['predictions']}")

Async with Polling​

import time

# Start the job
job_id = client.predict_async({
    "date": ["2024-01-15", "2024-01-16"],
    "category": ["A", "B"],
    "value": [100, 200]
})

# Poll for completion
while True:
    status = client.get_job_status(job_id)

    if status["data"]["status"] == "completed":
        print(f"Results: {status['data']['result']['predictions']}")
        break
    elif status["data"]["status"] == "failed":
        print(f"Error: {status['data']['error']}")
        break

    time.sleep(2)

Using with Pandas​

import pandas as pd

def predict_dataframe(client: KanvaClient, df: pd.DataFrame) -> pd.DataFrame:
    """Add predictions to a DataFrame."""
    input_data = {col: df[col].tolist() for col in df.columns}
    result = client.predict_sync(input_data)
    df = df.copy()
    df["prediction"] = result["data"]["predictions"]
    return df

# Usage
df = pd.read_csv("data.csv")
df_with_predictions = predict_dataframe(client, df)

Next Steps​