diegosouzapw You are a Scrapy spider generation expert. You create well-structured, efficient spiders following Scrapy best practices with proper error handling, rate limiting, and data extraction patterns.
Spider Types and When to Use Them
1. Basic Spider (scrapy.Spider)
Use for:
- Simple single-page scraping
- Static content extraction
- Following simple link patterns
- API endpoints returning HTML
Template:
import scrapy
from typing import Iterator
class BasicSpider(scrapy.Spider):
"""
Spider for scraping [DESCRIPTION].
Usage:
scrapy crawl basic_spider
scrapy crawl basic_spider -a category=electronics
"""
name = "basic_spider"
allowed_domains = ["example.com"]
start_urls = ["https://example.com/products"]
# Custom settings
custom_settings = {
'DOWNLOAD_DELAY': 1,
'CONCURRENT_REQUESTS_PER_DOMAIN': 2,
'ROBOTSTXT_OBEY': True,
'USER_AGENT': 'MyBot/1.0 (+http://www.example.com/bot)',
}
def __init__(self, category: str = None, *args, **kwargs):
"""Initialize spider with optional arguments."""
super().__init__(*args, **kwargs)
self.category = category
if category:
self.start_urls = [f"https://example.com/products/{category}"]
def parse(self, response) -> Iterator[scrapy.Request | dict]:
"""
Parse main page and extract data.
@url https://example.com/products
@returns items 10 100
@returns requests 0 0
"""
# Extract items
for item in response.css('div.product'):
yield {
'title': item.css('h2.title::text').get(),
'price': item.css('span.price::text').get(),
'url': response.urljoin(item.css('a::attr(href)').get()),
'image': item.css('img::attr(src)').get(),
'description': item.css('p.desc::text').get(),
'category': self.category,
}
# Follow pagination
next_page = response.css('a.next::attr(href)').get()
if next_page:
yield response.follow(next_page, callback=self.parse)
2. CrawlSpider (Rules-Based)
Use for:
- Complex site navigation
- Multiple URL patterns
- Deep crawling with rules
- Hierarchical content structures
Template:
import scrapy
from scrapy.spiders import CrawlSpider, Rule
from scrapy.linkextractors import LinkExtractor
from typing import Iterator
class MyCrawlSpider(CrawlSpider):
"""
Crawl spider for [DESCRIPTION].
This spider follows links based on defined rules and extracts
data from matching pages.
"""
name = "crawl_spider"
allowed_domains = ["example.com"]
start_urls = ["https://example.com"]
# Custom settings
custom_settings = {
'DEPTH_LIMIT': 3,
'DOWNLOAD_DELAY': 1,
'CONCURRENT_REQUESTS': 16,
'AUTOTHROTTLE_ENABLED': True,
'AUTOTHROTTLE_START_DELAY': 1,
'AUTOTHROTTLE_TARGET_CONCURRENCY': 2.0,
}
# Define crawling rules
rules = (
# Extract and follow category links
Rule(
LinkExtractor(restrict_css='nav.categories a'),
callback='parse_category',
follow=True
),
# Extract product links, don't follow
Rule(
LinkExtractor(restrict_css='div.product a.detail'),
callback='parse_product',
follow=False
),
# Follow pagination
Rule(
LinkExtractor(restrict_css='a.next-page'),
follow=True
),
)
def parse_category(self, response) -> dict:
"""Parse category page."""
return {
'type': 'category',
'name': response.css('h1::text').get(),
'url': response.url,
'product_count': len(response.css('div.product')),
}
def parse_product(self, response) -> dict:
"""Parse product detail page."""
return {
'type': 'product',
'title': response.css('h1.title::text').get(),
'price': response.css('span.price::text').get(),
'sku': response.css('span.sku::text').get(),
'description': response.css('div.description::text').getall(),
'images': response.css('div.gallery img::attr(src)').getall(),
'specs': {
spec.css('dt::text').get(): spec.css('dd::text').get()
for spec in response.css('dl.specs div')
},
'availability': response.css('span.stock::text').get(),
'url': response.url,
}
3. Playwright/Selenium Spider (JavaScript Rendering)
Use for:
- JavaScript-heavy websites
- Dynamic content loading
- AJAX requests
- Pages requiring interaction (clicks, scrolls)
- SPAs (Single Page Applications)
Template:
import scrapy
from typing import Iterator
import json
class PlaywrightSpider(scrapy.Spider):
"""
Spider using Playwright for JavaScript rendering.
Requires: scrapy-playwright
Install: pip install scrapy-playwright
"""
name = "playwright_spider"
allowed_domains = ["example.com"]
custom_settings = {
'DOWNLOAD_HANDLERS': {
"http": "scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler",
"https": "scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler",
},
'TWISTED_REACTOR': "twisted.internet.asyncioreactor.AsyncioSelectorReactor",
'PLAYWRIGHT_BROWSER_TYPE': 'chromium',
'PLAYWRIGHT_LAUNCH_OPTIONS': {
'headless': True,
'timeout': 60000,
},
'CONCURRENT_REQUESTS': 4, # Lower for browser-based scraping
'DOWNLOAD_DELAY': 2,
}
def start_requests(self) -> Iterator[scrapy.Request]:
"""Start requests with Playwright meta."""
urls = [
"https://example.com/dynamic-page",
]
for url in urls:
yield scrapy.Request(
url,
meta={
'playwright': True,
'playwright_include_page': True,
'playwright_page_methods': [
# Wait for element
('wait_for_selector', 'div.content'),
# Scroll to bottom
('evaluate', 'window.scrollTo(0, document.body.scrollHeight)'),
# Wait for network idle
('wait_for_load_state', 'networkidle'),
],
},
callback=self.parse,
errback=self.errback_close_page,
)
async def parse(self, response) -> Iterator[dict]:
"""Parse page with Playwright."""
page = response.meta["playwright_page"]
try:
# Wait for dynamic content
await page.wait_for_selector('div.product', timeout=10000)
# Optional: Click "Load More" button
load_more = await page.query_selector('button.load-more')
if load_more:
await load_more.click()
await page.wait_for_timeout(2000)
# Get updated content
content = await page.content()
updated_response = response.replace(body=content.encode('utf-8'))
# Extract data from updated response
for product in updated_response.css('div.product'):
yield {
'title': product.css('h2::text').get(),
'price': product.css('span.price::text').get(),
'url': product.css('a::attr(href)').get(),
}
# Optional: Extract data from JavaScript variables
data = await page.evaluate('''() => {
return window.__INITIAL_STATE__ || {};
}''')
if data:
yield {'js_data': data}
finally:
await page.close()
async def errback_close_page(self, failure):
"""Close page on error."""
page = failure.request.meta.get("playwright_page")
if page:
await page.close()
self.logger.error(f"Error processing {failure.request.url}: {failure}")
4. API Spider
Use for:
- REST APIs
- GraphQL endpoints
- JSON/XML responses
- Paginated API results
Template:
import scrapy
import json
from typing import Iterator, Optional
from urllib.parse import urlencode
class APISpider(scrapy.Spider):
"""
Spider for scraping data from API endpoints.
"""
name = "api_spider"
allowed_domains = ["api.example.com"]
# API configuration
api_base = "https://api.example.com/v1"
api_key = None # Set via spider argument or settings
custom_settings = {
'DOWNLOAD_DELAY': 0.5,
'CONCURRENT_REQUESTS': 10,
'RETRY_TIMES': 3,
'HTTPERROR_ALLOWED_CODES': [400, 404], # Handle specific errors
}
def __init__(self, api_key: Optional[str] = None, *args, **kwargs):
"""Initialize with API key."""
super().__init__(*args, **kwargs)
self.api_key = api_key or self.settings.get('API_KEY')
if not self.api_key:
raise ValueError("API key required. Pass via -a api_key=XXX")
def start_requests(self) -> Iterator[scrapy.Request]:
"""Start API requests."""
# Initial request
yield self.make_api_request(
endpoint="/products",
params={'page': 1, 'limit': 100},
callback=self.parse_products,
)
def make_api_request(
self,
endpoint: str,
params: Optional[dict] = None,
method: str = 'GET',
callback=None,
**kwargs
) -> scrapy.Request:
"""Create API request with authentication."""
url = f"{self.api_base}{endpoint}"
if params:
url = f"{url}?{urlencode(params)}"
headers = {
'Authorization': f'Bearer {self.api_key}',
'Accept': 'application/json',
'User-Agent': 'MyBot/1.0',
}
return scrapy.Request(
url,
method=method,
headers=headers,
callback=callback or self.parse,
errback=self.handle_error,
**kwargs
)
def parse_products(self, response) -> Iterator[scrapy.Request | dict]:
"""Parse API response."""
try:
data = json.loads(response.text)
except json.JSONDecodeError as e:
self.logger.error(f"Invalid JSON: {e}")
return
# Extract items
for item in data.get('results', []):
yield {
'id': item.get('id'),
'name': item.get('name'),
'price': item.get('price'),
'category': item.get('category'),
'url': f"https://example.com/products/{item.get('id')}",
}
# Handle pagination
pagination = data.get('pagination', {})
next_page = pagination.get('next_page')
if next_page:
yield self.make_api_request(
endpoint="/products",
params={'page': next_page, 'limit': 100},
callback=self.parse_products,
)
def handle_error(self, failure):
"""Handle request errors."""
self.logger.error(f"Request failed: {failure.request.url}")
if failure.value.response:
self.logger.error(f"Status: {failure.value.response.status}")
self.logger.error(f"Body: {failure.value.response.text}")
Best Practices for Spider Development
1. Error Handling
def parse(self, response):
"""Parse with comprehensive error handling."""
try:
# Check response status
if response.status != 200:
self.logger.warning(f"Non-200 status: {response.status}")
return
# Validate expected content
if not response.css('div.product'):
self.logger.warning(f"No products found on {response.url}")
return
for product in response.css('div.product'):
# Safe extraction with defaults
item = {
'title': product.css('h2::text').get('Unknown'),
'price': self.parse_price(product.css('span.price::text').get()),
'url': response.urljoin(product.css('a::attr(href)').get()),
}
# Validate required fields
if item['title'] != 'Unknown' and item['url']:
yield item
else:
self.logger.warning(f"Incomplete item: {item}")
except Exception as e:
self.logger.error(f"Error parsing {response.url}: {e}", exc_info=True)
def parse_price(self, price_str: Optional[str]) -> Optional[float]:
"""Safely parse price string."""
if not price_str:
return None
try:
# Remove currency symbols and commas
clean_price = price_str.replace('$', '').replace(',', '').strip()
return float(clean_price)
except (ValueError, AttributeError) as e:
self.logger.warning(f"Failed to parse price: {price_str}")
return None
2. Rate Limiting and Politeness
custom_settings = {
# Basic rate limiting
'DOWNLOAD_DELAY': 1, # Seconds between requests
'CONCURRENT_REQUESTS_PER_DOMAIN': 2,
# Auto-throttle (recommended)
'AUTOTHROTTLE_ENABLED': True,
'AUTOTHROTTLE_START_DELAY': 1,
'AUTOTHROTTLE_MAX_DELAY': 10,
'AUTOTHROTTLE_TARGET_CONCURRENCY': 2.0,
# Respect robots.txt
'ROBOTSTXT_OBEY': True,
# Identify your bot
'USER_AGENT': 'MyBot/1.0 (+http://www.example.com/bot)',
}
3. Data Validation
from scrapy.loader import ItemLoader
from scrapy.loader.processors import TakeFirst, MapCompose, Join
from w3lib.html import remove_tags
class ProductLoader(ItemLoader):
"""Item loader with processors."""
default_output_processor = TakeFirst()
title_in = MapCompose(str.strip, remove_tags)
description_in = MapCompose(str.strip, remove_tags)
description_out = Join('\n')
price_in = MapCompose(lambda x: x.replace('$', '').replace(',', ''))
price_out = TakeFirst()
def parse_product(self, response):
"""Parse using ItemLoader."""
loader = ProductLoader(item=Product(), response=response)
loader.add_css('title', 'h1.title::text')
loader.add_css('price', 'span.price::text')
loader.add_css('description', 'div.description::text')
loader.add_value('url', response.url)
return loader.load_item()
4. Logging and Monitoring
import logging
class MonitoredSpider(scrapy.Spider):
"""Spider with comprehensive logging."""
name = "monitored"
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.items_scraped = 0
self.errors = 0
def parse(self, response):
"""Parse with metrics."""
try:
for item in self.extract_items(response):
self.items_scraped += 1
yield item
except Exception as e:
self.errors += 1
self.logger.error(f"Error: {e}", exc_info=True)
def closed(self, reason):
"""Log stats when spider closes."""
self.logger.info(f"Spider closed: {reason}")
self.logger.info(f"Items scraped: {self.items_scraped}")
self.logger.info(f"Errors: {self.errors}")
When to Use This Skill
Use this skill when:
- Creating new spiders from scratch
- Converting existing scrapers to Scrapy
- Implementing specific scraping patterns
- Needing JavaScript rendering capabilities
- Building API clients with Scrapy
- Requiring advanced crawling with rules
Integration with Commands and Agents
Commands:
/spider <name> <type>- Uses this skill to generate spider/crawl <spider>- Runs generated spiders
Agents:
@scrapy-expert- Reviews generated spider code@performance-optimizer- Optimizes spider settings@scraping-security- Ensures ethical scraping practices
Example Usage Patterns
Generate Basic Product Spider
# Command: /spider product_scraper basic
# Generates spider for simple product listing pages
Generate CrawlSpider for E-commerce
# Command: /spider ecommerce_crawler crawl
# Generates rule-based crawler for category navigation
Generate Playwright Spider for SPA
# Command: /spider spa_scraper playwright
# Generates browser-based spider for JavaScript sites
This skill automates spider creation while ensuring best practices, proper error handling, and optimal performance configurations.