REST API

pip install alterlab

from alterlab import AlterLabSync

# Use AlterLabSync for synchronous code (no await needed)
with AlterLabSync(api_key="YOUR_API_KEY") as client:
    result = client.scrape("https://example.com")
    print(result["content"])
    print(f"Cost: {result['billing']['total_credits']}")

    # With schema filtering
    result = client.scrape(
        url="https://amazon.com/dp/B0123456789",
        extraction_schema={
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "price": {"type": "number"},
                "in_stock": {"type": "boolean"}
            }
        }
    )
    product = result["filtered_content"]  # Your filtered data
    print(f"Product: {product['name']} - ${product['price']}")

import asyncio
from alterlab import AlterLab

async def main():
    # AlterLab is async - requires await
    async with AlterLab(api_key="YOUR_API_KEY") as client:
        result = await client.scrape("https://example.com")
        print(result["content"])

        # With JS rendering
        result = await client.scrape(
            url="https://spa-app.com",
            mode="js",
            advanced=AdvancedOptions(render_js=True, screenshot=True)
        )

asyncio.run(main())

from alterlab import AlterLabSync, AdvancedOptions, CostControls

with AlterLabSync(api_key="YOUR_API_KEY") as client:
    # With advanced options
    result = client.scrape(
        url="https://example.com",
        mode="js",
        advanced=AdvancedOptions(
            render_js=True,
            screenshot=True,
            markdown=True,
            wait_condition="networkidle"
        ),
        cost_controls=CostControls(
            max_credits=10,
            max_tier="4",
            prefer_cost=True
        )
    )

    # Screenshot saved to result["screenshot_url"]
    # Markdown in result["content"]["markdown"] if formats specified

from alterlab import AlterLabSync, AdvancedOptions

with AlterLabSync(api_key="YOUR_API_KEY") as client:
    # Estimate cost before scraping
    estimate = client.estimate_cost(
        url="https://example.com",
        mode="auto",
        advanced=AdvancedOptions(render_js=True, screenshot=True)
    )

    print(f"Estimated credits: {estimate['estimated_credits']}")
    print(f"Max possible: {estimate['max_possible_credits']}")

    # Only scrape if within budget
    if estimate['estimated_credits'] <= 5:
        result = client.scrape(url="https://example.com")

from alterlab import AlterLabSync, AlterLabAPIError, AlterLabTimeoutError

with AlterLabSync(api_key="YOUR_API_KEY") as client:
    try:
        result = client.scrape("https://example.com")
    except AlterLabAPIError as e:
        if e.status_code == 402:
            print("Insufficient balance!")
        elif e.status_code == 429:
            print(f"Rate limited. Retry after {e.retry_after}s")
        else:
            print(f"API error: {e.detail}")
    except AlterLabTimeoutError:
        print("Request timed out")
    except Exception as e:
        print(f"Unexpected error: {e}")

# Get job status without waiting
job_id = "550e8400-e29b-41d4-a716-446655440000"
status = client.get_job_status(job_id)
print(f"Job status: {status['status']}")

# Poll job with custom settings
result = client.poll_job(
    job_id=job_id,
    poll_interval=2.0,  # Check every 2 seconds
    poll_timeout=300.0   # Give up after 5 minutes
)

# Custom configuration (sync client)
client = AlterLabSync(
    api_key="YOUR_API_KEY",
    base_url="https://api.alterlab.io",  # API base (paths include /api/v1)
    timeout=60.0,  # Request timeout
    max_retries=3,  # Retry failed requests
    retry_backoff=2.0  # Exponential backoff multiplier
)

# Or use environment variable
# export ALTERLAB_API_KEY="YOUR_API_KEY"
client = AlterLabSync()  # Auto-loads from env

# Extract product data with schema
result = client.scrape(
    url="https://amazon.com/dp/B0123456789",
    mode="auto",
    extraction_schema={
        "type": "object",
        "properties": {
            "name": {"type": "string"},
            "price": {"type": "number"},
            "currency": {"type": "string"},
            "in_stock": {"type": "boolean"},
            "rating": {"type": "number"},
            "reviews_count": {"type": "integer"}
        }
    }
)

# Access filtered data (matches your schema)
product = result["filtered_content"]
print(f"Name: {product['name']}")
print(f"Price: {product['price']}")
print(f"In Stock: {product['in_stock']}")

