Scrapy Item Pipelines Explained

How Pipelines Work

When a spider yields an item (a dictionary or Item object), Scrapy passes it through every enabled pipeline in priority order. Each pipeline class implements a process_item method that receives the item and the spider that produced it. The method must either return the item (passing it to the next pipeline) or raise a DropItem exception (discarding it from further processing).

Pipeline priority is defined by integer values in your settings.py. Lower numbers execute first. A typical configuration looks like this:

ITEM_PIPELINES = {
    "myproject.pipelines.ValidationPipeline": 100,
    "myproject.pipelines.CleaningPipeline": 200,
    "myproject.pipelines.DuplicatesPipeline": 300,
    "myproject.pipelines.DatabasePipeline": 400,
}

This ordering matters because you want validation first (reject malformed items early), then cleaning (normalize formats), then deduplication (discard items already seen), and finally storage (write clean, unique items to your database). Each stage can assume the previous stages have already run.

Pipelines also support lifecycle hooks. The open_spider method runs when a spider starts, letting you initialize resources like database connections or file handles. The close_spider method runs when a spider finishes, letting you flush buffers, close connections, and perform cleanup. These hooks ensure resources are properly managed regardless of how the crawl ends.

The Pipeline Interface

Every pipeline class follows a consistent interface. Here is the complete set of methods a pipeline can implement:

from scrapy.exceptions import DropItem

class ExamplePipeline:
    def open_spider(self, spider):
        """Called when spider opens. Initialize resources here."""
        pass

    def close_spider(self, spider):
        """Called when spider closes. Cleanup resources here."""
        pass

    def process_item(self, item, spider):
        """Process each item. Must return item or raise DropItem."""
        return item

    @classmethod
    def from_crawler(cls, crawler):
        """Factory method for accessing settings and signals."""
        return cls()

The from_crawler class method is optional but powerful. It receives the Crawler object, giving your pipeline access to project settings, signals, and statistics. Use it when your pipeline needs configuration values from settings.py:

@classmethod
def from_crawler(cls, crawler):
    return cls(
        db_url=crawler.settings.get("DATABASE_URL"),
        batch_size=crawler.settings.getint("PIPELINE_BATCH_SIZE", 100),
    )

Validation Pipeline

The first pipeline stage should validate that items contain required fields and that values meet expected formats. Dropping invalid items early prevents garbage data from reaching your database and makes debugging easier since validation errors appear in logs at the point of extraction.

from scrapy.exceptions import DropItem

class ValidationPipeline:
    required_fields = ["name", "price", "url"]

    def process_item(self, item, spider):
        # Check required fields exist and are non-empty
        for field in self.required_fields:
            if not item.get(field):
                raise DropItem(
                    f"Missing required field '{field}' in {item.get('url', 'unknown')}"
                )

        # Validate price is numeric
        try:
            price_str = item["price"].replace("$", "").replace(",", "")
            float(price_str)
        except (ValueError, AttributeError):
            raise DropItem(f"Invalid price format: {item['price']}")

        return item

When a pipeline raises DropItem, Scrapy logs the message and increments the item_dropped_count statistic. The item never reaches subsequent pipelines. This is the correct way to filter items, not by returning None or silently ignoring them.

Cleaning Pipeline

After validation confirms items have the right structure, a cleaning pipeline normalizes values into consistent formats. Common cleaning tasks include stripping whitespace, removing HTML entities, normalizing price formats, converting dates, and standardizing text fields.

import re
from datetime import datetime

class CleaningPipeline:
    def process_item(self, item, spider):
        # Strip whitespace from all string fields
        for field, value in item.items():
            if isinstance(value, str):
                item[field] = value.strip()

        # Normalize price to float
        if "price" in item:
            price_str = re.sub(r"[^\d.]", "", item["price"])
            item["price"] = float(price_str)

        # Clean description HTML
        if "description" in item:
            item["description"] = re.sub(r"<[^>]+>", "", item["description"])
            item["description"] = re.sub(r"\s+", " ", item["description"]).strip()

        # Normalize URL to absolute form
        if "url" in item and not item["url"].startswith("http"):
            item["url"] = f"https://example.com{item['url']}"

        return item

Keep cleaning pipelines focused on format normalization rather than business logic. If a value is wrong enough to warrant dropping the item, that belongs in validation. Cleaning transforms valid but messy data into a consistent output format.

Deduplication Pipeline

