Skip to main content

​
Full CRUD Example

import requests

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://catalogapi.rastro.ai/api"

headers = {"Authorization": f"Bearer {API_KEY}"}

# 1. Get your catalog ID from the dashboard or list catalogs
catalogs = requests.get(f"{BASE_URL}/public/catalogs", headers=headers).json()
catalog_id = catalogs["catalogs"][0]["id"]

# 2. Add items
items = [
    {"sku": "BEARING-001", "name": "6205-2RS Ball Bearing", "price": 12.99},
    {"sku": "BEARING-002", "name": "6206-2RS Ball Bearing", "price": 15.99},
    {"sku": "BEARING-003", "name": "6207-2RS Ball Bearing", "price": 18.99}
]

result = requests.post(
    f"{BASE_URL}/public/catalogs/{catalog_id}/items/bulk",
    headers=headers,
    json={"items": items}
).json()

print(f"Added {result['items_created']} items")

# 3. List items
items = requests.get(
    f"{BASE_URL}/public/catalogs/{catalog_id}/items",
    headers=headers
).json()

for item in items["items"]:
    print(f"  {item['sku']}: {item['name']} - ${item['price']}")

# 4. Update an item
item_id = items["items"][0]["id"]
updated = requests.put(
    f"{BASE_URL}/public/catalogs/{catalog_id}/items/{item_id}",
    headers=headers,
    json={"price": 10.99}
).json()

print(f"Updated {updated['sku']} price to ${updated['price']}")

# 5. Get single item
item = requests.get(
    f"{BASE_URL}/public/catalogs/{catalog_id}/items/{item_id}",
    headers=headers
).json()

print(f"Item: {item}")

​
Catalog Enrichment with Polling

Use POST /public/enrich with catalog_id to enrich catalog items. The catalog’s schema is applied automatically. 
import requests
import time

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://catalogapi.rastro.ai/api"

headers = {"Authorization": f"Bearer {API_KEY}"}

def enrich_catalog(catalog_id, items, speed="slow"):
    # Start enrichment using the catalog's schema
    response = requests.post(
        f"{BASE_URL}/public/enrich",
        headers=headers,
        json={
            "catalog_id": catalog_id,
            "items": items,
            "async_mode": True,
            "speed": speed
        }
    )
    job = response.json()
    job_id = job["job_id"]
    print(f"Started enrichment job: {job_id}")
    print(f"Total items: {job['total_items']}")

    # Poll for completion
    while True:
        status = requests.get(
            f"{BASE_URL}/public/enrich/{job_id}",
            headers=headers
        ).json()

        completed = status.get("completed_items", 0)
        total = status.get("total_items", 0)

        print(f"Progress: {completed}/{total}")

        if status["status"] != "running":
            return status

        time.sleep(10)


# Example usage
catalog_id = "cat_abc123"

items = [
    {"part_number": "6205-2RS", "name": "Deep Groove Ball Bearing"},
    {"part_number": "6206-2RS", "name": "Deep Groove Ball Bearing"},
    {"part_number": "6207-2RS", "name": "Deep Groove Ball Bearing"}
]

result = enrich_catalog(catalog_id, items)
print(f"\nEnrichment {result['status']}")
print(f"Completed: {result['completed_items']}")

​
Full Python Client

import requests
import time

class CatalogClient:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://catalogapi.rastro.ai/api"

    def _headers(self):
        return {"Authorization": f"Bearer {self.api_key}"}

    # Catalog operations
    def list_catalogs(self):
        response = requests.get(
            f"{self.base_url}/public/catalogs",
            headers=self._headers()
        )
        return response.json()["catalogs"]

    # Item operations
    def list_items(self, catalog_id, page=1, limit=100):
        response = requests.get(
            f"{self.base_url}/public/catalogs/{catalog_id}/items",
            headers=self._headers(),
            params={"page": page, "limit": limit}
        )
        return response.json()

    def get_item(self, catalog_id, item_id):
        response = requests.get(
            f"{self.base_url}/public/catalogs/{catalog_id}/items/{item_id}",
            headers=self._headers()
        )
        return response.json()

    def update_item(self, catalog_id, item_id, data):
        response = requests.put(
            f"{self.base_url}/public/catalogs/{catalog_id}/items/{item_id}",
            headers=self._headers(),
            json=data
        )
        return response.json()

    def bulk_upsert(self, catalog_id, items):
        response = requests.post(
            f"{self.base_url}/public/catalogs/{catalog_id}/items/bulk",
            headers=self._headers(),
            json={"items": items}
        )
        return response.json()

    # Enrichment (uses the Enrich API with catalog_id)
    def start_enrichment(self, catalog_id, items, speed="slow", **kwargs):
        response = requests.post(
            f"{self.base_url}/public/enrich",
            headers=self._headers(),
            json={"catalog_id": catalog_id, "items": items, "async_mode": True, "speed": speed, **kwargs}
        )
        return response.json()

    def get_enrichment_status(self, job_id):
        response = requests.get(
            f"{self.base_url}/public/enrich/{job_id}",
            headers=self._headers()
        )
        return response.json()

    def enrich_and_wait(self, catalog_id, items, poll_interval=10, **kwargs):
        job = self.start_enrichment(catalog_id, items, **kwargs)
        job_id = job["job_id"]
        print(f"Started job: {job_id}")

        while True:
            status = self.get_enrichment_status(job_id)
            print(f"Progress: {status.get('completed_items', 0)}/{status.get('total_items', '?')}")

            if status["status"] != "running":
                return status

            time.sleep(poll_interval)


