Python Web Scraping
Reading layout

Writing Scrapy Item Pipelines

A spider that opens a database connection is a spider you cannot test, and this article — part of Web Scraping with Scrapy — covers the component that exists to keep that code out of your parse methods: the item pipeline, its exact contract, and the ordering and lifecycle rules that decide whether a pipeline chain is fast or quietly lossy.

Scrapy item pipeline chain ordered by priority number A spider yields an item into a validation pipeline at priority 100, a normalization pipeline at 200 and a Postgres writer at 800. Raising DropItem in the first stage removes the record before it reaches storage. ITEM_PIPELINES: the lower number runs firstSpideryields itemValidatepriority 100required fieldsNormalizepriority 200prices, datesPostgrespriority 800batched INSERTTabledurable rowsraise DropItemdiscarded, stat counted
Every yielded item walks the chain in ascending priority order. Returning the item passes it on; raising DropItem removes it from the chain entirely and nothing downstream ever sees it.

A pipeline is an ordinary Python class with a process_item(self, item, spider) method that returns the item, returns an awaitable, or raises DropItem. Scrapy calls every enabled pipeline in ascending order of the integer you assign it in ITEM_PIPELINES, passing each one the object the previous pipeline returned. Everything else — connections, buffers, counters — hangs off the two lifecycle hooks open_spider and close_spider. Get those three methods right and the rest is domain logic.

The process_item Contract in Detail

The signature is fixed and the return value is load-bearing. Whatever process_item returns becomes the input to the next pipeline, and the object returned by the last pipeline is what feeds the feed exporters. Returning None is a bug that Scrapy will not warn you about in older versions: the item silently becomes None for everything downstream, and a feed export produces empty rows.

# pipelines.py
from itemadapter import ItemAdapter
from scrapy.exceptions import DropItem

class RequiredFieldsPipeline:
    """Reject records that are missing anything downstream depends on."""

    REQUIRED = ("url", "title", "price")

    def __init__(self) -> None:
        self.dropped = 0

    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        missing = [f for f in self.REQUIRED if not adapter.get(f)]
        if missing:
            self.dropped += 1
            raise DropItem(f"missing {', '.join(missing)} for {adapter.get('url', '?')}")
        return item          # returning the item is mandatory

    def close_spider(self, spider) -> None:
        spider.logger.info("RequiredFieldsPipeline dropped %d items", self.dropped)

Two things are worth copying from this. The first is ItemAdapter, which wraps scrapy.Item, dataclass, attrs, pydantic models and plain dicts behind one interface, so the pipeline works regardless of what the spider yields. Writing item["price"] directly locks the pipeline to dict-like items and raises KeyError on a missing field where adapter.get returns None.

The second is that DropItem carries a message. It is logged at WARNING and increments the item_dropped_count stat, so a run that quietly discards 40% of its records is visible in the stats dictionary at the end rather than only in the row count. That number is one of the more useful things to ship to a dashboard, as described in Exporting Scrapy Metrics to Prometheus.

Raising anything other than DropItem behaves differently: the exception is logged as an error, the item is discarded, and the crawl continues, but no drop stat is recorded. A TypeError from a careless float() call therefore looks like data loss with no accounting, which is why parsing that can fail belongs behind an explicit try that converts the failure into a DropItem.

Ordering: The Priority Map Is a Sort Key, Not an Address

ITEM_PIPELINES maps a dotted path to an integer between 0 and 1000. The integers are sorted; nothing else about them matters, and gaps are the point. Numbering stages 100, 200, 300 leaves room to insert a stage at 150 later without renumbering everything.

# settings.py
ITEM_PIPELINES = {
    "bookstore.pipelines.RequiredFieldsPipeline": 100,
    "bookstore.pipelines.NormalizePipeline": 200,
    "bookstore.pipelines.DuplicateUrlPipeline": 300,
    "bookstore.pipelines.PostgresPipeline": 800,
}

The ordering rule that actually matters is cost. Every stage before a DropItem runs on records that will be thrown away, so the cheapest and most selective checks belong first. Validating required fields at 100 costs a dictionary lookup; discovering the same record is unusable inside the database writer at 800 means it has already been normalised, deduplicated, and possibly serialised. On a crawl that drops a third of its records, moving validation from last to first removes a third of the work from every other stage.

Storage belongs last for the same reason, and deduplication belongs immediately before it. A common mistake is putting a seen-URL filter at 100 to "save work" — it saves nothing, because the record has already been fetched and parsed by then. Its value is preventing a duplicate write, so it only has to run before the writer.

Scrapy also disables a pipeline if you assign it None in a per-spider custom_settings block, which is the clean way to let one spider skip the database writer without a conditional inside the pipeline itself.

