Appearance
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: int2. 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
limitto control crawls - Use
maxAgefor caching - Batch operations where possible
- Monitor credit usage