Many crawls encounter the same item multiple times, especially when pages are reachable through different navigation paths. A deduplication pipeline tracks seen items by a unique identifier and drops duplicates before they reach storage.

from scrapy.exceptions import DropItem

class DuplicatesPipeline:
    def open_spider(self, spider):
        self.seen_ids = set()

    def process_item(self, item, spider):
        # Use URL as unique identifier
        item_id = item.get("url") or item.get("sku")
        if not item_id:
            return item  # Cannot deduplicate without identifier

        if item_id in self.seen_ids:
            raise DropItem(f"Duplicate item: {item_id}")

        self.seen_ids.add(item_id)
        return item

For crawls that run repeatedly (daily price monitoring, for example), in-memory deduplication only catches duplicates within a single run. Cross-run deduplication requires checking against your database in the storage pipeline or maintaining a persistent fingerprint store between crawl sessions.

Storage Pipelines

Storage pipelines write validated, cleaned, deduplicated items to their final destination. Common targets include SQL databases, MongoDB, Elasticsearch, CSV files, and cloud storage. Scrapy supports multiple simultaneous storage pipelines, so you can write to both a database and a backup file in the same crawl.

Here is a PostgreSQL storage pipeline using batch inserts for performance:

import psycopg2

class PostgresPipeline:
    def __init__(self, db_url, batch_size):
        self.db_url = db_url
        self.batch_size = batch_size
        self.items_buffer = []

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            db_url=crawler.settings.get("DATABASE_URL"),
            batch_size=crawler.settings.getint("DB_BATCH_SIZE", 50),
        )

    def open_spider(self, spider):
        self.conn = psycopg2.connect(self.db_url)
        self.cursor = self.conn.cursor()

    def close_spider(self, spider):
        # Flush remaining items
        if self.items_buffer:
            self._flush_buffer()
        self.conn.close()

    def process_item(self, item, spider):
        self.items_buffer.append(item)
        if len(self.items_buffer) >= self.batch_size:
            self._flush_buffer()
        return item

    def _flush_buffer(self):
        for item in self.items_buffer:
            self.cursor.execute(
                "INSERT INTO products (name, price, url) VALUES (%s, %s, %s) "
                "ON CONFLICT (url) DO UPDATE SET price = EXCLUDED.price",
                (item["name"], item["price"], item["url"]),
            )
        self.conn.commit()
        self.items_buffer = []

Batch inserts significantly improve database performance by reducing round trips. The ON CONFLICT clause handles cases where a product already exists, updating the price instead of failing. The close_spider hook ensures any items remaining in the buffer are written even if the crawl ends before a full batch accumulates.

JSON and File Storage

While Scrapy's built-in feed exports handle basic file output, custom file pipelines give you more control over format, rotation, and compression:

import json

class JsonLinesPipeline:
    def open_spider(self, spider):
        self.file = open(f"output/{spider.name}.jsonl", "a")

    def close_spider(self, spider):
        self.file.close()

    def process_item(self, item, spider):
        line = json.dumps(dict(item), ensure_ascii=False)
        self.file.write(line + "\n")
        return item

JSON Lines format (one JSON object per line) is preferred over a JSON array for large crawls because you can append to the file without loading and rewriting the entire contents. It also allows line-by-line processing with standard Unix tools and streaming parsers.

Pipeline Best Practices

Keep each pipeline focused on one concern. A pipeline that validates, cleans, and stores is doing too much. Split it into separate classes that compose through the priority ordering system. This makes each component testable, reusable, and independently configurable.

Use from_crawler for configuration. Never hardcode database URLs, file paths, or thresholds in pipeline code. Pull these from settings so you can change them per environment (development, staging, production) without modifying pipeline source code.

Handle exceptions gracefully in storage pipelines. Database connection failures, disk full errors, and API rate limits can all occur during storage. Implement retry logic or dead-letter queues for items that fail to store rather than crashing the entire crawl.

Buffer writes for performance. Individual inserts to databases are slow. Collect items into batches and flush periodically. Choose a batch size that balances memory usage against write efficiency, typically between 50 and 500 items depending on your database and item size.

Log pipeline statistics. Track items processed, items dropped (with reasons), storage successes, and storage failures. These metrics help you monitor crawl health and identify extraction problems. Scrapy's stats collector integrates cleanly with pipeline methods for this purpose.

Test pipelines independently. Since each pipeline has a simple interface (process_item takes an item and returns an item or raises DropItem), you can unit test pipeline logic without running a full crawl. Create test items and verify that your pipeline produces the expected output or raises the expected exceptions.