feat(spiders): Add XML/CSV feed templates + docs

agent-skill/Scrapling-Skill/SKILL.md

@@ -326,6 +326,8 @@ class BlogCrawler(CrawlSpider):
 ```
 For sitemap-driven crawls, use `SitemapSpider` with the same `rules()` API. It fetches `sitemap_urls`, descends into sitemap indexes, and dispatches each URL through your rules. Put a `robots.txt` URL directly in `sitemap_urls` and the spider extracts each `Sitemap:` directive from it automatically. See `references/spiders/generic-templates.md` for the full reference, including `LinkExtractor`'s allow/deny/restrict_css/canonicalize options.
 
+For XML feeds (RSS, Atom, product feeds), use `XMLFeedSpider`: set `itertag` to the node name and override `parse_node(response, node)`, which receives each matching node as a namespace-stripped `lxml` element (`node.findtext("title")`). For CSV feeds, use `CSVFeedSpider`: override `parse_row(response, row)`, which receives each row as a dictionary, with `headers`/`delimiter`/`quotechar` for non-standard feeds. Both decompress gzipped feeds automatically. See `references/spiders/generic-templates.md`.
+
 For Shopify-powered stores, subclass `ShopifySpider` and set `target_website` to the store's domain; it extracts every product variant through Shopify's JSON API without touching the HTML. See `references/spiders/platform-templates.md`.
 
 ### Advanced Parsing & Navigation

agent-skill/Scrapling-Skill/references/spiders/generic-templates.md

@@ -121,6 +121,76 @@ class MySitemap(SitemapSpider):
 
 Set `sitemap_alternate_links = True` to also dispatch `<xhtml:link rel="alternate" hreflang="...">` URLs through your rules.
 
+## XMLFeedSpider
+
+`XMLFeedSpider` iterates over the nodes of an XML feed (RSS, Atom, product feeds, etc.). Set `itertag` to the node name you want (default: `"item"`) and override `parse_node()`, which is called once per matching node:
+
+```python
+from scrapling.spiders import XMLFeedSpider
+
+class RSSSpider(XMLFeedSpider):
+    name = "rss"
+    start_urls = ["https://example.com/feed.xml"]
+    itertag = "item"
+
+    async def parse_node(self, response, node):
+        yield {
+            "title": node.findtext("title"),
+            "link": node.findtext("link"),
+            "date": node.findtext("pubDate"),
+        }
+
+result = RSSSpider().start()
+```
+
+Like the other callbacks, `parse_node()` can also yield `Request` objects (for example, `response.follow(node.findtext("link"), callback=self.parse_post)`) to crawl into the pages the feed points to.
+
+### How nodes are matched and parsed
+
+Each node passed to `parse_node()` is an `lxml` element with all namespaces stripped, so `node.findtext("title")`, `node.find("thumbnail").get("url")`, and case-sensitive `node.xpath(...)` work on any feed without namespace maps. A plain `itertag` like `"entry"` matches nodes by name regardless of their namespace, which is what you want for Atom and most namespaced feeds. To match a node in one specific namespace, use a prefixed `itertag` and define the prefix in `namespaces`:
+
+```python
+class ThumbnailSpider(XMLFeedSpider):
+    name = "thumbs"
+    start_urls = ["https://example.com/feed.xml"]
+    itertag = "media:thumbnail"
+    namespaces = (("media", "http://search.yahoo.com/mrss/"),)
+
+    async def parse_node(self, response, node):
+        yield {"thumbnail": node.get("url")}
+```
+
+Gzipped feeds (`.xml.gz` or served with a gzip content-type) are decompressed automatically with the same protections the sitemap spider uses, and malformed XML logs a warning instead of crashing the crawl.
+
+## CSVFeedSpider
+
+`CSVFeedSpider` iterates over the rows of a CSV feed. Override `parse_row()`, which receives each row as a dictionary keyed by the column names:
+
+```python
+from scrapling.spiders import CSVFeedSpider
+
+class PriceSpider(CSVFeedSpider):
+    name = "prices"
+    start_urls = ["https://example.com/products.csv"]
+
+    async def parse_row(self, response, row):
+        yield {"product": row["title"], "price": float(row["price"])}
+
+result = PriceSpider().start()
+```
+
+By default, the first row of the feed is used as the header. If the feed has no header row, set `headers` to the column names yourself, and use `delimiter`/`quotechar` for feeds that don't follow the standard comma format:
+
+```python
+class PriceSpider(CSVFeedSpider):
+    name = "prices"
+    start_urls = ["https://example.com/products.csv"]
+    headers = ["title", "price", "url"]
+    delimiter = ";"
+```
+
+Gzipped feeds are decompressed automatically here as well, as shown above for **XMLFeedSpider**.
+
 ## Using `LinkExtractor` directly
 
 You don't have to use the templates. `LinkExtractor` works inside any plain `Spider`:

docs/spiders/generic-templates.md

@@ -122,6 +122,76 @@ class MySitemap(SitemapSpider):
 
 Set `sitemap_alternate_links = True` to also dispatch `<xhtml:link rel="alternate" hreflang="...">` URLs through your rules.
 
+## XMLFeedSpider
+
+`XMLFeedSpider` iterates over the nodes of an XML feed (RSS, Atom, product feeds, etc.). Set `itertag` to the node name you want (default: `"item"`) and override `parse_node()`, which is called once per matching node:
+
+```python
+from scrapling.spiders import XMLFeedSpider
+
+class RSSSpider(XMLFeedSpider):
+    name = "rss"
+    start_urls = ["https://example.com/feed.xml"]
+    itertag = "item"
+
+    async def parse_node(self, response, node):
+        yield {
+            "title": node.findtext("title"),
+            "link": node.findtext("link"),
+            "date": node.findtext("pubDate"),
+        }
+
+result = RSSSpider().start()
+```
+
+Like the other callbacks, `parse_node()` can also yield `Request` objects (for example, `response.follow(node.findtext("link"), callback=self.parse_post)`) to crawl into the pages the feed points to.
+
+### How nodes are matched and parsed
+
+Each node passed to `parse_node()` is an `lxml` element with all namespaces stripped, so `node.findtext("title")`, `node.find("thumbnail").get("url")`, and case-sensitive `node.xpath(...)` work on any feed without namespace maps. A plain `itertag` like `"entry"` matches nodes by name regardless of their namespace, which is what you want for Atom and most namespaced feeds. To match a node in one specific namespace, use a prefixed `itertag` and define the prefix in `namespaces`:
+
+```python
+class ThumbnailSpider(XMLFeedSpider):
+    name = "thumbs"
+    start_urls = ["https://example.com/feed.xml"]
+    itertag = "media:thumbnail"
+    namespaces = (("media", "http://search.yahoo.com/mrss/"),)
+
+    async def parse_node(self, response, node):
+        yield {"thumbnail": node.get("url")}
+```
+
+Gzipped feeds (`.xml.gz` or served with a gzip content-type) are decompressed automatically with the same protections the sitemap spider uses, and malformed XML logs a warning instead of crashing the crawl.
+
+## CSVFeedSpider
+
+`CSVFeedSpider` iterates over the rows of a CSV feed. Override `parse_row()`, which receives each row as a dictionary keyed by the column names:
+
+```python
+from scrapling.spiders import CSVFeedSpider
+
+class PriceSpider(CSVFeedSpider):
+    name = "prices"
+    start_urls = ["https://example.com/products.csv"]
+
+    async def parse_row(self, response, row):
+        yield {"product": row["title"], "price": float(row["price"])}
+
+result = PriceSpider().start()
+```
+
+By default, the first row of the feed is used as the header. If the feed has no header row, set `headers` to the column names yourself, and use `delimiter`/`quotechar` for feeds that don't follow the standard comma format:
+
+```python
+class PriceSpider(CSVFeedSpider):
+    name = "prices"
+    start_urls = ["https://example.com/products.csv"]
+    headers = ["title", "price", "url"]
+    delimiter = ";"
+```
+
+Gzipped feeds are decompressed automatically here as well, as shown above for **XMLFeedSpider**.
+
 ## Using `LinkExtractor` directly
 
 You don't have to use the templates. `LinkExtractor` works inside any plain `Spider`:

scrapling/spiders/__init__.py

@@ -5,7 +5,7 @@ from .engine import CrawlerEngine
 from .session import SessionManager
 from .spider import Spider, SessionConfigurationError
 from .links import LinkExtractor
-from .templates import CrawlSpider, SitemapSpider, CrawlRule, ShopifySpider
+from .templates import CrawlSpider, SitemapSpider, CrawlRule, ShopifySpider, XMLFeedSpider, CSVFeedSpider
 from scrapling.engines.toolbelt.custom import Response
 
 __all__ = [
@@ -22,4 +22,6 @@ __all__ = [
     "CrawlRule",
     "SitemapSpider",
     "ShopifySpider",
+    "XMLFeedSpider",
+    "CSVFeedSpider",
 ]

scrapling/spiders/templates/__init__.py

@@ -1,10 +1,13 @@
 from .crawler import CrawlSpider, CrawlRule
 from .sitemap import SitemapSpider
 from .shopify import ShopifySpider
+from .feed import XMLFeedSpider, CSVFeedSpider
 
 __all__ = [
     "CrawlSpider",
     "CrawlRule",
     "SitemapSpider",
     "ShopifySpider",
+    "XMLFeedSpider",
+    "CSVFeedSpider",
 ]

scrapling/spiders/templates/_utils.py

@@ -0,0 +1,24 @@
+"""Shared helpers for template spiders."""
+
+from gzip import GzipFile
+from io import BytesIO
+
+from scrapling.core._types import Optional
+
+__all__ = ["_decompress"]
+
+_GZIP_MAGIC = b"\x1f\x8b"
+_GUNZIP_MAX_SIZE = 64 * 1024 * 1024  # 64 MiB cap, defends against gzip bombs
+
+
+def _decompress(body: bytes, content_type: Optional[str]) -> bytes:
+    """Gunzip `body` when the content-type or the magic bytes say it's gzipped, capped against gzip bombs."""
+    if (content_type and ("gzip" in content_type.lower())) or (body[:2] == _GZIP_MAGIC):
+        out = bytearray()
+        with GzipFile(fileobj=BytesIO(body)) as f:
+            while chunk := f.read1(8192):
+                out.extend(chunk)
+                if len(out) > _GUNZIP_MAX_SIZE:
+                    raise OSError(f"gzip output exceeds {_GUNZIP_MAX_SIZE} bytes")
+        return bytes(out)
+    return body

scrapling/spiders/templates/feed.py

@@ -0,0 +1,139 @@
+"""Feed template spiders for XML and CSV feeds."""
+
+from copy import deepcopy
+from csv import DictReader
+from io import StringIO
+
+from lxml import etree
+
+from scrapling.core._types import (
+    TYPE_CHECKING,
+    Any,
+    AsyncGenerator,
+    Dict,
+    Iterator,
+    List,
+    Optional,
+    Tuple,
+    Union,
+)
+from scrapling.spiders.request import Request
+from scrapling.spiders.spider import Spider
+from scrapling.spiders.templates._utils import _decompress
+
+if TYPE_CHECKING:
+    from scrapling.engines.toolbelt.custom import Response
+
+
+__all__ = ["XMLFeedSpider", "CSVFeedSpider"]
+
+
+class XMLFeedSpider(Spider):
+    """A Spider that iterates over the nodes of an XML feed (RSS, Atom, product feeds, etc.).
+
+    Override `parse_node()` to process each node matching `itertag`. Gzipped feeds are decompressed automatically.
+
+    Each node is passed as a namespace-stripped `lxml` element, so `node.findtext("title")` and case-sensitive
+    `node.xpath(...)` work on any feed without namespace maps.
+
+    :cvar itertag: Name of the node to iterate over. A plain name ("item") matches regardless of namespace;
+        a prefixed name ("media:content") matches only the namespace the prefix maps to in `namespaces`.
+    :cvar namespaces: Tuple of `(prefix, uri)` pairs defining the prefixes usable in `itertag`.
+    """
+
+    itertag: str = "item"
+    namespaces: Tuple[Tuple[str, str], ...] = ()
+
+    async def parse(self, response: "Response") -> AsyncGenerator[Union[Dict[str, Any], Request, None], None]:
+        """Iterate over the feed's `itertag` nodes and dispatch each one to `parse_node`."""
+        content_type = response.headers.get("content-type") if response.headers else None
+        try:
+            body = _decompress(response.body, content_type)
+        except OSError as e:
+            self.logger.warning(f"Failed to decompress feed: {e}")
+            return
+
+        try:
+            root = etree.fromstring(body)
+        except etree.XMLSyntaxError as e:
+            self.logger.warning(f"Failed to parse XML feed from {response.url}: {e}")
+            return
+
+        for node in self._iter_nodes(root):
+            async for result in self.parse_node(response, node):
+                yield result
+
+    async def parse_node(
+        self, response: "Response", node: Any
+    ) -> AsyncGenerator[Union[Dict[str, Any], Request, None], None]:
+        """Override to process one feed node; `node` is a namespace-stripped `lxml` element."""
+        raise NotImplementedError(f"{self.__class__.__name__} must implement parse_node() method")
+        yield  # Make this a generator for type checkers
+
+    def _wanted_tag(self) -> Tuple[Optional[str], str]:
+        """Resolve `itertag` into a `(namespace uri or None, localname)` pair."""
+        prefix, _, name = self.itertag.rpartition(":")
+        if not prefix:
+            return None, name
+        uri = dict(self.namespaces).get(prefix)
+        if not uri:
+            raise ValueError(f"`itertag` prefix {prefix!r} is not defined in `namespaces`")
+        return uri, name
+
+    def _iter_nodes(self, root: Any) -> Iterator[Any]:
+        uri, name = self._wanted_tag()
+        for el in root.iter():
+            if isinstance(el.tag, str):
+                qname = etree.QName(el.tag)
+                if qname.localname == name and (uri is None or qname.namespace == uri):
+                    yield self._strip_namespaces(el)
+
+    @staticmethod
+    def _strip_namespaces(node: Any) -> Any:
+        """Return a copy of `node` with namespaces removed from every tag and attribute."""
+        node = deepcopy(node)
+        for el in node.iter():
+            if isinstance(el.tag, str):
+                el.tag = etree.QName(el.tag).localname
+            for key in list(el.attrib):
+                if key.startswith("{"):
+                    el.attrib[etree.QName(key).localname] = el.attrib.pop(key)
+        etree.cleanup_namespaces(node)
+        return node
+
+
+class CSVFeedSpider(Spider):
+    """A Spider that iterates over the rows of a CSV feed.
+
+    Override `parse_row()` to process each row as a dictionary. Gzipped feeds are decompressed automatically.
+
+    :cvar delimiter: The character separating fields.
+    :cvar quotechar: The character enclosing fields that contain special characters.
+    :cvar headers: The column names. When left unset, the first row of the feed is used as the header.
+    """
+
+    delimiter: str = ","
+    quotechar: str = '"'
+    headers: Optional[List[str]] = None
+
+    async def parse(self, response: "Response") -> AsyncGenerator[Union[Dict[str, Any], Request, None], None]:
+        """Read the feed's rows and dispatch each one to `parse_row`."""
+        content_type = response.headers.get("content-type") if response.headers else None
+        try:
+            body = _decompress(response.body, content_type)
+        except OSError as e:
+            self.logger.warning(f"Failed to decompress feed: {e}")
+            return
+
+        text = body.decode(response.encoding or "utf-8", errors="replace")
+        reader = DictReader(StringIO(text), fieldnames=self.headers, delimiter=self.delimiter, quotechar=self.quotechar)
+        for row in reader:
+            async for result in self.parse_row(response, dict(row)):
+                yield result
+
+    async def parse_row(
+        self, response: "Response", row: Dict[str, Any]
+    ) -> AsyncGenerator[Union[Dict[str, Any], Request, None], None]:
+        """Override to process one feed row as a `{column: value}` dictionary."""
+        raise NotImplementedError(f"{self.__class__.__name__} must implement parse_row() method")
+        yield  # Make this a generator for type checkers

scrapling/spiders/templates/sitemap.py

@@ -1,8 +1,6 @@
 """Sitemap template spider."""
 
 from dataclasses import dataclass, field
-from gzip import GzipFile
-from io import BytesIO
 from urllib.parse import urlsplit
 
 from lxml import etree
@@ -21,6 +19,7 @@ from scrapling.spiders.links import LinkExtractor
 from scrapling.spiders.request import Request
 from scrapling.spiders.spider import Spider
 from scrapling.spiders.templates.crawler import CrawlRule
+from scrapling.spiders.templates._utils import _decompress
 
 if TYPE_CHECKING:
     from scrapling.engines.toolbelt.custom import Response
@@ -29,10 +28,6 @@ if TYPE_CHECKING:
 __all__ = ["SitemapSpider"]
 
 
-_GZIP_MAGIC = b"\x1f\x8b"
-_GUNZIP_MAX_SIZE = 64 * 1024 * 1024  # 64 MiB cap, defends against gzip bombs
-
-
 @dataclass
 class SitemapResult:
     """Parsed sitemap body.
@@ -90,18 +85,6 @@ class SitemapSpider(Spider):
             return []
         return list(parser.sitemaps)
 
-    @staticmethod
-    def _decompress(body: bytes, content_type: Optional[str]) -> bytes:
-        if (content_type and ("gzip" in content_type.lower())) or (body[:2] == _GZIP_MAGIC):
-            out = bytearray()
-            with GzipFile(fileobj=BytesIO(body)) as f:
-                while chunk := f.read1(8192):
-                    out.extend(chunk)
-                    if len(out) > _GUNZIP_MAX_SIZE:
-                        raise OSError(f"gzip output exceeds {_GUNZIP_MAX_SIZE} bytes")
-            return bytes(out)
-        return body
-
     def _extract_urls(self, root: Any) -> List[str]:
         urls: List[str] = []
         for url_el in root:
@@ -125,7 +108,7 @@ class SitemapSpider(Spider):
     def _sm_body(self, body: bytes, content_type: Optional[str] = None) -> SitemapResult:
         """Parse a sitemap body and return its URLs and any child sitemaps."""
         try:
-            body = self._decompress(body, content_type)
+            body = _decompress(body, content_type)
         except OSError as e:
             self.logger.warning(f"Failed to decompress sitemap: {e}")
             return SitemapResult()

tests/spiders/test_feed.py

@@ -0,0 +1,266 @@
+"""Tests for `XMLFeedSpider` and `CSVFeedSpider`."""
+
+import gzip
+import logging
+
+import pytest
+
+from scrapling.engines.toolbelt.custom import Response
+from scrapling.spiders.request import Request
+from scrapling.spiders.templates.feed import CSVFeedSpider, XMLFeedSpider
+from scrapling.core._types import AsyncGenerator
+
+
+RSS_XML = b"""<?xml version="1.0" encoding="UTF-8"?>
+<rss version="2.0" xmlns:media="http://search.yahoo.com/mrss/">
+  <channel>
+    <title>Feed Title</title>
+    <item>
+      <title>First Post</title>
+      <link>https://example.com/posts/1</link>
+      <pubDate>Mon, 01 Jan 2026 00:00:00 GMT</pubDate>
+      <media:thumbnail url="https://example.com/thumb1.jpg"/>
+    </item>
+    <item>
+      <title>Second Post</title>
+      <link>https://example.com/posts/2</link>
+    </item>
+  </channel>
+</rss>
+"""
+
+ATOM_XML = b"""<?xml version="1.0" encoding="utf-8"?>
+<feed xmlns="http://www.w3.org/2005/Atom">
+  <title>Atom Feed</title>
+  <entry>
+    <title>Atom Post</title>
+    <link href="https://example.com/atom/1"/>
+  </entry>
+</feed>
+"""
+
+CSV_BODY = b"""title,price,url
+First,10.5,https://example.com/products/1
+Second,20,https://example.com/products/2
+"""
+
+CSV_NO_HEADER = b"""First,10.5
+Second,20
+"""
+
+CSV_SEMICOLON = b"""title;price
+'First;Post';10.5
+"""
+
+
+def _make_response(body: bytes, url: str = "https://example.com/feed.xml", headers: dict | None = None) -> Response:
+    resp = Response(
+        url=url,
+        content=body,
+        status=200,
+        reason="OK",
+        cookies={},
+        headers=headers or {},
+        request_headers={},
+    )
+    resp.request = Request(url, sid="default")
+    return resp
+
+
+async def _collect(agen: AsyncGenerator) -> list:
+    return [item async for item in agen]
+
+
+class _RSSSpider(XMLFeedSpider):
+    name = "rss"
+    start_urls = ["https://example.com/feed.xml"]
+
+    async def parse_node(self, response, node):
+        yield {
+            "title": node.findtext("title"),
+            "link": node.findtext("link"),
+            "date": node.findtext("pubDate"),
+        }
+
+
+class TestXMLFeedSpider:
+    @pytest.mark.asyncio
+    async def test_iterates_default_itertag(self):
+        items = await _collect(_RSSSpider().parse(_make_response(RSS_XML)))
+
+        assert len(items) == 2
+        assert items[0] == {
+            "title": "First Post",
+            "link": "https://example.com/posts/1",
+            "date": "Mon, 01 Jan 2026 00:00:00 GMT",
+        }
+        assert items[1]["title"] == "Second Post" and items[1]["date"] is None
+
+    @pytest.mark.asyncio
+    async def test_nodes_are_namespace_stripped(self):
+        class S(XMLFeedSpider):
+            name = "s"
+
+            async def parse_node(self, response, node):
+                thumbnail = node.find("thumbnail")
+                yield {"thumb": thumbnail.get("url") if thumbnail is not None else None}
+
+        items = await _collect(S().parse(_make_response(RSS_XML)))
+        assert items[0]["thumb"] == "https://example.com/thumb1.jpg"
+
+    @pytest.mark.asyncio
+    async def test_plain_itertag_matches_namespaced_nodes(self):
+        class S(XMLFeedSpider):
+            name = "s"
+            itertag = "entry"
+
+            async def parse_node(self, response, node):
+                link = node.find("link")
+                yield {"title": node.findtext("title"), "href": link.get("href") if link is not None else None}
+
+        items = await _collect(S().parse(_make_response(ATOM_XML)))
+        assert items == [{"title": "Atom Post", "href": "https://example.com/atom/1"}]
+
+    @pytest.mark.asyncio
+    async def test_prefixed_itertag_matches_by_namespace(self):
+        class S(XMLFeedSpider):
+            name = "s"
+            itertag = "media:thumbnail"
+            namespaces = (("media", "http://search.yahoo.com/mrss/"),)
+
+            async def parse_node(self, response, node):
+                yield {"url": node.get("url")}
+
+        items = await _collect(S().parse(_make_response(RSS_XML)))
+        assert items == [{"url": "https://example.com/thumb1.jpg"}]
+
+    @pytest.mark.asyncio
+    async def test_unknown_itertag_prefix_raises(self):
+        class S(XMLFeedSpider):
+            name = "s"
+            itertag = "media:thumbnail"
+
+        with pytest.raises(ValueError, match="namespaces"):
+            await _collect(S().parse(_make_response(RSS_XML)))
+
+    @pytest.mark.asyncio
+    async def test_gzipped_feed_is_decompressed(self):
+        items = await _collect(_RSSSpider().parse(_make_response(gzip.compress(RSS_XML))))
+        assert len(items) == 2
+
+    @pytest.mark.asyncio
+    async def test_malformed_xml_logs_warning_and_yields_nothing(self):
+        spider = _RSSSpider()
+        records = []
+
+        class Capture(logging.Handler):
+            def emit(self, record):
+                records.append(record.getMessage())
+
+        spider.logger.addHandler(Capture())
+        items = await _collect(spider.parse(_make_response(b"this is <<< not xml")))
+
+        assert items == []
+        assert any("Failed to parse XML feed" in message for message in records)
+
+    @pytest.mark.asyncio
+    async def test_requests_yielded_from_parse_node_flow_through(self):
+        class S(XMLFeedSpider):
+            name = "s"
+
+            async def parse_node(self, response, node):
+                yield response.follow(node.findtext("link"), callback=self.parse_post)
+
+            async def parse_post(self, response):
+                yield {"url": response.url}
+
+        results = await _collect(S().parse(_make_response(RSS_XML)))
+        assert len(results) == 2
+        assert all(isinstance(r, Request) for r in results)
+        assert results[0].url == "https://example.com/posts/1"
+
+    @pytest.mark.asyncio
+    async def test_parse_node_not_overridden_raises(self):
+        class S(XMLFeedSpider):
+            name = "s"
+
+        with pytest.raises(NotImplementedError, match="parse_node"):
+            await _collect(S().parse(_make_response(RSS_XML)))
+
+    @pytest.mark.asyncio
+    async def test_start_requests_uses_start_urls(self):
+        requests = await _collect(_RSSSpider().start_requests())
+        assert len(requests) == 1 and requests[0].url == "https://example.com/feed.xml"
+
+
+class _PriceSpider(CSVFeedSpider):
+    name = "prices"
+    start_urls = ["https://example.com/feed.csv"]
+
+    async def parse_row(self, response, row):
+        yield row
+
+
+class TestCSVFeedSpider:
+    @pytest.mark.asyncio
+    async def test_first_row_is_the_header(self):
+        rows = await _collect(_PriceSpider().parse(_make_response(CSV_BODY)))
+
+        assert len(rows) == 2
+        assert rows[0] == {"title": "First", "price": "10.5", "url": "https://example.com/products/1"}
+
+    @pytest.mark.asyncio
+    async def test_explicit_headers(self):
+        class S(_PriceSpider):
+            headers = ["name", "cost"]
+
+        rows = await _collect(S().parse(_make_response(CSV_NO_HEADER)))
+        assert rows == [{"name": "First", "cost": "10.5"}, {"name": "Second", "cost": "20"}]
+
+    @pytest.mark.asyncio
+    async def test_custom_delimiter_and_quotechar(self):
+        class S(_PriceSpider):
+            delimiter = ";"
+            quotechar = "'"
+
+        rows = await _collect(S().parse(_make_response(CSV_SEMICOLON)))
+        assert rows == [{"title": "First;Post", "price": "10.5"}]
+
+    @pytest.mark.asyncio
+    async def test_gzipped_feed_is_decompressed(self):
+        rows = await _collect(_PriceSpider().parse(_make_response(gzip.compress(CSV_BODY))))
+        assert len(rows) == 2
+
+    @pytest.mark.asyncio
+    async def test_empty_body_yields_nothing(self):
+        assert await _collect(_PriceSpider().parse(_make_response(b""))) == []
+
+    @pytest.mark.asyncio
+    async def test_non_utf8_bytes_do_not_crash(self):
+        body = "title,price\nCafé,10\n".encode("latin-1")
+        rows = await _collect(_PriceSpider().parse(_make_response(body)))
+        assert len(rows) == 1 and rows[0]["price"] == "10"
+
+    @pytest.mark.asyncio
+    async def test_parse_row_not_overridden_raises(self):
+        class S(CSVFeedSpider):
+            name = "s"
+
+        with pytest.raises(NotImplementedError, match="parse_row"):
+            await _collect(S().parse(_make_response(CSV_BODY)))
+
+    @pytest.mark.asyncio
+    async def test_requests_yielded_from_parse_row_flow_through(self):
+        class S(CSVFeedSpider):
+            name = "s"
+
+            async def parse_row(self, response, row):
+                yield response.follow(row["url"], callback=self.parse_product)
+
+            async def parse_product(self, response):
+                yield {"url": response.url}
+
+        results = await _collect(S().parse(_make_response(CSV_BODY, url="https://example.com/feed.csv")))
+        assert len(results) == 2
+        assert all(isinstance(r, Request) for r in results)
+        assert results[0].url == "https://example.com/products/1"