{"schemaVersion":"1.0","type":"TechArticle","types":["Article","TechArticle"],"slug":"build-a-web-scraping-api-with-fastapi-celery-redis-2026-2osqv","url":"https://zyvop.com/build-a-web-scraping-api-with-fastapi-celery-redis-2026-2osqv","title":"Build a Web Scraping API with FastAPI, Celery & Redis (2026)","subtitle":"Build a production-grade web scraping API with FastAPI, Celery, and Redis in Python. Schedule scrape jobs, return async results, add rate limiting, and deploy with Docker in 2026.","tldr":"Build an asynchronous web scraping API with FastAPI, Celery, Redis, and Python. Learn how to queue scraping jobs, run dedicated workers, store results, handle retries and rate limits, monitor tasks, and deploy the complete scraping service with Docker.","keywords":["turn scraper into api python","fastapi async scraping service","python scraping api tutorial fastapi","celery redis scraper background tasks","Scraping"],"entities":["ZyVOP","turn scraper into api python","fastapi async scraping service","python scraping api tutorial fastapi","celery redis scraper background tasks","Scraping"],"keyTakeaways":["Why Your Scraper Needs an API Layer At some point every scraper outgrows its original script form.","Maybe you want to trigger scraping from a dashboard.","Maybe a frontend team needs to query scraped data."],"headings":["Why Your Scraper Needs an API Layer","Project Structure","Step 1: Install Dependencies","Step 2: Database Models","Step 3: Pydantic Schemas","Step 4: The Scraping Logic","Step 5: Celery Tasks","Step 6: The FastAPI Application","Step 7: Database Setup","Step 8: Docker Compose","Step 9: Using the API","Step 10: Monitoring with Flower","Production Checklist","FAQ"],"outboundLinks":[],"contentText":"Why Your Scraper Needs an API Layer At some point every scraper outgrows its original script form. Maybe you want to trigger scraping from a dashboard. Maybe a frontend team needs to query scraped data. Maybe you're building a product — a rank tracker, a price monitor, a lead generation tool — and users need to submit URLs and get results back. The moment your scraper needs to serve multiple callers, run on a schedule, handle concurrent requests, or return results asynchronously, you need to wrap it in a proper API. In this practical approach, we'll build a scraper wrapped in a web API powered by FastAPI to scrape and deliver data on demand — implementing FastAPI web services with asynchronous request handling, configuring real-time data scraping with caching and webhook support, and handling concurrent requests with rate limiting for scalable scraping API endpoints. The stack we'll build is a proven pattern: Client → FastAPI (HTTP layer) → Redis (task queue) → Celery Workers (scraping) ↓ PostgreSQL / SQLite (results storage) FastAPI handles HTTP requests and returns immediate job IDs. Redis acts as the message broker. Celery workers run the actual scraping in the background. Results are stored in a database and fetched when the client polls. This architecture means: your API never blocks, workers can scale horizontally, failed jobs retry automatically, and your scraping logic is completely decoupled from your HTTP layer. Project Structure scraping_api/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI application │ ├── models.py # SQLAlchemy models │ ├── schemas.py # Pydantic request/response schemas │ ├── database.py # DB connection │ ├── tasks.py # Celery scraping tasks │ ├── scrapers/ │ │ ├── __init__.py │ │ ├── generic.py # Generic HTTP scraper │ │ └── browser.py # Playwright scraper │ └── middleware/ │ ├── ratelimit.py # Rate limiting │ └── auth.py # API key auth ├── worker.py # Celery worker entrypoint ├── requirements.txt ├── docker-compose.yml └── .env Step 1: Install Dependencies pip install fastapi uvicorn celery redis httpx \\ beautifulsoup4 sqlalchemy aiosqlite \\ python-dotenv pydantic slowapi # requirements.txt fastapi==0.115.0 uvicorn[standard]==0.30.0 celery[redis]==5.4.0 redis==5.0.1 httpx==0.27.0 beautifulsoup4==4.12.3 lxml==5.2.2 sqlalchemy==2.0.30 aiosqlite==0.20.0 python-dotenv==1.0.1 pydantic==2.7.0 slowapi==0.1.9 flower==2.0.1 Step 2: Database Models # app/models.py from sqlalchemy import Column, String, Float, Text, DateTime, Integer, Enum from sqlalchemy.orm import DeclarativeBase from datetime import datetime, timezone import enum, uuid class Base(DeclarativeBase): pass class JobStatus(str, enum.Enum): PENDING = \"pending\" RUNNING = \"running\" SUCCESS = \"success\" FAILED = \"failed\" RETRYING = \"retrying\" class ScrapeJob(Base): __tablename__ = \"scrape_jobs\" id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4())) url = Column(String, nullable=False) scraper_type = Column(String, default=\"generic\") # \"generic\" | \"browser\" status = Column(String, default=JobStatus.PENDING) created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc)) started_at = Column(DateTime, nullable=True) completed_at = Column(DateTime, nullable=True) api_key = Column(String, nullable=True) error = Column(Text, nullable=True) retry_count = Column(Integer, default=0) class ScrapeResult(Base): __tablename__ = \"scrape_results\" id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4())) job_id = Column(String, nullable=False, index=True) url = Column(String, nullable=False) title = Column(String, nullable=True) content = Column(Text, nullable=True) html_len = Column(Integer, nullable=True) status_code= Column(Integer, nullable=True) scraped_at = Column(DateTime, default=lambda: datetime.now(timezone.utc)) metadata_ = Column(Text, nullable=True) # JSON string for extra fields Step 3: Pydantic Schemas # app/schemas.py from pydantic import BaseModel, HttpUrl, Field from typing import Optional, List from datetime import datetime from app.models import JobStatus class ScrapeRequest(BaseModel): url: HttpUrl scraper_type: str = Field(\"generic\", pattern=\"^(generic|browser)$\") css_selectors: Optional[dict] = None # {\"title\": \"h1\", \"price\": \".price\"} wait_for: Optional[str] = None # CSS selector to wait for (browser mode) webhook_url: Optional[HttpUrl] = None # POST results here when done model_config = {\"json_schema_extra\": { \"example\": { \"url\": \"https://books.toscrape.com/\", \"scraper_type\": \"generic\", \"css_selectors\": {\"title\": \"h1\", \"books\": \".product_pod h3 a\"} } }} class BulkScrapeRequest(BaseModel): urls: List[HttpUrl] scraper_type: str = \"generic\" css_selectors: Optional[dict] = None class JobResponse(BaseModel): job_id: str status: JobStatus url: str created_at: datetime message: str = \"Job queued successfully\" class JobStatusResponse(BaseModel): job_id: str status: JobStatus url: str created_at: datetime started_at: Optional[datetime] completed_at: Optional[datetime] retry_count: int error: Optional[str] class ScrapeResultResponse(BaseModel): job_id: str url: str title: Optional[str] content: Optional[str] html_len: Optional[int] status_code: Optional[int] scraped_at: datetime data: Optional[dict] # Parsed CSS selector results Step 4: The Scraping Logic # app/scrapers/generic.py import httpx import asyncio import random from bs4 import BeautifulSoup from curl_cffi.requests import AsyncSession from datetime import datetime, timezone from typing import Optional HEADERS = { \"User-Agent\": ( \"Mozilla/5.0 (Windows NT 10.0; Win64; x64) \" \"AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36\" ), \"Accept-Language\": \"en-US,en;q=0.9\", \"Accept\": \"text/html,application/xhtml+xml,*/*;q=0.8\", } async def scrape_generic( url: str, css_selectors: Optional[dict] = None, max_retries: int = 3 ) -> dict: \"\"\" Generic HTTP scraper using curl_cffi for TLS impersonation. Returns title, content, status_code, and any custom selector results. \"\"\" for attempt in range(max_retries): try: async with AsyncSession(impersonate=\"chrome120\") as session: r = await session.get(url, headers=HEADERS, timeout=20) result = { \"url\": url, \"status_code\": r.status_code, \"html_len\": len(r.text), \"scraped_at\": datetime.now(timezone.utc).isoformat(), \"data\": {}, } if r.status_code != 200: result[\"error\"] = f\"HTTP {r.status_code}\" return result soup = BeautifulSoup(r.text, \"lxml\") # Extract title title_el = soup.find(\"title\") result[\"title\"] = title_el.get_text(strip=True) if title_el else None # Extract body text (clean) for tag in soup([\"script\", \"style\", \"nav\", \"footer\", \"header\"]): tag.decompose() result[\"content\"] = soup.get_text(separator=\" \", strip=True)[:5000] # Apply custom CSS selectors if provided if css_selectors: for key, selector in css_selectors.items(): elements = soup.select(selector) result[\"data\"][key] = [ el.get_text(strip=True) for el in elements ] if len(elements) > 1 else ( elements[0].get_text(strip=True) if elements else None ) return result except Exception as e: wait = 2 ** attempt + random.random() if attempt < max_retries - 1: await asyncio.sleep(wait) else: return { \"url\": url, \"status_code\": None, \"error\": str(e), \"scraped_at\": datetime.now(timezone.utc).isoformat(), \"data\": {}, } # app/scrapers/browser.py import asyncio import random from playwright.async_api import async_playwright from playwright_stealth import stealth_async from datetime import datetime, timezone from typing import Optional async def scrape_browser( url: str, css_selectors: Optional[dict] = None, wait_for: Optional[str] = None, ) -> dict: \"\"\" Playwright-based browser scraper for JavaScript-rendered pages. \"\"\" async with async_playwright() as p: browser = await p.chromium.launch( headless=True, args=[\"--no-sandbox\", \"--disable-blink-features=AutomationControlled\"] ) context = await browser.new_context( user_agent=\"Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120\", viewport={\"width\": 1280, \"height\": 800} ) # Block unnecessary resources await context.route( \"**/*.{png,jpg,gif,woff,woff2}\", lambda route: route.abort() ) page = await context.new_page() await stealth_async(page) result = { \"url\": url, \"scraped_at\": datetime.now(timezone.utc).isoformat(), \"data\": {}, } try: await page.goto(url, wait_until=\"domcontentloaded\", timeout=30000) if wait_for: await page.wait_for_selector(wait_for, timeout=10000) else: await asyncio.sleep(random.uniform(1.5, 3.0)) result[\"title\"] = await page.title() result[\"status_code\"] = 200 result[\"html_len\"] = len(await page.content()) # Apply CSS selectors if css_selectors: for key, selector in css_selectors.items(): elements = await page.query_selector_all(selector) texts = [await el.inner_text() for el in elements] result[\"data\"][key] = texts if len(texts) > 1 else ( texts[0] if texts else None ) except Exception as e: result[\"error\"] = str(e) result[\"status_code\"] = None finally: await browser.close() return result Step 5: Celery Tasks # app/tasks.py import asyncio import json from celery import Celery from datetime import datetime, timezone from app.scrapers.generic import scrape_generic from app.scrapers.browser import scrape_browser import httpx # Celery app — connects to Redis as broker and result backend celery_app = Celery( \"scraping_api\", broker=\"redis://localhost:6379/0\", backend=\"redis://localhost:6379/1\", ) celery_app.conf.update( task_serializer = \"json\", result_serializer = \"json\", accept_content = [\"json\"], result_expires = 3600, # Results expire after 1 hour task_acks_late = True, # Only ack after task completes worker_prefetch_multiplier = 1, # One task per worker at a time task_track_started = True, task_soft_time_limit = 60, # Warn after 60s task_time_limit = 90, # Kill after 90s # Route browser tasks to a dedicated queue task_routes = { \"app.tasks.scrape_url_task\": {\"queue\": \"generic\"}, \"app.tasks.scrape_browser_task\": {\"queue\": \"browser\"}, \"app.tasks.scrape_bulk_task\": {\"queue\": \"bulk\"}, }, ) def run_async(coro): \"\"\"Run an async coroutine from a sync Celery task.\"\"\" loop = asyncio.new_event_loop() try: return loop.run_until_complete(coro) finally: loop.close() @celery_app.task( bind=True, name=\"app.tasks.scrape_url_task\", max_retries=3, default_retry_delay=30, autoretry_for=(Exception,), retry_backoff=True, ) def scrape_url_task( self, job_id: str, url: str, css_selectors: dict = None, webhook_url: str = None, ): \"\"\" Celery task: scrape a single URL using the generic HTTP scraper. Automatically retries up to 3 times on failure. \"\"\" from app.database import SessionLocal from app.models import ScrapeJob, ScrapeResult, JobStatus db = SessionLocal() try: # Mark job as running job = db.query(ScrapeJob).filter(ScrapeJob.id == job_id).first() if job: job.status = JobStatus.RUNNING job.started_at = datetime.now(timezone.utc) db.commit() # Run the scraper result = run_async(scrape_generic(url, css_selectors)) # Save result scrape_result = ScrapeResult( job_id = job_id, url = url, title = result.get(\"title\"), content = result.get(\"content\"), html_len = result.get(\"html_len\"), status_code = result.get(\"status_code\"), metadata_ = json.dumps(result.get(\"data\", {})), ) db.add(scrape_result) # Update job status if job: job.status = JobStatus.SUCCESS if result.get(\"status_code\") == 200 \\ else JobStatus.FAILED job.completed_at = datetime.now(timezone.utc) job.error = result.get(\"error\") db.commit() # Fire webhook if configured if webhook_url: run_async(_fire_webhook(webhook_url, job_id, result)) return {\"job_id\": job_id, \"status\": \"success\"} except Exception as exc: if job: job.status = JobStatus.RETRYING job.retry_count = (job.retry_count or 0) + 1 db.commit() db.close() raise self.retry(exc=exc, countdown=2 ** self.request.retries) finally: db.close() @celery_app.task(name=\"app.tasks.scrape_browser_task\", max_retries=2) def scrape_browser_task( job_id: str, url: str, css_selectors: dict = None, wait_for: str = None, webhook_url: str = None, ): \"\"\"Celery task: scrape with Playwright browser.\"\"\" from app.database import SessionLocal from app.models import ScrapeJob, ScrapeResult, JobStatus db = SessionLocal() try: job = db.query(ScrapeJob).filter(ScrapeJob.id == job_id).first() if job: job.status = JobStatus.RUNNING job.started_at = datetime.now(timezone.utc) db.commit() result = run_async(scrape_browser(url, css_selectors, wait_for)) scrape_result = ScrapeResult( job_id = job_id, url = url, title = result.get(\"title\"), content = None, html_len = result.get(\"html_len\"), status_code = result.get(\"status_code\"), metadata_ = json.dumps(result.get(\"data\", {})), ) db.add(scrape_result) if job: job.status = JobStatus.SUCCESS job.completed_at = datetime.now(timezone.utc) db.commit() if webhook_url: run_async(_fire_webhook(webhook_url, job_id, result)) return {\"job_id\": job_id, \"status\": \"success\"} finally: db.close() async def _fire_webhook(webhook_url: str, job_id: str, data: dict): \"\"\"POST results to a webhook URL when a job completes.\"\"\" async with httpx.AsyncClient() as client: try: await client.post( webhook_url, json={\"job_id\": job_id, \"result\": data}, timeout=10 ) except Exception as e: print(f\"Webhook delivery failed for {job_id}: {e}\") Step 6: The FastAPI Application # app/main.py import json from typing import List, Optional from fastapi import FastAPI, Depends, HTTPException, Header, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded from sqlalchemy.orm import Session from app.database import get_db, engine from app.models import Base, ScrapeJob, ScrapeResult, JobStatus from app.schemas import ( ScrapeRequest, BulkScrapeRequest, JobResponse, JobStatusResponse, ScrapeResultResponse ) from app.tasks import scrape_url_task, scrape_browser_task # Create all tables Base.metadata.create_all(bind=engine) # Rate limiter — 60 requests per minute per IP limiter = Limiter(key_func=get_remote_address) app = FastAPI( title=\"Python Scraping API\", description=\"Production-grade web scraping as a service\", version=\"1.0.0\", ) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) app.add_middleware( CORSMiddleware, allow_origins=[\"*\"], allow_methods=[\"*\"], allow_headers=[\"*\"], ) # Simple API key auth VALID_API_KEYS = {\"dev-key-123\", \"prod-key-456\"} # In prod: load from DB or env def verify_api_key(x_api_key: str = Header(...)): if x_api_key not in VALID_API_KEYS: raise HTTPException(status_code=401, detail=\"Invalid API key\") return x_api_key # ── Endpoints ───────────────────────────────────────────────── @app.get(\"/health\") async def health(): return {\"status\": \"ok\", \"version\": \"1.0.0\"} @app.post(\"/scrape\", response_model=JobResponse, status_code=202) @limiter.limit(\"30/minute\") async def submit_scrape_job( request: Request, payload: ScrapeRequest, db: Session = Depends(get_db), api_key: str = Depends(verify_api_key), ): \"\"\" Submit a URL for scraping. Returns a job_id immediately. Poll /jobs/{job_id} for status, /jobs/{job_id}/result for data. \"\"\" job = ScrapeJob( url = str(payload.url), scraper_type = payload.scraper_type, api_key = api_key, ) db.add(job) db.commit() db.refresh(job) # Dispatch to appropriate Celery queue task_kwargs = { \"job_id\": job.id, \"url\": str(payload.url), \"css_selectors\": payload.css_selectors, \"webhook_url\": str(payload.webhook_url) if payload.webhook_url else None, } if payload.scraper_type == \"browser\": task_kwargs[\"wait_for\"] = payload.wait_for scrape_browser_task.apply_async( kwargs=task_kwargs, queue=\"browser\", task_id=job.id ) else: scrape_url_task.apply_async( kwargs=task_kwargs, queue=\"generic\", task_id=job.id ) return JobResponse( job_id = job.id, status = JobStatus.PENDING, url = str(payload.url), created_at = job.created_at, ) @app.post(\"/scrape/bulk\", status_code=202) @limiter.limit(\"5/minute\") async def submit_bulk_scrape( request: Request, payload: BulkScrapeRequest, db: Session = Depends(get_db), api_key: str = Depends(verify_api_key), ): \"\"\"Submit multiple URLs at once. Returns list of job_ids.\"\"\" if len(payload.urls) > 50: raise HTTPException(400, \"Max 50 URLs per bulk request\") job_ids = [] for url in payload.urls: job = ScrapeJob(url=str(url), scraper_type=payload.scraper_type, api_key=api_key) db.add(job) db.commit() db.refresh(job) scrape_url_task.apply_async( kwargs={\"job_id\": job.id, \"url\": str(url), \"css_selectors\": payload.css_selectors}, queue=\"bulk\", ) job_ids.append(job.id) return {\"submitted\": len(job_ids), \"job_ids\": job_ids} @app.get(\"/jobs/{job_id}\", response_model=JobStatusResponse) async def get_job_status(job_id: str, db: Session = Depends(get_db)): \"\"\"Check the status of a scrape job.\"\"\" job = db.query(ScrapeJob).filter(ScrapeJob.id == job_id).first() if not job: raise HTTPException(404, f\"Job {job_id} not found\") return job @app.get(\"/jobs/{job_id}/result\", response_model=ScrapeResultResponse) async def get_job_result(job_id: str, db: Session = Depends(get_db)): \"\"\"Fetch the scraped data for a completed job.\"\"\" job = db.query(ScrapeJob).filter(ScrapeJob.id == job_id).first() if not job: raise HTTPException(404, f\"Job {job_id} not found\") if job.status not in (JobStatus.SUCCESS, JobStatus.FAILED): raise HTTPException(202, f\"Job is still {job.status}\") result = db.query(ScrapeResult).filter(ScrapeResult.job_id == job_id).first() if not result: raise HTTPException(404, \"No result found for this job\") return ScrapeResultResponse( job_id = job_id, url = result.url, title = result.title, content = result.content, html_len = result.html_len, status_code = result.status_code, scraped_at = result.scraped_at, data = json.loads(result.metadata_ or \"{}\"), ) @app.get(\"/jobs\") async def list_jobs( status: Optional[str] = None, limit: int = 20, offset: int = 0, db: Session = Depends(get_db), ): \"\"\"List all scrape jobs with optional status filter.\"\"\" query = db.query(ScrapeJob) if status: query = query.filter(ScrapeJob.status == status) total = query.count() jobs = query.order_by(ScrapeJob.created_at.desc()).offset(offset).limit(limit).all() return {\"total\": total, \"jobs\": jobs} @app.delete(\"/jobs/{job_id}\") async def cancel_job(job_id: str, db: Session = Depends(get_db)): \"\"\"Cancel a pending job.\"\"\" job = db.query(ScrapeJob).filter(ScrapeJob.id == job_id).first() if not job: raise HTTPException(404, f\"Job {job_id} not found\") if job.status != JobStatus.PENDING: raise HTTPException(400, f\"Cannot cancel a {job.status} job\") from app.tasks import celery_app as celery celery.control.revoke(job_id, terminate=True) job.status = JobStatus.FAILED job.error = \"Cancelled by user\" db.commit() return {\"message\": f\"Job {job_id} cancelled\"} Step 7: Database Setup # app/database.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker import os DATABASE_URL = os.getenv(\"DATABASE_URL\", \"sqlite:///./scraping_api.db\") engine = create_engine( DATABASE_URL, connect_args={\"check_same_thread\": False} if \"sqlite\" in DATABASE_URL else {} ) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) def get_db(): db = SessionLocal() try: yield db finally: db.close() Step 8: Docker Compose # docker-compose.yml version: \"3.9\" services: redis: image: redis:7-alpine ports: - \"6379:6379\" healthcheck: test: [\"CMD\", \"redis-cli\", \"ping\"] interval: 5s timeout: 3s retries: 5 api: build: . command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload ports: - \"8000:8000\" environment: - DATABASE_URL=sqlite:///./scraping_api.db - REDIS_URL=redis://redis:6379/0 depends_on: redis: condition: service_healthy volumes: - .:/app worker_generic: build: . command: celery -A app.tasks.celery_app worker -Q generic -c 8 --loglevel=info environment: - REDIS_URL=redis://redis:6379/0 depends_on: - redis volumes: - .:/app worker_browser: build: . command: celery -A app.tasks.celery_app worker -Q browser -c 2 --loglevel=info environment: - REDIS_URL=redis://redis:6379/0 depends_on: - redis volumes: - .:/app worker_bulk: build: . command: celery -A app.tasks.celery_app worker -Q bulk -c 4 --loglevel=info environment: - REDIS_URL=redis://redis:6379/0 depends_on: - redis volumes: - .:/app flower: build: . command: celery -A app.tasks.celery_app flower --port=5555 ports: - \"5555:5555\" depends_on: - redis Step 9: Using the API Start everything: docker-compose up --build Submit a scrape job: # Submit a scrape job curl -X POST http://localhost:8000/scrape \\ -H \"Content-Type: application/json\" \\ -H \"X-API-Key: dev-key-123\" \\ -d '{ \"url\": \"https://books.toscrape.com/\", \"scraper_type\": \"generic\", \"css_selectors\": { \"title\": \"h1\", \"books\": \".product_pod h3 a\" } }' # Response: # {\"job_id\": \"abc-123\", \"status\": \"pending\", \"url\": \"https://...\"} # Check status curl http://localhost:8000/jobs/abc-123 # Get results when complete curl http://localhost:8000/jobs/abc-123/result Python client: import httpx import time BASE = \"http://localhost:8000\" KEY = \"dev-key-123\" def scrape(url: str, selectors: dict = None) -> dict: \"\"\"Submit a scrape job and poll until complete.\"\"\" r = httpx.post( f\"{BASE}/scrape\", json={\"url\": url, \"scraper_type\": \"generic\", \"css_selectors\": selectors}, headers={\"X-API-Key\": KEY} ) job_id = r.json()[\"job_id\"] # Poll for completion for _ in range(30): status_r = httpx.get(f\"{BASE}/jobs/{job_id}\") status = status_r.json()[\"status\"] if status in (\"success\", \"failed\"): break print(f\" Status: {status}...\") time.sleep(2) # Fetch result result_r = httpx.get(f\"{BASE}/jobs/{job_id}/result\") return result_r.json() result = scrape( \"https://books.toscrape.com/\", selectors={\"books\": \".product_pod h3 a\"} ) print(f\"Title: {result['title']}\") print(f\"Books found: {len(result['data'].get('books', []))}\") Bulk scrape: # Submit 20 URLs at once urls = [f\"https://books.toscrape.com/catalogue/page-{i}.html\" for i in range(1, 21)] r = httpx.post( f\"{BASE}/scrape/bulk\", json={\"urls\": urls, \"scraper_type\": \"generic\"}, headers={\"X-API-Key\": KEY} ) print(f\"Submitted {r.json()['submitted']} jobs\") print(f\"Job IDs: {r.json()['job_ids'][:3]}...\") Step 10: Monitoring with Flower Flower dashboard provides real-time monitoring of your task queue system. Open http://localhost:5555 to see: Live task execution graph Worker status and load Task success/failure rates Queue depths per queue Task history and retry counts For production alerting, add Prometheus metrics: # Add to requirements.txt: celery-prometheus-exporter # Then scrape http://worker:9808/metrics with Prometheus Production Checklist Replace SQLite with PostgreSQL (DATABASE_URL=postgresql://...) Store API keys in database with rate limits per key Add request logging with structlog Set task_time_limit and task_soft_time_limit per task type Configure dead-letter queue for permanently failed tasks Add Redis Sentinel or Cluster for HA Redis Run behind Nginx with TLS termination Set CELERY_WORKER_CONCURRENCY per queue based on task type Add Sentry for error tracking Expose /metrics endpoint for Prometheus FAQ Q: Why use Celery instead of FastAPI BackgroundTasks? FastAPI's BackgroundTasks runs in the same process as the web server — if the server restarts, running tasks are lost. Celery with Redis persists tasks in the broker, supports retries, scales across multiple machines, and provides monitoring via Flower. For any serious scraping API, Celery is the right choice. Q: How many concurrent scrapers can I run? Generic HTTP scrapers (curl_cffi/httpx) are cheap — a single worker can handle 8–16 concurrent tasks. Browser scrapers (Playwright) are expensive — limit to 2 per worker and 1 worker per CPU core. Scale horizontally by adding more worker containers. Q: How do I handle proxy rotation in the task layer? Pass a proxy parameter through the task kwargs, or implement proxy selection inside the scraper using a ProxyPool class that round-robins or selects randomly on each task invocation.","contentHash":"sha256:0a8eecfa12c40406f9511b11e3d406ca27d0b95633cb515029a12e4455637edb","authorName":"ZyVOP","authorUrl":"https://zyvop.com/author/zyvop","authorSameAs":["https://zyvop.com","https://github.com/zyvop","https://x.com/zyvop1"],"category":"Scraping","tags":["turn scraper into api python","fastapi async scraping service","python scraping api tutorial fastapi","celery redis scraper background tasks"],"audience":"Developers, software engineers, and students learning Scraping","tone":"Instructional, practical, code-first","readingTimeMinutes":13,"wordCount":2812,"faqs":null,"primaryTopic":"Scraping","publishedAt":"2026-07-29T11:23:27.859Z","updatedAt":"2026-08-30T10:30:00.413Z","canonicalUrl":"https://zyvop.com/build-a-web-scraping-api-with-fastapi-celery-redis-2026-2osqv"}