Python SDK

from searchhive import SearchHive

# Initialize client with your API key
client = SearchHive(api_key="sk_live_your_key_here")

# Or use environment variable (recommended)
# export SEARCHHIVE_API_KEY="sk_live_your_key_here"
# client = SearchHive()

# SwiftSearch: Real-time web search
search_results = client.swift_search(
    query="Bitcoin price 2025 predictions",
    max_results=10,
    auto_scrape_top=3
)

print(f"Found {search_results.results_count} results")
print(f"Credits used: {search_results.credits_used}")

# Access results
for result in search_results.search_results:
    print(f"Title: {result.title}")
    print(f"URL: {result.link}")
    print(f"Snippet: {result.snippet}\n")

# ScrapeForge: Extract content from URLs
scraped_data = client.scrape_forge(
    url="https://coindesk.com/price/bitcoin"
)

print(f"Page title: {scraped_data.title}")
print(f"Content length: {len(scraped_data.text)}")

# DeepDive: Multi-source research
research = client.deep_dive(
    query="cryptocurrency market trends 2025",
    max_pages=5
)

print(f"Scraped {len(research.scraped_content)} pages")
if research.summary:
    print(f"Summary: {research.summary}")

from searchhive import SearchHive

# Basic configuration with API key
client = SearchHive(
    api_key="sk_live_your_key_here"
)

# Full configuration
client = SearchHive(
    api_key="sk_live_your_key_here",
    base_url="https://www.searchhive.dev/api/v1",  # Custom endpoint
    timeout=30,  # Request timeout in seconds
)

# Environment-based configuration (recommended)
# export SEARCHHIVE_API_KEY="sk_live_your_key_here"
import os
client = SearchHive()  # Reads SEARCHHIVE_API_KEY env var

# Check API health
health = client.health()
print(f"API Status: {health}")

# SwiftSearch examples

# Basic search
results = client.swift_search("Python web scraping tutorial")

# Advanced search with all options
results = client.swift_search(
    query="AI startups funding 2025",
    max_results=20,
    auto_scrape_top=5,           # Scrape top 5 results
    include_contacts=True,       # Extract contact info
    include_social=True,         # Find social profiles
)

# Access different result types
print("Search Results:")
for result in results.search_results:
    print(f"- {result.title} ({result.link})")

print(f"\nResults count: {results.results_count}")
print(f"Scraped count: {results.scraped_count}")
print(f"Credits used: {results.credits_used}")
print(f"Remaining credits: {results.remaining_credits}")

# Access scraped content (when auto_scrape_top is set)
print("\nScraped Content:")
for content in results.scraped_content:
    if not content.error:
        print(f"- {content.title}: {len(content.text)} chars")

# Access contacts (when include_contacts=True)
if results.contacts:
    for contact in results.contacts:
        print(f"- {contact.name} ({contact.email}) at {contact.company}")

# Access social profiles (when include_social=True)
if results.social_profiles:
    for profile in results.social_profiles:
        print(f"- {profile.platform}: {profile.url}")

# Handle errors gracefully
try:
    results = client.swift_search("test query")
except SearchHiveError as e:
    print(f"API Error: {e}")
except RateLimitError as e:
    print(f"Rate limited: {e}")
except AuthenticationError as e:
    print(f"Auth error: {e}")

# ScrapeForge examples

# Basic web scraping
scraped = client.scrape_forge(
    url="https://example.com/article"
)

# Advanced scraping with options
scraped = client.scrape_forge(
    url="https://news.ycombinator.com",
    extract=["title", "text", "links", "images", "metadata"],
    wait_for=".storylink",       # CSS selector to wait for
    use_browser=True,            # Use headless browser
    timeout=30,                  # Override timeout for this request
)

# Access extracted content
print(f"Title: {scraped.title}")
print(f"Text: {scraped.text[:500]}...")  # First 500 chars
print(f"Credits used: {scraped.credits_used}")
print(f"Remaining credits: {scraped.remaining_credits}")

# Links found on page
if scraped.links:
    print(f"\nFound {len(scraped.links)} links:")
    for link in scraped.links[:5]:
        print(f"- {link.text}: {link.url}")

# Images found on page
if scraped.images:
    print(f"\nFound {len(scraped.images)} images:")
    for img in scraped.images[:3]:
        print(f"- {img.alt}: {img.src}")

# Metadata
if scraped.metadata:
    print(f"\nMetadata: {scraped.metadata}")

# Handle errors
if scraped.error:
    print(f"Scrape error: {scraped.error}")

# Batch scraping multiple URLs
urls = [
    "https://example.com/page1",
    "https://example.com/page2",
    "https://example.com/page3"
]

batch_results = client.batch_scrape(urls, timeout=15000)
for result in batch_results:
    print(f"{result.url}: {result.title if not result.error else 'Failed: ' + result.error}")

# DeepDive research examples

