Ingest Documents and Data

// BadRequestError class
{
  name: "BadRequestError",
  status: 400,
  message: "400 Bad Request",
  error: {
    message: "Invalid request parameters",
    details: {
      content: "Content cannot be empty",
      customId: "customId exceeds maximum length"
    }
  }
}

// RateLimitError class
{
  name: "RateLimitError",
  status: 429,  // NOT 402!
  message: "429 Too Many Requests",
  error: {
    message: "Rate limit exceeded",
    retry_after: 60
  }
}

// NotFoundError class
{
  name: "NotFoundError",
  status: 404,
  message: "404 Not Found",
  error: {
    message: "Memory not found",
    resource_id: "invalid_memory_id"
  }
}

// PermissionDeniedError class
{
  name: "PermissionDeniedError",
  status: 403,
  message: "403 Forbidden",
  error: {
    message: "Insufficient permissions",
    required_permission: "memories:write"
  }
}

// InternalServerError class
{
  name: "InternalServerError",
  status: 500,
  message: "500 Internal Server Error",
  error: {
    message: "Processing failed",
    details: "Content extraction service unavailable"
  }
}

    // APIConnectionError class - NEW
  {
    name: "APIConnectionError",
    message: "Connection error.",
    cause: Error // Original network error
  }

  // APIConnectionTimeoutError class - NEW
  {
    name: "APIConnectionTimeoutError",
    message: "Request timed out."
  }

    import asyncio
import time
import logging
from typing import List, Dict, Any, Optional
from supermemory import Supermemory, BadRequestError, RateLimitError

async def batch_ingest(
    documents: List[Dict[str, Any]],
    options: Optional[Dict[str, Any]] = None
):
    options = options or {}
    batch_size = options.get('batch_size', 5)  # CORRECTED: Conservative size
    delay_between_batches = options.get('delay_between_batches', 2.0)  # CORRECTED: 2 seconds
    max_retries = options.get('max_retries', 3)

    results = []

    for i in range(0, len(documents), batch_size):
        batch = documents[i:i + batch_size]
        batch_num = i // batch_size + 1
        total_batches = (len(documents) + batch_size - 1) // batch_size

        print(f"Processing batch {batch_num}/{total_batches}")

        # Process batch with proper error handling
        tasks = [ingest_with_retry(doc, max_retries) for doc in batch]
        batch_results = await asyncio.gather(*tasks, return_exceptions=True)

        results.extend(batch_results)

        # Rate limiting between batches
        if i + batch_size < len(documents):
            await asyncio.sleep(delay_between_batches)

    return results

async def ingest_with_retry(doc: Dict[str, Any], max_retries: int):
    for attempt in range(1, max_retries + 1):
        try:
            return await client.memories.add(
                content=doc['content'],
                custom_id=doc['id'],
                container_tags=["batch_import_user_123"],  # CORRECTED: List
                metadata={
                    "source": "migration",
                    "batch_id": generate_batch_id(),
                    "original_created": doc.get('created_at', ''),
                    "title": doc.get('title', ''),
                    **doc.get('metadata', {})
                }
            )
        except BadRequestError as e:
            logging.error(f"Invalid document {doc['id']}: {e}")
            raise  # Don't retry validation errors

        except RateLimitError as e:
            logging.warning(f"Rate limited on attempt {attempt}")
            delay = 2 ** attempt * 2  # Longer delays for rate limits
            await asyncio.sleep(delay)
            continue

        except Exception as error:
            if attempt == max_retries:
                raise error

            # Exponential backoff
            delay = 2 ** attempt
            logging.info(f"Retry {attempt}/{max_retries} for {doc['id']} in {delay}s")
            await asyncio.sleep(delay)

def generate_batch_id() -> str:
    import random
    import string
    return f"batch_{int(time.time())}_{random.choices(string.ascii_lowercase, k=8)}"