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

> Source: <https://dev.to/amandeep-sms/how-we-built-scraping-ai-turning-500-enterprise-projects-into-a-self-serve-api-oii>
> Published: 2026-09-15 01:27:15+00:00

**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/](https://pig-data.jp/service/scraping-ai/).

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:

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

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    │
└─────────────────┘                                 └─────────────────┘
```

`httpx`, `BeautifulSoup`, `pydantic`, `openai`, and `google-genai` directly without cross-language serialization overhead.
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]
```

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')
```

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)
```

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

``` python
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)
```

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)
```

While our backend handles complex async state machines, celery queues, and token ledgers, developers interact with our official published PyPI package ([`scraping-ai`](https://pypi.org/project/scraping-ai/)):

```
pip install scraping-ai
python
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)
```

Looking back at our 6-month journey:

Stop writing fragile scrapers and fixing broken CSS selectors.

`https://pypi.org/project/scraping-ai/`
**Scraping AI** ([https://pig-data.jp/service/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.