Apify SDK: Finally Got Scraping to Scale

# apify_scraper.py - Production configuration

from apify import Actor, Request
from apify_playwright import PlaywrightCrawlingContext
from typing import Dict
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ProductScraper:
    """
    Production scraper with Apify SDK

    Features:
    - Automatic queue management
    - Proxy rotation
    - Error handling with retries
    - Automatic storage
    """

    def __init__(self):
        self.actor_config = None

    async def run(self):
        """Main entry point"""
        async with Actor:
            self.actor_config = Actor.config

            # Configure crawler
            crawler = self.actor_config.crawler(
                PlaywrightCrawlingContext,
                # Browser configuration
                headless=True,
                browser_type='chromium',
                max_pages_per_crawl=1000,
                # Concurrency
                max_concurrency=10,
                # Proxy configuration
                proxy_configuration={
                    'apifyProxy': {
                        'groups': ['RESIDENTIAL'],  # Use residential proxies
                        'countryCode': 'US'
                    }
                },
                # Session management
                use_session_pool=True,
                session_pool_options={
                    'max_pool_size': 100,
                    'session_options': {
                        'max_usage_count': 10,  # Reuse each session 10 times
                        'error_score_threshold': 0.5  # Abandon after errors
                    }
                },
                # Retry configuration
                retry_on_blocked=True,
                max_request_retries=3
            )

            # Add initial URLs
            await crawler.add_requests([
                'https://example.com/products?page=1',
                'https://example.com/products?page=2',
                'https://example.com/products?page=3'
            ])

            # Run crawler
            await crawler.run(self.handle_page)

    async def handle_page(self, context: PlaywrightCrawlingContext):
        """
        Handle each page

        Args:
            context: Playwright crawling context
        """
        page = context.page
        request: Request = context.request

        try:
            logger.info(f"Processing: {request.url}")

            # Wait for content to load
            await page.wait_for_selector('.product-list', timeout=10000)

            # Extract product data
            products = await page.eval_on_selector_all(
                '.product-item',
                '''elements => elements.map(el => ({
                    id: el.getAttribute('data-id'),
                    name: el.querySelector('.name')?.textContent,
                    price: el.querySelector('.price')?.textContent,
                    in_stock: el.querySelector('.stock')?.textContent === 'In Stock'
                }))'''
            )

            # Add metadata
            for product in products:
                product['url'] = request.url
                product['scraped_at'] = Actor.config.started_at.isoformat()

            # Save to dataset (automatic storage)
            if products:
                await context.push_data(products)
                logger.info(f"Saved {len(products)} products")

            # Find and enqueue next pages
            next_pages = await page.eval_on_selector_all(
                'a.pagination-link[href*="/products?page="]',
                '''elements => elements.map(el => el.href)'''
            )

            for next_page_url in next_pages:
                await context.add_requests(
                    [Request.from_url(next_page_url)]
                )

        except Exception as e:
            logger.error(f"Error processing {request.url}: {e}")
            # Apify will retry this request automatically

# Deployment helper
def deploy_to_apify():
    """
    Deploy scraper to Apify Platform

    Requires:
    - Apify account
    - APIFY_TOKEN environment variable
    """
    import subprocess

    # Login (first time only)
    subprocess.run(['apify', 'login'], check=True)

    # Push to Apify Platform
    subprocess.run(['apify', 'push'], check=True)

    print("Deployed to Apify Platform!")
    print("View at: https://console.apify.com/actors")

# Local development helper
async def run_locally():
    """Run scraper locally for testing"""
    import asyncio

    scraper = ProductScraper()
    await scraper.run()

# Main entry point
if __name__ == '__main__':
    import asyncio

    # Run locally
    asyncio.run(run_locally())

    # Or deploy to Apify
    # deploy_to_apify()

from apify import Actor

async def dataset_operations():
    """Work with Apify datasets"""
    async with Actor:
        # Push individual items
        await Actor.push_data({
            'url': 'https://example.com',
            'title': 'Example'
        })

        # Push multiple items
        await Actor.push_data([
            {'id': 1, 'name': 'Item 1'},
            {'id': 2, 'name': 'Item 2'},
            {'id': 3, 'name': 'Item 3'}
        ])

        # Get dataset info
        dataset = await Actor.open_dataset()
        info = await dataset.get_info()
        print(f"Dataset has {info.item_count} items")

        # Export to various formats
        await dataset.export_to_csv('output.csv')
        await dataset.export_to_json('output.json')
        await dataset.export_to_xml('output.xml')

        # Stream items (for large datasets)
        async for item in dataset.iterate_items():
            process_item(item)

async def key_value_store_operations():
    """Store files, screenshots, etc."""
    async with Actor:
        store = await Actor.open_key_value_store()

        # Store screenshots
        # await page.screenshot(path='screenshot.png')
        # await store.set_value(
        #     'homepage.png',
        #     open('screenshot.png', 'rb'),
        #     content_type='image/png'
        # )

        # Store JSON config
        await store.set_value('config', {
            'last_run': '2026-03-23',
            'pages_processed': 1000
        })

        # Store HTML
        await store.set_value('page.html', '...')

        # Retrieve values
        config = await store.get_value('config')
        print(f"Config: {config}")

# Install Apify CLI
npm install -g apify-cli

# Initialize new actor
apify init my-scraper

# Run locally (with live reload)
apify run

# Test with specific input
echo '{"urls": ["https://example.com"]}' | apify run