DEV Community

Cover image for How We Built Scraping AI: Turning 500+ Enterprise Projects Into a Self-Serve API
Aman Deep Singh
Aman Deep Singh

Posted on

How We Built Scraping AI: Turning 500+ Enterprise Projects Into a Self-Serve API

#ai #api #django #python

From managed enterprise scraping at PigData to a high-scale developer API in six months.

[!NOTE]
TL;DR / Engineering Retrospective:

  • Origin: PigData delivered 500+ custom enterprise scraping projects (spanning Tier-1 automotive, e-commerce, and mega-bank financial institutions) via managed services before codifying core scraping patterns into a self-serve developer API.
  • Tech Stack: Django REST Framework + Celery + RabbitMQ + PostgreSQL (VersionedModel optimistic locking) + S3 / MinIO storage.
  • Key Innovation: A versioned state-machine pipeline (InputState) powering modular Crawlers, LLM Extractors (OpenAI / Gemini), and BM25 + Vector Rankers.
  • Zero-Risk Trial: Get 200 free tokens (no credit card required) at https://pig-data.jp/service/scraping-ai/.

The Origin Problem

For years, PigData operated as a managed data extraction service in Japan, building bespoke scrapers for enterprise data pipelines. Whether extracting product catalogs or market intelligence, our engineers handled the end-to-end process.

The problem? Every project started from scratch. Even when two clients needed similar data (e.g., e-commerce product listings), we were rebuilding identical parsing logic, browser automation routines, and anti-bot retry loops.

We faced four core engineering bottlenecks:

  1. No economies of scale: Every client required dedicated maintenance hours when target sites updated CSS selectors.
  2. Small developer teams were priced out: Managed agency contracts have high monetary floors.
  3. Developers wanted self-serve APIs: Modern engineering teams don't want a 3-week sales cycle—they want an API key and a 3-line SDK call.
  4. Siloed expertise: Anti-bot bypasses and rate-limiting tricks learned in one project weren't automatically shared across systems.

We needed an architecture capable of running 10 jobs or 10,000 concurrent crawling jobs on the exact same infrastructure.

Architecture Overview

We chose a Python stack centered around Django REST Framework (DRF), Celery, RabbitMQ, and PostgreSQL:

┌─────────────────┐       ┌─────────────────┐       ┌─────────────────┐
│   Django API    │ ─────▶│   RabbitMQ      │ ─────▶│ Celery Workers  │
│   (DRF Layer)   │       │ (Message Queue) │       │ (Distributed)   │
└─────────────────┘       └─────────────────┘       └─────────────────┘
         │                                                   │
         ▼                                                   ▼
┌─────────────────┐                                 ┌─────────────────┐
│ PostgreSQL State│ ◀───────────────────────────────│ S3 / MinIO      │
│ (Optimistic Lock│                                 │ Data Exports    │
└─────────────────┘                                 └─────────────────┘
Enter fullscreen mode Exit fullscreen mode

Why Django & Celery in 2026?

  • Battle-Tested Reliability: Web scraping is heavily asynchronous. Django ORM paired with Celery + RabbitMQ provided proven distributed task queuing out of the box.
  • Admin & Support Tooling: Django's built-in Admin gave our team immediate visual access to monitor stuck tasks, retry failed extraction pipelines, and debug user jobs without building custom internal tooling.
  • Productivity over Hype: Staying in Python allowed us to leverage httpx, BeautifulSoup, pydantic, openai, and google-genai directly without cross-language serialization overhead.

The State Machine Pipeline Design

Every data extraction job follows a predictable lifecycle:

[Keywords / Search Query]
          │
          ▼
   ┌─────────────┐
   │ URL Finder  │ (Discovers link graph up to max_depth)
   └─────────────┘
          │
          ▼
   ┌─────────────┐
   │  Crawler    │ (Fetches HTML via httpx or headless browser)
   └─────────────┘
          │
          ▼
   ┌─────────────┐
   │ AI Ranker   │ (Ranks pages via BM25 + Vector embeddings)
   └─────────────┘
          │
          ▼
   ┌─────────────┐
   │ LLM Extractor│ (Applies JSON Schema via GPT-4o / Gemini)
   └─────────────┘
          │
          ▼
   [Structured JSON / CSV Export]
Enter fullscreen mode Exit fullscreen mode

We codified this workflow into a single state machine backed by our central InputState model:

class InputState(VersionedModel):
    """Central state machine model for an extraction task."""

    # Configuration & Instructions
    base_url = models.URLField(max_length=2048)
    user_instruction = models.TextField()
    schema_instruction = models.TextField()

    # Pipeline Execution State
    site_type = models.CharField(max_length=20, choices=[('general', 'General'), ('ec', 'E-Commerce')])
    auto_flow = models.BooleanField(default=True)
    current_step = models.CharField(max_length=50, choices=PIPELINE_STEPS)

    # Step Status Trackers
    keyword_generator_status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='PENDING')
    url_finder_status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='PENDING')
    url_crawler_status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='PENDING')
    url_ranker_status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='PENDING')
    schema_generator_status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='PENDING')
    extraction_status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='PENDING')
