Skip to content

Workflow: Data Pipeline

Fresh 🌱

Extract, process, and store web data at scale.

Architecture

Implementation

1. Define Data Schema

python
from pydantic import BaseModel

class WebpageData(BaseModel):
    title: str
    url: str
    content: str
    extracted_at: str
    word_count: int

2. Extract Data

python
from firecrawl import Firecrawl
from datetime import datetime

firecrawl = Firecrawl(api_key="fc-YOUR-API-KEY")

urls = [
    "https://example.com/1",
    "https://example.com/2",
    "https://example.com/3"
]

extracted_data = []

for url in urls:
    result = firecrawl.scrape(url, formats=["markdown"])

    data = WebpageData(
        title=result['metadata']['title'],
        url=result['metadata']['sourceURL'],
        content=result['markdown'],
        extracted_at=datetime.now().isoformat(),
        word_count=len(result['markdown'].split())
    )

    extracted_data.append(data)

3. Store Data

To JSON:

python
import json

with open("data.json", "w") as f:
    json.dump([d.dict() for d in extracted_data], f)

To Database:

python
import sqlite3

conn = sqlite3.connect("website_data.db")
cursor = conn.cursor()

for data in extracted_data:
    cursor.execute("""
        INSERT INTO pages (title, url, content, extracted_at)
        VALUES (?, ?, ?, ?)
    """, (data.title, data.url, data.content, data.extracted_at))

conn.commit()

4. Query & Use

python
# Find pages by title
relevant = [d for d in extracted_data if "keyword" in d.title.lower()]

# Statistics
total_words = sum(d.word_count for d in extracted_data)
print(f"Total words extracted: {total_words}")

Full Example

python
from firecrawl import Firecrawl
from pydantic import BaseModel
import json
from datetime import datetime

class WebData(BaseModel):
    title: str
    url: str
    content: str
    extracted_at: str

firecrawl = Firecrawl(api_key="YOUR_API_KEY")

# Define URLs
urls = ["https://example.com/page1", "https://example.com/page2"]

# Extract
results = []
for url in urls:
    try:
        data = firecrawl.scrape(url)
        results.append(WebData(
            title=data['metadata']['title'],
            url=url,
            content=data['markdown'],
            extracted_at=datetime.now().isoformat()
        ))
    except Exception as e:
        print(f"Error scraping {url}: {e}")

# Save
with open("results.json", "w") as f:
    json.dump([r.dict() for r in results], f, indent=2)

print(f"Extracted {len(results)} pages")

Cost Optimization

  • Set limit to control crawls
  • Use maxAge for caching
  • Batch operations where possible
  • Monitor credit usage

← Back to Workflows