القائمة
Performance optimization patterns for Python scraper pipelines — connection pooling, memory efficiency, async I/O, and batch operations for high-throughput data ingestion.
التثبيت باستخدام Codex أو Claude انسخ هذا Prompt والصقه في Codex أو Claude أو مساعد آخر ليراجع صفحة Skill ويثبّتها لك.
استنادا إلى تصنيف SOC المهني
Use this skill when expanding the scraper network to new Florida counties. Contains the 67-county expansion roadmap, prioritization framework, site reconnaissance procedure, and lead analysis strategy. The goal is total Florida coverage for maximum bail bond lead generation.
Use this skill when creating a new county arrest scraper, modifying an existing solver, or adding a new county to the scraping pipeline. Contains the canonical architecture, naming conventions, and integration checklist.
Non-negotiable rules, safe refactoring patterns, shared code modification guidelines, secrets management, config hierarchy, and schema change process for the swfl-arrest-scrapers repo. Read this before modifying any core infrastructure.
Testing patterns, fixture guidelines, parser test templates, smoke tests, and red-green TDD methodology for the scraper pipeline. Use when writing tests for new counties, verifying core/ changes, or debugging test failures.
Google Sheets: Read and write spreadsheets via the Sheets API v4.
Systematically strengthen scraper pipelines against network failures, anti-bot measures, data edge cases, and real-world brittleness that breaks idealized scraper designs.
| name | python-performance-optimization |
| description | Performance optimization patterns for Python scraper pipelines — connection pooling, memory efficiency, async I/O, and batch operations for high-throughput data ingestion. |
# Reuse sessions across requests — don't create new connections per page
session = requests.Session()
session.headers.update(HEADERS)
# Session persists TCP connections, cookies, and auth
# Process records in streams, not full lists
def process_records(records):
for record in records:
yield normalize(record) # Generator, not list comprehension
# Use __slots__ for high-volume record objects
class ArrestRecord:
__slots__ = ['name', 'booking_number', 'dob', 'charges', 'bond']
# ❌ Bad: One API call per record
for record in records:
sheet.append_row(record)
# ✅ Good: Batch all records in single API call
sheet.batch_update([{
'range': f'{tab}!A2',
'values': [record_to_row(r) for r in records]
}])
# O(1) lookup instead of O(n) scan
existing_ids = set(sheet.col_values(booking_number_col))
new_records = [r for r in records if r['Booking_Number'] not in existing_ids]
from concurrent.futures import ThreadPoolExecutor
def scrape_detail_page(url):
return session.get(url, timeout=30).text
with ThreadPoolExecutor(max_workers=3) as executor:
results = list(executor.map(scrape_detail_page, detail_urls))
# Time the entire pipeline
time python counties/lee/runner.py --days-back 1
# Profile bottlenecks
python -m cProfile -s cumulative counties/lee/runner.py