Enter fullscreen mode Exit fullscreen mode

Hard Technical Challenges & Solutions

1. Solving Worker Race Conditions (Optimistic Locking)

With dozens of Celery workers processing URLs concurrently, multiple workers attempted to update InputState status simultaneously, causing lost updates.

Solution: We built optimistic locking into VersionedModel:

class VersionedModel(models.Model):
    version = models.IntegerField(default=0)

    class Meta:
        abstract = True

    def save(self, *args, **kwargs):
        if self.pk:
            affected = self.__class__.objects.filter(
                pk=self.pk, version=self.version
            ).update(version=models.F('version') + 1, **kwargs.get('update_fields_dict', {}))

            if not affected:
                raise ConcurrencyError(f"Version conflict on {self.__class__.__name__} ID {self.pk}")
            self.version += 1
            return
        super().save(*args, **kwargs)
Enter fullscreen mode Exit fullscreen mode

2. High-Throughput Batch Upserts (ConcurrencyManager)

Calling ORM .save() inside loops on 10,000 discovered URLs overwhelmed PostgreSQL. We implemented a custom ConcurrencyManager:

class ConcurrencyManager(models.Manager):
    def bulk_claim_and_create(self, urls_data: list, state_id: int):
        """Batch upserts URLs using PostgreSQL bulk ON CONFLICT handling."""
        existing_urls = set(
            self.filter(input_state_id=state_id, url__in=[u['url'] for u in urls_data])
            .values_list('url', flat=True)
        )
        new_objects = [
            self.model(input_state_id=state_id, url=u['url'], status='PENDING')
            for u in urls_data if u['url'] not in existing_urls
        ]
        self.bulk_create(new_objects, batch_size=1000, ignore_conflicts=True)
Enter fullscreen mode Exit fullscreen mode

3. Fair Ledger-Based Token Accounting

Instead of complex billing per CPU second, we implemented a real-time transactional token ledger:

class TokenLedger(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    amount = models.IntegerField()  # Negative for debits, positive for credits
    action = models.CharField(max_length=50)  # e.g., 'task.start', 'url.extractor'
    balance_after = models.IntegerField()
    timestamp = models.DateTimeField(auto_now_add=True)
Enter fullscreen mode Exit fullscreen mode

What the End-User Developer Experience Looks Like

While our backend handles complex async state machines, celery queues, and token ledgers, developers interact with our official published PyPI package (scraping-ai):

pip install scraping-ai
Enter fullscreen mode Exit fullscreen mode 
from scraping_ai import ScrapingAIClient

client = ScrapingAIClient(api_key="YOUR_API_KEY")

# Extract web data directly in one step (polls until finished)
data = client.extract(
    url="https://example.com/products",
    schema={"title": "string", "price": "number", "in_stock": "boolean"}
)

print(data.results)
Enter fullscreen mode Exit fullscreen mode

Honest Retrospective: What We'd Do Differently

Looking back at our 6-month journey:

  1. Start with v2 API Specs First: Our v1 API bundled too many steps synchronously. Splitting v2 into clean async endpoints made client integration significantly smoother.
  2. Rate-Limiting on Day 1: We initially relied on Celery concurrency limits, but dedicated per-domain rate limiters were necessary to prevent hitting target site blocks.
  3. Transparent Anti-Bot Expectations: Smart Stealth works ~85% of the time, but extreme bot defenses require specialized headless setups. Volunteering these boundaries built stronger trust with developer teams.

Try Scraping AI Free

Stop writing fragile scrapers and fixing broken CSS selectors.

About the Team & Company

Scraping AI (https://pig-data.jp/service/scraping-ai/) is developed and operated by indigodata Inc., an AI venture subsidiary of SMS DataTech Co., Ltd. (Tokyo, Japan). Built upon PigData's track record of 500+ enterprise data extraction projects, Scraping AI provides a self-serve LLM extraction API for developers worldwide.

Top comments (1)

Collapse
 
doushabao profile image
Doushabao •

Great article on scaling from enterprise projects to a self-serve API! I've been building similar aggregation services and found that combining multiple data sources with proper caching really improves reliability. For weather data specifically, we've seen that Open-Meteo provides excellent free coverage without API keys. What's your approach to handling rate limiting across different upstream providers?