# Example usage
client = CatalogClient("YOUR_API_KEY")

# Get catalog ID from dashboard or list
catalogs = client.list_catalogs()
catalog_id = catalogs[0]["id"]

# Add items
items = [
    {"sku": "6205-2RS", "name": "Deep Groove Ball Bearing"},
    {"sku": "6206-2RS", "name": "Deep Groove Ball Bearing"},
    {"sku": "6207-2RS", "name": "Deep Groove Ball Bearing"}
]
client.bulk_upsert(catalog_id, items)

# Enrich and wait
result = client.enrich_and_wait(
    catalog_id,
    items=items,
    speed="slow"
)

# List enriched items
catalog_items = client.list_items(catalog_id)
for item in catalog_items["items"]:
    print(f"{item['sku']}: bore={item.get('bore_diameter', 'N/A')}")

​
Full TypeScript Client

const API_KEY = "YOUR_API_KEY";
const BASE_URL = "https://catalogapi.rastro.ai/api";

class CatalogClient {
  private apiKey: string;

  constructor(apiKey: string) {
    this.apiKey = apiKey;
  }

  private headers() {
    return {
      "Authorization": `Bearer ${this.apiKey}`,
      "Content-Type": "application/json"
    };
  }

  // Catalog operations
  async listCatalogs() {
    const response = await fetch(`${BASE_URL}/public/catalogs`, {
      headers: this.headers()
    });
    const data = await response.json();
    return data.catalogs;
  }

  // Item operations
  async listItems(catalogId: string, page = 1, limit = 100) {
    const response = await fetch(
      `${BASE_URL}/public/catalogs/${catalogId}/items?page=${page}&limit=${limit}`,
      { headers: this.headers() }
    );
    return response.json();
  }

  async getItem(catalogId: string, itemId: string) {
    const response = await fetch(
      `${BASE_URL}/public/catalogs/${catalogId}/items/${itemId}`,
      { headers: this.headers() }
    );
    return response.json();
  }

  async updateItem(catalogId: string, itemId: string, data: object) {
    const response = await fetch(
      `${BASE_URL}/public/catalogs/${catalogId}/items/${itemId}`,
      {
        method: "PUT",
        headers: this.headers(),
        body: JSON.stringify(data)
      }
    );
    return response.json();
  }

  async bulkUpsert(catalogId: string, items: object[]) {
    const response = await fetch(
      `${BASE_URL}/public/catalogs/${catalogId}/items/bulk`,
      {
        method: "POST",
        headers: this.headers(),
        body: JSON.stringify({ items, source_info: { source: "api" } })
      }
    );
    return response.json();
  }

  // Enrichment (uses the Enrich API with catalog_id)
  async startEnrichment(catalogId: string, items: object[], options: object = {}) {
    const response = await fetch(
      `${BASE_URL}/public/enrich`,
      {
        method: "POST",
        headers: this.headers(),
        body: JSON.stringify({ catalog_id: catalogId, items, async_mode: true, ...options })
      }
    );
    return response.json();
  }

  async getEnrichmentStatus(jobId: string) {
    const response = await fetch(
      `${BASE_URL}/public/enrich/${jobId}`,
      { headers: this.headers() }
    );
    return response.json();
  }

  async enrichAndWait(
    catalogId: string,
    items: object[],
    options: object = {},
    pollInterval = 10000
  ) {
    const job = await this.startEnrichment(catalogId, items, options);
    const jobId = job.job_id;
    console.log(`Started job: ${jobId}`);

    while (true) {
      const status = await this.getEnrichmentStatus(jobId);
      console.log(`Progress: ${status.completed_items || 0}/${status.total_items || "?"}`);

      if (status.status !== "running") {
        return status;
      }

      await new Promise(r => setTimeout(r, pollInterval));
    }
  }
}

// Example usage
const client = new CatalogClient("YOUR_API_KEY");

// Get catalog ID from dashboard or list
const catalogs = await client.listCatalogs();
const catalogId = catalogs[0].id;

// Add items
const items = [
  { sku: "6205-2RS", name: "Deep Groove Ball Bearing" },
  { sku: "6206-2RS", name: "Deep Groove Ball Bearing" }
];
await client.bulkUpsert(catalogId, items);

// Enrich and wait
const result = await client.enrichAndWait(
  catalogId,
  items,
  { speed: "slow" }
);

// List enriched items
const enrichedItems = await client.listItems(catalogId);
for (const item of enrichedItems.items) {
  console.log(`${item.sku}: bore=${item.bore_diameter || "N/A"}`);
}