Lifecycle Hooks: from_crawler, open_spider, close_spider

Connections do not belong in __init__, and settings do not belong in module-level constants. Scrapy instantiates a pipeline through from_crawler if that classmethod exists, which is where settings, stats, and signal access come from.

Timeline of open_spider, process_item and close_spider A crawl timeline with three marked points. open_spider opens the database connection once, process_item validates and buffers each record, and close_spider flushes the final partial batch and closes the connection. One crawl, three pipeline momentsopen_spiderprocess_itemclose_spiderRuns once, firstopen the connectioncreate tables if absentreset the row bufferRuns per itemvalidate, then cleanbuffer, flush at 500return item or drop itRuns once, lastflush the last batchcommit and closelog the final counts
A pipeline has three moments, not one. Connection setup happens once at the start, the per-item work happens thousands of times, and the final flush happens once at the end — putting work in the wrong moment is the usual cause of slow or lossy pipelines.
# pipelines.py
import psycopg

class PostgresPipeline:
    def __init__(self, dsn: str, batch_size: int) -> None:
        self.dsn = dsn
        self.batch_size = batch_size
        self.conn: psycopg.Connection | None = None
        self.buffer: list[tuple] = []

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            dsn=crawler.settings.get("POSTGRES_DSN"),
            batch_size=crawler.settings.getint("POSTGRES_BATCH_SIZE", 500),
        )

    def open_spider(self, spider) -> None:
        self.conn = psycopg.connect(self.dsn, autocommit=False)
        with self.conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS books (
                    url   TEXT PRIMARY KEY,
                    title TEXT NOT NULL,
                    price NUMERIC(10, 2),
                    seen  TIMESTAMPTZ DEFAULT now()
                )
            """)
        self.conn.commit()

    def close_spider(self, spider) -> None:
        self._flush()                    # the final partial batch
        if self.conn is not None:
            self.conn.close()

open_spider runs once, after the engine starts and before the first request completes. close_spider runs once when the engine finishes, including when the crawl is stopped by CLOSESPIDER_ITEMCOUNT or a KeyboardInterrupt that Scrapy handles gracefully. It does not run on SIGKILL, which is the argument for flushing on a size threshold rather than relying entirely on the final flush.

Both hooks receive the spider, so a pipeline can behave differently per spider — writing to a table named after spider.name, for instance — without a registry of special cases.

Batching Writes Instead of One INSERT per Item

The default shape of a storage pipeline is one INSERT and one commit() inside process_item. It is simple, it is correct, and it is roughly an order of magnitude slower than it needs to be, because each commit is a network round trip plus a synchronous flush of the write-ahead log.

Insert throughput for per-item commits, batched executemany and COPY A three-bar comparison. Committing after every item reaches about 180 rows per second, executemany over batches of 500 reaches about 6,400, and a server-side COPY from a buffer reaches about 12,000. Write strategyRows inserted per secondcommit() per itemone round trip per row180executemany, batch 500one round trip per batch6,400COPY from a bufferserver-side bulk load12,000
Rows per second inserted from a pipeline into a local PostgreSQL 16 instance. The cost is dominated by round trips and commits, not by the rows themselves, so batching is worth roughly an order of magnitude.

Buffering rows and writing them with executemany amortises both costs across the batch.

    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        self.buffer.append((adapter["url"], adapter["title"], adapter["price"]))
        if len(self.buffer) >= self.batch_size:
            self._flush()
        return item

    def _flush(self) -> None:
        if not self.buffer or self.conn is None:
            return
        with self.conn.cursor() as cur:
            cur.executemany(
                """
                INSERT INTO books (url, title, price)
                VALUES (%s, %s, %s)
                ON CONFLICT (url) DO UPDATE
                    SET title = EXCLUDED.title,
                        price = EXCLUDED.price,
                        seen  = now()
                """,
                self.buffer,
            )
        self.conn.commit()
        self.buffer.clear()

Batch size trades throughput against loss on a hard crash. Five hundred rows is a reasonable default: large enough that the per-commit overhead disappears into the noise, small enough that an abrupt termination costs seconds of work rather than minutes. The ON CONFLICT clause makes the write idempotent, which is what allows a resumed or overlapping crawl to be re-run without producing duplicate rows — the same property discussed for other sinks in Saving Scraped Data to PostgreSQL.

One caveat about psycopg specifically: executemany in psycopg 3 pipelines the statements rather than concatenating them, which is why it is fast. In psycopg 2 it loops in Python and is barely quicker than individual calls, so a project still on the older driver should use execute_values instead.

Blocking, Deferreds, and Async Pipelines

Scrapy runs on Twisted, in a single thread. A synchronous conn.commit() inside process_item blocks the reactor, which means no requests are sent, no responses are processed, and the downloader idles for the duration. At 500-row batches and a 40 ms commit, that is a 40 ms stall roughly every few seconds — invisible. At one commit per item it is 40 ms on every single record, and the crawl runs at 25 items per second no matter how you set CONCURRENT_REQUESTS.

Scrapy 2.0 and later allow process_item to be a coroutine, provided the project sets the asyncio reactor:

# settings.py
TWISTED_REACTOR = "twisted.internet.asyncioreactor.AsyncioSelectorReactor"

# pipelines.py
import psycopg
from itemadapter import ItemAdapter

class AsyncPostgresPipeline:
    def __init__(self, dsn: str, batch_size: int = 500) -> None:
        self.dsn = dsn
        self.batch_size = batch_size
        self.conn: psycopg.AsyncConnection | None = None
        self.buffer: list[tuple] = []

    @classmethod
    def from_crawler(cls, crawler):
        return cls(crawler.settings.get("POSTGRES_DSN"))

    async def open_spider(self, spider) -> None:
        self.conn = await psycopg.AsyncConnection.connect(self.dsn, autocommit=False)

    async def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        self.buffer.append((adapter["url"], adapter["title"], adapter["price"]))
        if len(self.buffer) >= self.batch_size:
            await self._flush()
        return item

    async def _flush(self) -> None:
        if not self.buffer or self.conn is None:
            return
        async with self.conn.cursor() as cur:
            await cur.executemany(
                "INSERT INTO books (url, title, price) VALUES (%s, %s, %s) "
                "ON CONFLICT (url) DO NOTHING",
                self.buffer,
            )
        await self.conn.commit()
        self.buffer.clear()

    async def close_spider(self, spider) -> None:
        await self._flush()
        if self.conn is not None:
            await self.conn.close()

The async version does not make the database faster; it makes the waiting free, because the reactor keeps downloading while the write is in flight. Note that open_spider and close_spider may also be coroutines under the asyncio reactor, which is what makes an async connection possible at all — there is no way to await a connect from a synchronous hook.

If the blocking library has no async equivalent, the middle path is twisted.internet.threads.deferToThread, which moves the call to a thread pool and returns a Deferred that Scrapy will await. That keeps the reactor free at the cost of ordinary thread-safety concerns around the connection object.

Edge Cases and Caveats

  • A pipeline instance is shared across the whole crawl, not per spider run in the same process. Running two spiders concurrently in one CrawlerProcess gives each its own instance, but state kept in self still spans every item of that crawl. Reset counters in open_spider, not in __init__.
  • DropItem from a later stage cannot un-write an earlier one. If a stage at 300 writes to Redis and a stage at 800 drops the record, the Redis key remains. Order stages so that anything with a side effect runs after every check that can reject.
  • Feed exporters are not pipelines. -o books.csv writes whatever the last pipeline returned, and it runs even if no pipelines are enabled. A drop in the chain removes the row from the export; an exception inside a feed exporter does not touch the pipelines.
  • Items yielded from errback handlers still traverse the chain. A pipeline that assumes every item came from a successful parse will see partially-populated records from error paths, which is another reason validation belongs first.
  • close_spider runs even when the crawl fails. If open_spider raised before the connection was assigned, close_spider will still be called with self.conn is None. Guard it, or the real error is buried under an AttributeError from the teardown.
  • Validation logic is worth sharing with non-Scrapy code. Keeping the schema in a Pydantic model and calling it from the pipeline means the same rules apply to a backfill script; see Validating Scraped Data with Pydantic.

Frequently Asked Questions

Why does my item come out empty at the end of the chain? Almost always because a process_item fell off the end without returning the item. Every branch of the method must return item or raise; a missing return sends None to the next stage, which then fails or silently produces an empty row in the export.

How do I write to two destinations at once? Use two pipelines with different priorities, each owning its own connection, rather than one pipeline with two sinks. Failures then stay isolated, and disabling one destination for a particular spider is a single None entry in that spider's custom_settings.

Should deduplication live in a pipeline or in the scheduler? Both, for different things. Scrapy's built-in duplicate filter stops the same URL being requested twice; a pipeline filter stops the same record being stored twice when two different URLs yield identical content. Only the second one needs to see the item.

Can a pipeline yield new requests? No. process_item returns an item or raises, and the engine does not schedule anything it returns. If a record needs an extra fetch to complete, do it in the spider with a follow-up request and a callback that merges the fields, or use scrapy.Request from a downloader middleware.

Back to home