# Python Scrapy at Enterprise Scale: A Practical Guide [Technology](/blog/categories/technology) 15 May 2025 12 min read ## Why Scrapy for Enterprise Web Scraping? Scrapy stands out as the premier Python framework for large-scale web scraping operations. Unlike simple scripts or basic tools, Scrapy provides a solid architecture, built-in features, and extensibility that enterprise applications need. This guide covers deploying Scrapy in production environments, from initial setup through to advanced optimisation. ## Scrapy Architecture for Production Use ### Core Components Overview - **Scrapy Engine:** Controls data flow between components - **Scheduler:** Receives requests and queues them for processing - **Downloader:** Fetches web pages and returns responses - **Spiders:** Custom classes that define scraping logic - **Item Pipeline:** Processes extracted data - **Middlewares:** Hooks for customising request/response processing ### Production Project Structure ``` enterprise_scraper/ ├── scrapy.cfg ├── requirements.txt ├── docker-compose.yml ├── enterprise_scraper/ │ ├── __init__.py │ ├── settings/ │ │ ├── __init__.py │ │ ├── base.py │ │ ├── development.py │ │ ├── staging.py │ │ └── production.py │ ├── spiders/ │ │ ├── __init__.py │ │ ├── base_spider.py │ │ └── ecommerce_spider.py │ ├── items.py │ ├── pipelines.py │ ├── middlewares.py │ └── utils/ │ ├── __init__.py │ ├── database.py │ └── monitoring.py ├── deploy/ │ ├── Dockerfile │ └── kubernetes/ └── tests/ ├── unit/ └── integration/ ``` ## Advanced Configuration Management ### Environment-Specific Settings ``` # settings/base.py BOT_NAME = 'enterprise_scraper' SPIDER_MODULES = ['enterprise_scraper.spiders'] NEWSPIDER_MODULE = 'enterprise_scraper.spiders' # Respect robots.txt for compliance ROBOTSTXT_OBEY = True # Configure concurrent requests CONCURRENT_REQUESTS = 32 CONCURRENT_REQUESTS_PER_DOMAIN = 8 # Download delays for respectful scraping DOWNLOAD_DELAY = 1 RANDOMIZE_DOWNLOAD_DELAY = 0.5 # Production settings/production.py from .base import * # Increase concurrency for production CONCURRENT_REQUESTS = 100 CONCURRENT_REQUESTS_PER_DOMAIN = 16 # Enable autothrottling AUTOTHROTTLE_ENABLED = True AUTOTHROTTLE_START_DELAY = 1 AUTOTHROTTLE_MAX_DELAY = 10 AUTOTHROTTLE_TARGET_CONCURRENCY = 2.0 # Logging configuration LOG_LEVEL = 'INFO' LOG_FILE = '/var/log/scrapy/scrapy.log' # Database settings DATABASE_URL = os.environ.get('DATABASE_URL') REDIS_URL = os.environ.get('REDIS_URL') ``` ### Dynamic Settings with Environment Variables ``` import os from scrapy.utils.project import get_project_settings def get_scrapy_settings(): settings = get_project_settings() # Environment-specific overrides if os.environ.get('SCRAPY_ENV') == 'production': settings.set('CONCURRENT_REQUESTS', 200) settings.set('DOWNLOAD_DELAY', 0.5) elif os.environ.get('SCRAPY_ENV') == 'development': settings.set('CONCURRENT_REQUESTS', 16) settings.set('DOWNLOAD_DELAY', 2) return settings ``` ## Enterprise Spider Development ### Base Spider Class ``` import scrapy from scrapy.http import Request from typing import Generator, Optional import logging class BaseSpider(scrapy.Spider): """Base spider with common enterprise functionality""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.setup_logging() self.setup_monitoring() def setup_logging(self): """Configure structured logging""" self.logger = logging.getLogger(self.name) def setup_monitoring(self): """Initialize monitoring metrics""" self.stats = { 'pages_scraped': 0, 'items_extracted': 0, 'errors': 0 } def parse_with_error_handling(self, response): """Parse with comprehensive error handling""" try: yield from self.parse_content(response) except Exception as e: self.logger.error(f"Error parsing {response.url}: {e}") self.stats['errors'] += 1 def make_request(self, url: str, callback=None, meta: dict = None) -> Request: """Create request with standard metadata""" return Request( url=url, callback=callback or self.parse_with_error_handling, meta={ 'spider_name': self.name, 'timestamp': time.time(), **(meta or {}) }, dont_filter=False ) ``` ### Advanced E-commerce Spider ``` from enterprise_scraper.spiders.base_spider import BaseSpider from enterprise_scraper.items import ProductItem class EcommerceSpider(BaseSpider): name = 'ecommerce' allowed_domains = ['example-store.com'] custom_settings = { 'ITEM_PIPELINES': { 'enterprise_scraper.pipelines.ValidationPipeline': 300, 'enterprise_scraper.pipelines.DatabasePipeline': 400, }, 'DOWNLOAD_DELAY': 2, } def start_requests(self): """Generate initial requests with pagination""" base_url = "https://example-store.com/products" for page in range(1, 101): # First 100 pages url = f"{base_url}?page={page}" yield self.make_request( url=url, callback=self.parse_product_list, meta={'page': page} ) def parse_product_list(self, response): """Extract product URLs from listing pages""" product_urls = response.css('.product-link::attr(href)').getall() for url in product_urls: yield self.make_request( url=response.urljoin(url), callback=self.parse_product, meta={'category': response.meta.get('category')} ) # Handle pagination next_page = response.css('.pagination .next::attr(href)').get() if next_page: yield self.make_request( url=response.urljoin(next_page), callback=self.parse_product_list ) def parse_product(self, response): """Extract product details""" item = ProductItem() item['url'] = response.url item['name'] = response.css('h1.product-title::text').get() item['price'] = self.extract_price(response) item['description'] = response.css('.product-description::text').getall() item['images'] = response.css('.product-images img::attr(src)').getall() item['availability'] = response.css('.stock-status::text').get() item['rating'] = self.extract_rating(response) item['reviews_count'] = self.extract_reviews_count(response) self.stats['items_extracted'] += 1 yield item def extract_price(self, response): """Extract and normalize price data""" price_text = response.css('.price::text').get() if price_text: # Remove currency symbols and normalize import re price = re.sub(r'[^\d.]', '', price_text) return float(price) if price else None return None ``` ## Enterprise Pipeline System ### Validation Pipeline ``` from itemadapter import ItemAdapter from scrapy.exceptions import DropItem import validators class ValidationPipeline: """Validate items before processing""" def process_item(self, item, spider): adapter = ItemAdapter(item) # Required field validation if not adapter.get('name'): raise DropItem(f"Missing product name: {item}") # URL validation if not validators.url(adapter.get('url')): raise DropItem(f"Invalid URL: {adapter.get('url')}") # Price validation price = adapter.get('price') if price is not None: try: price = float(price) if price < 0: raise DropItem(f"Invalid price: {price}") adapter['price'] = price except (ValueError, TypeError): raise DropItem(f"Invalid price format: {price}") spider.logger.info(f"Item validated: {adapter.get('name')}") return item ``` ### Database Pipeline with Connection Pooling ``` import asyncio import asyncpg from itemadapter import ItemAdapter class DatabasePipeline: """Asynchronous database pipeline""" def __init__(self, db_url, pool_size=20): self.db_url = db_url self.pool_size = pool_size self.pool = None @classmethod def from_crawler(cls, crawler): return cls( db_url=crawler.settings.get('DATABASE_URL'), pool_size=crawler.settings.get('DB_POOL_SIZE', 20) ) async def open_spider(self, spider): """Initialize database connection pool""" self.pool = await asyncpg.create_pool( self.db_url, min_size=5, max_size=self.pool_size ) spider.logger.info("Database connection pool created") async def close_spider(self, spider): """Close database connection pool""" if self.pool: await self.pool.close() spider.logger.info("Database connection pool closed") async def process_item(self, item, spider): """Insert item into database""" adapter = ItemAdapter(item) async with self.pool.acquire() as connection: await connection.execute(''' INSERT INTO products (url, name, price, description) VALUES ($1, $2, $3, $4) ON CONFLICT (url) DO UPDATE SET name = EXCLUDED.name, price = EXCLUDED.price, description = EXCLUDED.description, updated_at = NOW() ''', adapter.get('url'), adapter.get('name'), adapter.get('price'), '\n'.join(adapter.get('description', [])) ) spider.logger.info(f"Item saved: {adapter.get('name')}") return item ``` ## Middleware for Enterprise Features ### Rotating Proxy Middleware ``` import random from scrapy.downloadermiddlewares.httpproxy import HttpProxyMiddleware class RotatingProxyMiddleware(HttpProxyMiddleware): """Rotate proxies for each request""" def __init__(self, proxy_list): self.proxy_list = proxy_list @classmethod def from_crawler(cls, crawler): proxy_list = crawler.settings.get('PROXY_LIST', []) return cls(proxy_list) def process_request(self, request, spider): if self.proxy_list: proxy = random.choice(self.proxy_list) request.meta['proxy'] = proxy spider.logger.debug(f"Using proxy: {proxy}") return None ``` ### Rate Limiting Middleware ``` import time from collections import defaultdict from scrapy.downloadermiddlewares.retry import RetryMiddleware class RateLimitMiddleware(RetryMiddleware): """Implement per-domain rate limiting""" def __init__(self, settings): super().__init__(settings) self.domain_delays = defaultdict(float) self.last_request_time = defaultdict(float) def process_request(self, request, spider): domain = request.url.split('/')[2] current_time = time.time() # Calculate required delay min_delay = self.domain_delays.get(domain, 1.0) time_since_last = current_time - self.last_request_time[domain] if time_since_last < min_delay: delay = min_delay - time_since_last spider.logger.debug(f"Rate limiting {domain}: {delay:.2f}s") time.sleep(delay) self.last_request_time[domain] = time.time() return None ``` ## Monitoring and Observability ### Custom Stats Collection ``` from scrapy.statscollectors import StatsCollector import time class EnterpriseStatsCollector(StatsCollector): """Enhanced stats collection for monitoring""" def __init__(self, crawler): super().__init__(crawler) self.start_time = time.time() self.custom_stats = {} def get_stats(self): """Enhanced stats with custom metrics""" stats = super().get_stats() # Add runtime statistics runtime = time.time() - self.start_time stats['runtime_seconds'] = runtime # Add rate calculations pages_count = stats.get('response_received_count', 0) if runtime > 0: stats['pages_per_minute'] = (pages_count / runtime) * 60 # Add custom metrics stats.update(self.custom_stats) return stats def inc_value(self, key, count=1, start=0): """Increment custom counter""" super().inc_value(key, count, start) # Log significant milestones current_value = self.get_value(key, 0) if current_value % 1000 == 0: # Every 1000 items self.crawler.spider.logger.info(f"{key}: {current_value}") ``` ## Production Deployment Deploying Scrapy at enterprise scale requires solid infrastructure and monitoring. For end-to-end [data pipeline services](../../services/data-cleaning), consider our managed deployment options that handle scaling, monitoring, and compliance automatically. ### Docker Configuration ``` # Dockerfile FROM python:3.9-slim WORKDIR /app # Install system dependencies RUN apt-get update && apt-get install -y \ gcc \ libc-dev \ libffi-dev \ libssl-dev \ && rm -rf /var/lib/apt/lists/* # Install Python dependencies COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # Copy application code COPY . . # Create non-root user RUN useradd -m -u 1000 scrapy && chown -R scrapy:scrapy /app USER scrapy # Default command CMD ["scrapy", "crawl", "ecommerce"] ``` ### Kubernetes Deployment ``` apiVersion: apps/v1 kind: Deployment metadata: name: scrapy-deployment spec: replicas: 3 selector: matchLabels: app: scrapy template: metadata: labels: app: scrapy spec: containers: - name: scrapy image: enterprise-scrapy:latest resources: requests: memory: "1Gi" cpu: "500m" limits: memory: "2Gi" cpu: "1000m" env: - name: SCRAPY_ENV value: "production" - name: DATABASE_URL valueFrom: secretKeyRef: name: db-secret key: url --- apiVersion: v1 kind: Service metadata: name: scrapy-service spec: selector: app: scrapy ports: - port: 6800 targetPort: 6800 ``` ## Performance Optimisation ### Memory Management - **Item Pipeline:** Process items immediately to avoid memory buildup - **Response Caching:** Disable for production unless specifically needed - **Request Filtering:** Use duplicate filters efficiently - **Large Responses:** Stream large files instead of loading into memory ### Scaling Strategies - **Horizontal Scaling:** Multiple spider instances - **Domain Sharding:** Distribute domains across instances - **Queue Management:** Redis-based distributed queuing - **Load Balancing:** Distribute requests across proxy pools ## Summary and recommendations ### Code Organisation - Use inheritance for common spider functionality - Separate settings by environment - Implement thorough error handling - Write unit tests for custom components ### Operational Excellence - Monitor performance metrics continuously - Implement circuit breakers for external services - Use structured logging for better observability - Plan for graceful degradation ### Compliance and Ethics - Respect robots.txt and rate limits - Implement proper user agent identification - Handle personal data according to GDPR - Maintain audit trails for data collection ### Related Articles Web Scraping #### [Scraping JavaScript-Heavy Sites: Advanced Techniques](javascript-heavy-sites-scraping) 6 min read Technology #### [Cloud-Native Scraping Architecture for Enterprise Scale](cloud-native-scraping-architecture) 11 min read Compliance #### [Complete Guide to Web Scraping Compliance in the UK](web-scraping-compliance-uk-guide) 12 min read --- Source: https://ukdataservices.co.uk/blog/articles/python-scrapy-enterprise-guide