#!/usr/bin/env python3
"""
SmartScraper Pagination Example (Async)
This example demonstrates how to use pagination functionality with SmartScraper API using the asynchronous client.
"""
import asyncio
import json
import logging
import os
import time
from pydantic import BaseModel
from typing import List, Optional
from dotenv import load_dotenv
from scrapegraph_py import AsyncClient
from scrapegraph_py.exceptions import APIError
# Load environment variables from .env file
load_dotenv()
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
handlers=[logging.StreamHandler()],
)
logger = logging.getLogger(__name__)
class ProductInfo(BaseModel):
"""Schema for product information"""
name: str
price: Optional[str] = None
rating: Optional[str] = None
image_url: Optional[str] = None
description: Optional[str] = None
class ProductList(BaseModel):
"""Schema for list of products"""
products: List[ProductInfo]
async def smartscraper_pagination_example():
"""Example of using pagination with SmartScraper (async)"""
print("SmartScraper Pagination Example (Async)")
print("=" * 50)
# Initialize client from environment variable
api_key = os.getenv("SGAI_API_KEY")
if not api_key:
print("❌ Error: SGAI_API_KEY environment variable not set")
return
try:
client = AsyncClient(api_key=api_key)
except Exception as e:
print(f"❌ Error initializing client: {e}")
return
# Configuration
website_url = "https://www.amazon.in/s?k=tv&crid=1TEF1ZFVLU8R8&sprefix=t%2Caps%2C390&ref=nb_sb_noss_2"
user_prompt = "Extract all product info including name, price, rating, image_url, and description"
total_pages = 3 # Number of pages to scrape
print(f"🌐 Website URL: {website_url}")
print(f"📝 User Prompt: {user_prompt}")
print(f"📄 Total Pages: {total_pages}")
print("-" * 50)
try:
# Start timing
start_time = time.time()
# Make the request with pagination
result = await client.smartscraper(
user_prompt=user_prompt,
website_url=website_url,
output_schema=ProductList,
total_pages=total_pages
)
# Calculate duration
duration = time.time() - start_time
print(f"✅ Request completed in {duration:.2f} seconds")
print(f"📊 Response type: {type(result)}")
# Display results
if isinstance(result, dict):
print("\n🔍 Response:")
print(json.dumps(result, indent=2, ensure_ascii=False))
# Check for pagination success indicators
if "data" in result:
print(f"\n✨ Pagination successful! Data extracted from {total_pages} pages")
elif isinstance(result, list):
print(f"\n✅ Pagination successful! Extracted {len(result)} items")
for i, item in enumerate(result[:5]): # Show first 5 items
print(f" {i+1}. {item}")
if len(result) > 5:
print(f" ... and {len(result) - 5} more items")
else:
print(f"\n📋 Result: {result}")
except APIError as e:
print(f"❌ API Error: {e}")
except Exception as e:
print(f"❌ Unexpected error: {e}")
async def test_concurrent_pagination():
"""Test multiple pagination requests concurrently"""
print("\n" + "=" * 50)
print("Testing concurrent pagination requests")
print("=" * 50)
api_key = os.getenv("SGAI_API_KEY")
if not api_key:
print("❌ Error: SGAI_API_KEY environment variable not set")
return
try:
client = AsyncClient(api_key=api_key)
except Exception as e:
print(f"❌ Error initializing client: {e}")
return
# Test concurrent requests
urls = [
"https://example.com/products?page=1",
"https://example.com/products?page=2",
"https://example.com/products?page=3",
]
tasks = []
for i, url in enumerate(urls):
print(f"🚀 Creating task {i+1} for URL: {url}")
tasks.append(asyncio.create_task(
simulate_pagination_request(client, url, i+1)
))
print(f"⏱️ Starting {len(tasks)} concurrent tasks...")
start_time = time.time()
try:
results = await asyncio.gather(*tasks, return_exceptions=True)
duration = time.time() - start_time
print(f"✅ All tasks completed in {duration:.2f} seconds")
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"❌ Task {i+1} failed: {result}")
else:
print(f"✅ Task {i+1} succeeded: {result}")
except Exception as e:
print(f"❌ Concurrent execution failed: {e}")
async def simulate_pagination_request(client: AsyncClient, url: str, task_id: int):
"""Simulate a pagination request (for demonstration)"""
print(f"📋 Task {task_id}: Processing {url}")
# Simulate some work
await asyncio.sleep(0.5)
# Return a simulated result
return f"Task {task_id} completed successfully"
async def main():
"""Main function to run the pagination examples"""
print("ScrapeGraph SDK - SmartScraper Pagination Examples (Async)")
print("=" * 60)
# Run the main example
await smartscraper_pagination_example()
# Test concurrent pagination
await test_concurrent_pagination()
print("\n" + "=" * 60)
print("Examples completed!")
print("\nNext steps:")
print("1. Set SGAI_API_KEY environment variable")
print("2. Replace example URLs with real websites")
print("3. Adjust total_pages parameter (1-10)")
print("4. Customize user_prompt for your use case")
print("5. Define output_schema for structured data")
if __name__ == "__main__":
asyncio.run(main())
