Skip to main content

SDK Examples

The Qarion SDK provides typed, high-level clients for Python (with both async and sync variants) and a lightweight JavaScript client. This page walks through common usage patterns, from basic setup to advanced patterns like batch processing and error handling wrappers.

Python Client​

Installation​

Install the SDK from PyPI:

pip install qarion-sdk

Basic Usage (Async)​

The async client is the primary interface and is recommended for most use cases. It uses httpx under the hood and supports Python's native async/await syntax:

from qarion import QarionClient

async def main():
    client = QarionClient(
        base_url="https://api.qarion.com",
        api_key="your-key"
    )
    
    # List products in a space
    products = await client.products.list(space_slug="analytics")
    
    for product in products.items:
        print(f"{product.name}: {product.product_type}")
    
    await client.close()

The client manages HTTP connections, authentication headers, and response parsing automatically. All resource methods return typed Pydantic models, so you get IDE autocompletion and type checking out of the box.

Synchronous Client​

For scripts, notebooks, and environments where async/await isn't practical, the SDK also provides a synchronous client with an identical API surface:

from qarion import QarionSyncClient

client = QarionSyncClient(
    base_url="https://api.qarion.com",
    api_key="your-key"
)

products = client.products.list(space_slug="analytics")
for product in products.items:
    print(f"{product.name}: {product.product_type}")

client.close()

The sync client wraps the same underlying logic in a synchronous interface, so you can use it in Jupyter notebooks, CLI tools, and other contexts where async support is limited.

Context Managers​

Both clients support context manager syntax, which ensures the HTTP connection is properly closed when you're done — even if an exception occurs:

# Async
async with QarionClient(base_url="...", api_key="...") as client:
    products = await client.products.list(space_slug="analytics")

# Sync
with QarionSyncClient(base_url="...", api_key="...") as client:
    products = client.products.list(space_slug="analytics")

Creating Products​

The SDK handles request serialization and response parsing, so creating resources is as simple as calling a method with keyword arguments:

from qarion import QarionClient

async def register_product():
    async with QarionClient(base_url="...", api_key="...") as client:
        product = await client.products.create(
            space_slug="analytics",
            name="Customer Metrics",
            product_type="table",
            description="Aggregated customer behavior metrics",
            hosting_location="warehouse.analytics.customer_metrics"
        )
        
        print(f"Created: {product.name} (ID: {product.id})")

The create method returns the newly created product as a typed model, including all server-generated fields like id, created_at, and slug.

JavaScript Client​

For JavaScript and TypeScript environments, you can use a lightweight client built on the Fetch API. While less feature-rich than the Python SDK, it provides a clean wrapper around authentication and common operations:

class QarionClient {
  constructor({ baseUrl, apiKey }) {
    this.baseUrl = baseUrl;
    this.headers = {
      'Authorization': `Bearer ${apiKey}`,
      'Content-Type': 'application/json'
    };
  }

  async listProducts(spaceSlug) {
    const response = await fetch(
      `${this.baseUrl}/catalog/spaces/${spaceSlug}/products`,
      { headers: this.headers }
    );
    return response.json();
  }

  async getProduct(spaceSlug, productId) {
    const response = await fetch(
      `${this.baseUrl}/catalog/spaces/${spaceSlug}/products/${productId}`,
      { headers: this.headers }
    );
    return response.json();
  }

  async createProduct(spaceSlug, data) {
    const response = await fetch(
      `${this.baseUrl}/catalog/spaces/${spaceSlug}/products`,
      {
        method: 'POST',
        headers: this.headers,
        body: JSON.stringify(data)
      }
    );
    return response.json();
  }
}
// Usage
const client = new QarionClient({
  baseUrl: 'https://api.qarion.com',
  apiKey: 'your-key'
});

const products = await client.listProducts('analytics');

Common Patterns​

Pagination Helper​

When working with paginated endpoints, a helper function that iterates through all pages saves you from repeating the pagination loop in every call site:

async def get_all_items(client, resource, **params):
    """Fetch all items across multiple pages."""
    all_items = []
    page = 1
    
    while True:
        result = await resource.list(page=page, size=100, **params)
        all_items.extend(result.items)
        
        if page * 100 >= result.total:
            break
        page += 1
    
    return all_items

# Usage
all_products = await get_all_items(client, client.products, space_slug="analytics")

Batch Processing​

For operations that need to process many items — such as syncing metadata from an external source into Qarion — a batch processor with concurrency control prevents you from overwhelming the API:

import asyncio

async def batch_process(items, process_func, batch_size=10):
    """Process items in batches with controlled concurrency."""
    results = []
    
    for i in range(0, len(items), batch_size):
        batch = items[i:i + batch_size]
        batch_results = await asyncio.gather(
            *[process_func(item) for item in batch]
        )
        results.extend(batch_results)
    
    return results

# Usage
async def update_product(product_data):
    return await client.products.update(
        space_slug="analytics",
        product_id=product_data["id"],
        description=product_data["description"]
    )

results = await batch_process(products_to_update, update_product)

By processing items in batches of 10 (or whatever batch size suits your rate limits), you get the performance benefits of concurrency without exceeding the API's rate limits.

Error Handling Wrapper​

A centralized error handling wrapper transforms raw HTTP errors into typed exceptions that are easier to handle in application code:

from qarion.exceptions import (
    QarionError,
    AuthenticationError,
    NotFoundError,
    RateLimitError,
    ValidationError
)

async def safe_api_call(func, *args, **kwargs):
    """Wrap API calls with structured error handling."""
    try:
        return await func(*args, **kwargs)
    except AuthenticationError:
        print("Re-authenticate and retry")
        raise
    except NotFoundError as e:
        print(f"Resource not found: {e}")
        return None
    except RateLimitError as e:
        print(f"Rate limited, wait {e.retry_after}s")
        await asyncio.sleep(e.retry_after)
        return await func(*args, **kwargs)
    except ValidationError as e:
        print(f"Validation failed: {e.errors}")
        raise
    except QarionError as e:
        print(f"API error: {e}")
        raise

The SDK's built-in exception hierarchy maps HTTP status codes to specific exception types, so you can catch and handle different failure modes separately rather than inspecting status codes manually.