{
  "url": "https://amazon.com/dp/B0123456789",
  "status_code": 200,
  "content": { ... },           // Full extraction (Schema.org, metadata, etc.)
  "filtered_content": {         // YOUR filtered data (only when extraction_schema provided)
    "name": "Product Name",
    "price": 29.99,
    "in_stock": true
  },
  "billing": { "total_credits": 3 }
}

{
  "url": "https://example.com",
  "mode": "auto",
  "cost_controls": {
    "max_credits": 0.004,
    "max_tier": "4",
    "prefer_cost": true,
    "fail_fast": false
  }
}

// Response billing breakdown:
{
  "billing": {
    "total_cost": 0.002,
    "tier_used": "3",
    "escalations": [
      {"tier": "1", "result": "failed", "cost": 0.0002, "duration_ms": 250, "error": "403 Forbidden"},
      {"tier": "2", "result": "failed", "cost": 0.0003, "duration_ms": 2100, "error": "Blocked by WAF"},
      {"tier": "3", "result": "success", "cost": 0.002, "duration_ms": 4200}
    ],
    "optimization_suggestion": "Site requires Stealth tier. Consider using prefer_speed with Tier 4 for faster results."
  }
}

// Connect with API key authentication
const ws = new WebSocket('wss://api.alterlab.io/api/v1/ws/jobs?api_key=sk_live_...');

// Handle connection open
ws.onopen = () => {
  console.log('WebSocket connected');

  // Subscribe to job updates
  ws.send(JSON.stringify({
    action: 'subscribe',
    job_id: '550e8400-e29b-41d4-a716-446655440000'
  }));
};

// Receive real-time updates
ws.onmessage = (event) => {
  const message = JSON.parse(event.data);

  switch (message.type) {
    case 'connected':
      console.log('✓ Connection established');
      break;

    case 'subscribed':
      console.log('✓ Subscribed to job:', message.job_id);
      break;

    case 'job_update':
      console.log('Job status:', message.status);

      if (message.status === 'completed') {
        console.log('✓ Job completed:', message.result);
        ws.close();
      } else if (message.status === 'failed') {
        console.error('✗ Job failed:', message.error);
        ws.close();
      } else {
        console.log('⟳ Job in progress...');
      }
      break;

    case 'heartbeat':
      // Server is alive
      break;

    case 'error':
      console.error('WebSocket error:', message.message);
      break;
  }
};

ws.onerror = (error) => {
  console.error('WebSocket connection error:', error);
};

ws.onclose = () => {
  console.log('WebSocket disconnected');
};

import asyncio
import json
import websockets

async def watch_job(api_key: str, job_id: str):
    uri = f"wss://api.alterlab.io/api/v1/ws/jobs?api_key={api_key}"

    async with websockets.connect(uri) as ws:
        # Subscribe to job
        await ws.send(json.dumps({
            "action": "subscribe",
            "job_id": job_id
        }))

        # Listen for updates
        async for message in ws:
            data = json.loads(message)

            if data["type"] == "job_update":
                status = data["status"]
                print(f"Job status: {status}")

                if status == "completed":
                    print("Job completed:", data["result"])
                    break
                elif status == "failed":
                    print("Job failed:", data["error"])
                    break

# Usage
asyncio.run(watch_job("sk_live_...", "550e8400-..."))

curl -X POST https://api.alterlab.io/api/v1/scrape \
  -H "X-API-Key: sk_live_..." \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://example.com",
    "location": {
      "country": "DE",
      "language": "de"
    }
  }'

from alterlab import AlterLabSync

client = AlterLabSync(api_key="YOUR_API_KEY")

# Scrape as if browsing from Germany in German
result = client.scrape(
    url="https://example.com",
    location={"country": "DE", "language": "de"}
)

print(result["content"])

# First request: fetches fresh content and caches it
curl -X POST https://api.alterlab.io/api/v1/scrape \
  -H "X-API-Key: sk_live_..." \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://example.com",
    "cache": true,
    "cache_ttl": 7200
  }'

# Second request within TTL: returns cached result (no credit charge)
curl -X POST https://api.alterlab.io/api/v1/scrape \
  -H "X-API-Key: sk_live_..." \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://example.com",
    "cache": true
  }'

from alterlab import AlterLabSync

client = AlterLabSync(api_key="YOUR_API_KEY")

# Enable caching with a 2-hour TTL
result = client.scrape(
    url="https://example.com",
    cache=True,
    cache_ttl=7200
)

print(f"Cached: {result.get('cached', False)}")