# Basic research
research = client.deep_dive(
    query="quantum computing breakthroughs 2025"
)

# Comprehensive research with options
research = client.deep_dive(
    query="sustainable energy investment opportunities",
    max_pages=10,
    extract_content=True,
    include_domains=["reuters.com", "bloomberg.com"],
    exclude_domains=["spam-site.com"]
)

# Access research results
print(f"Sources analyzed: {len(research.search_results)}")
print(f"Pages scraped: {len(research.scraped_content)}")

# Search results
print("\nKey Sources:")
for result in research.search_results:
    print(f"- {result.title} ({result.link})")

# Scraped and analyzed content
print("\nContent Analysis:")
for content in research.scraped_content[:3]:
    if not content.error:
        summary = content.text[:200] + "..." if len(content.text) > 200 else content.text
        print(f"- {content.title}:\n  {summary}\n")

# AI-generated summary
if research.summary:
    print(f"Summary:\n{research.summary}")

# Credit tracking
print(f"Credits used: {research.credits_used}")
print(f"Remaining credits: {research.remaining_credits}")

import asyncio
from searchhive import AsyncSearchHive

async def main():
    # Async client for high-performance applications
    client = AsyncSearchHive(api_key="sk_live_your_key_here")
    
    try:
        # Async search
        results = await client.swift_search("Python async tutorial")
        print(f"Found {results.results_count} results")
        
        # Concurrent requests
        search_task = asyncio.create_task(client.swift_search("AI news"))
        scrape_task = asyncio.create_task(client.scrape_forge("https://example.com"))
        research_task = asyncio.create_task(client.deep_dive("machine learning trends"))
        
        # Wait for all to complete
        search_res, scrape_res, research_res = await asyncio.gather(
            search_task, scrape_task, research_task
        )
        
        print(f"Search: {search_res.results_count} results")
        print(f"Scrape: {scrape_res.title}")
        print(f"Research: {len(research_res.scraped_content)} sources")
        
    finally:
        await client.close()

# Run the async function
asyncio.run(main())

# Or use context manager (recommended)
async def with_context_manager():
    async with AsyncSearchHive(api_key="sk_live_your_key_here") as client:
        results = await client.swift_search("test query")
        return results

from searchhive import SearchHive
from searchhive.exceptions import SearchHiveError, AuthenticationError, RateLimitError

client = SearchHive(api_key="sk_live_your_key_here")

try:
    results = client.swift_search("test query")
    
except AuthenticationError:
    print("Invalid API key - check your credentials")
    
except RateLimitError as e:
    print(f"Rate limited: {e}")
    
except SearchHiveError as e:
    print(f"API error: {e}")
    
except Exception as e:
    print(f"Unexpected error: {e}")

# Retry logic with exponential backoff
import time
import random

def search_with_retry(query, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.swift_search(query)
        except RateLimitError as e:
            if attempt < max_retries - 1:
                wait_time = (2 ** attempt) + random.uniform(0, 1)
                print(f"Rate limited, retrying in {wait_time:.1f}s...")
                time.sleep(wait_time)
            else:
                raise
        except SearchHiveError as e:
            if attempt < max_retries - 1:
                wait_time = (2 ** attempt) + random.uniform(0, 1)
                print(f"API error, retrying in {wait_time:.1f}s...")
                time.sleep(wait_time)
            else:
                raise

# Usage
try:
    results = search_with_retry("Bitcoin price analysis")
    print(f"Success! Found {results.results_count} results")
except Exception as e:
    print(f"All retries failed: {e}")

# Response shapes

# SwiftSearchResponse
search_results = client.swift_search("test query")
print(search_results.query)              # str
print(search_results.search_results)     # list of results
print(search_results.scraped_content)    # list of scraped pages (when auto_scrape_top set)
print(search_results.contacts)           # Optional[list] (when include_contacts=True)
print(search_results.social_profiles)    # Optional[list] (when include_social=True)
print(search_results.credits_used)       # int
print(search_results.remaining_credits)  # int
print(search_results.results_count)      # int
print(search_results.scraped_count)      # int

# ScrapeForgeResponse
scrape_result = client.scrape_forge("https://example.com")
print(scrape_result.url)                 # str
print(scrape_result.title)               # str
print(scrape_result.text)                # str
print(scrape_result.links)               # list
print(scrape_result.images)              # list
print(scrape_result.metadata)            # dict
print(scrape_result.error)               # Optional[str]
print(scrape_result.credits_used)        # int
print(scrape_result.remaining_credits)   # int

# DeepDiveResponse
research = client.deep_dive("AI trends")
print(research.search_results)           # list
print(research.scraped_content)          # list
print(research.summary)                  # Optional[str]
print(research.credits_used)             # int
print(research.remaining_credits)        # int

# Convert to dict/JSON for storage
data_dict = search_results.model_dump()
json_str = search_results.model_dump_json()