diff --git a/apps/python-sdk/.env.example b/apps/python-sdk/.env.example index 7bea7e7f2..81394d178 100644 --- a/apps/python-sdk/.env.example +++ b/apps/python-sdk/.env.example @@ -1 +1,2 @@ -FIRECRAWL_API_KEY= \ No newline at end of file +FIRECRAWL_API_KEY= +FIRECRAWL_API_URL=https://api.firecrawl.dev \ No newline at end of file diff --git a/apps/python-sdk/example.py b/apps/python-sdk/example.py index 76535654e..29dc8e9e4 100644 --- a/apps/python-sdk/example.py +++ b/apps/python-sdk/example.py @@ -4,6 +4,7 @@ Example demonstrating the v2 search functionality with individual parameters. """ import os +import time from dotenv import load_dotenv from firecrawl import Firecrawl from firecrawl.v2.types import ScrapeOptions, ScrapeFormats @@ -15,21 +16,34 @@ def main(): api_key = os.getenv("FIRECRAWL_API_KEY") if not api_key: raise ValueError("FIRECRAWL_API_KEY is not set") + + api_url = os.getenv("FIRECRAWL_API_URL") + if not api_url: + raise ValueError("FIRECRAWL_API_URL is not set") - firecrawl = Firecrawl(api_key=api_key) + firecrawl = Firecrawl(api_key=api_key, api_url=api_url) + + # crawl + crawl_response = firecrawl.crawl("docs.firecrawl.dev", limit=5) + print(crawl_response) + + crawl_job = firecrawl.start_crawl('docs.firecrawl.dev', limit=5) + print(crawl_job) + + while (crawl_job.status != 'completed'): + crawl_job = firecrawl.get_crawl_status(crawl_job.id) + time.sleep(2) + + print(crawl_job) # search examples search_response = firecrawl.search( query="What is the capital of France?", - sources=[ - { type: "web" }, - { type: "news" }, - { type: "images" } - ], - limit=10) + sources=["web", "news", "images"], + limit=10 + ) + print(search_response) - - if __name__ == "__main__": main() \ No newline at end of file diff --git a/apps/python-sdk/firecrawl/__tests__/e2e/v2/test_crawl.py b/apps/python-sdk/firecrawl/__tests__/e2e/v2/test_crawl.py new file mode 100644 index 000000000..d4139d312 --- /dev/null +++ b/apps/python-sdk/firecrawl/__tests__/e2e/v2/test_crawl.py @@ -0,0 +1,189 @@ +import pytest +import os +from dotenv import load_dotenv +from firecrawl import Firecrawl +from firecrawl.v2.types import ScrapeOptions + +load_dotenv() + +if not os.getenv("API_KEY"): + raise ValueError("API_KEY is not set") + +if not os.getenv("API_URL"): + raise ValueError("API_URL is not set") + +class TestCrawlE2E: + """End-to-end tests for crawl functionality.""" + + def setup_method(self): + """Set up test client.""" + self.client = Firecrawl(api_key=os.getenv("API_KEY"), api_url=os.getenv("API_URL")) + + def test_start_crawl_minimal_request(self): + """Test starting a crawl with minimal parameters.""" + crawl_job = self.client.start_crawl("https://example.com") + + # Check response structure + assert crawl_job.id is not None + assert crawl_job.status in ["scraping", "completed", "failed"] + + def test_start_crawl_with_options(self): + """Test starting a crawl with options.""" + crawl_job = self.client.start_crawl( + "https://example.com", + limit=5, + max_discovery_depth=2 + ) + + assert crawl_job.id is not None + + def test_start_crawl_with_prompt(self): + """Test starting a crawl with prompt.""" + crawl_job = self.client.start_crawl( + "https://example.com", + prompt="Extract all blog posts" + ) + + assert crawl_job.id is not None + + def test_get_crawl_status(self): + """Test getting crawl status.""" + # First start a crawl + start_job = self.client.start_crawl("https://example.com") + assert start_job.id is not None + + job_id = start_job.id + + # Get status + status_job = self.client.get_crawl_status(job_id) + + assert status_job.status in ["scraping", "completed", "failed"] + assert status_job.current >= 0 + assert status_job.total >= 0 + assert isinstance(status_job.data, list) + + def test_cancel_crawl(self): + """Test canceling a crawl.""" + # First start a crawl + start_job = self.client.start_crawl("https://example.com") + assert start_job.id is not None + + job_id = start_job.id + + # Cancel the crawl + cancel_job = self.client.cancel_crawl(job_id) + print(f"DEBUG: cancel_job: {cancel_job}") + + assert cancel_job.status == "failed" + + def test_crawl_with_wait(self): + """Test crawl with wait for completion.""" + crawl_job = self.client.crawl( + "docs.firecrawl.dev", + limit=3, + max_discovery_depth=2, + poll_interval=1, + timeout=60 + ) + + assert crawl_job.status in ["completed", "failed"] + assert crawl_job.current >= 0 + assert crawl_job.total >= 0 + assert isinstance(crawl_job.data, list) + + def test_crawl_with_prompt_and_wait(self): + """Test crawl with prompt and wait for completion.""" + crawl_job = self.client.crawl( + "https://example.com", + prompt="Extract all blog posts", + poll_interval=1, + timeout=30 + ) + + assert crawl_job.status in ["completed", "failed"] + assert crawl_job.current >= 0 + assert crawl_job.total >= 0 + assert isinstance(crawl_job.data, list) + + def test_crawl_with_scrape_options(self): + """Test crawl with scrape options.""" + scrape_opts = ScrapeOptions( + formats=["markdown"], + only_main_content=False, + mobile=True + ) + + crawl_job = self.client.start_crawl( + "https://example.com", + limit=2, + scrape_options=scrape_opts + ) + + assert crawl_job.id is not None + + def test_crawl_all_parameters(self): + """Test crawl with all possible parameters.""" + scrape_opts = ScrapeOptions( + formats=["markdown", "html"], + headers={"User-Agent": "Test Bot"}, + include_tags=["h1", "h2"], + exclude_tags=["nav"], + only_main_content=False, + timeout=15000, + wait_for=2000, + mobile=True, + skip_tls_verification=True, + remove_base64_images=False + # Note: raw_html and screenshot_full_page are not supported by v2 API yet + ) + + crawl_job = self.client.start_crawl( + "https://example.com", + prompt="Extract all blog posts and documentation", + include_paths=["/blog/*", "/docs/*"], + exclude_paths=["/admin/*"], + max_discovery_depth=3, + ignore_sitemap=False, + limit=5, + crawl_entire_domain=True, + allow_external_links=False, + scrape_options=scrape_opts + ) + + assert crawl_job.id is not None + + def test_crawl_progress_callback(self): + """Test crawl with progress callback.""" + progress_calls = [] + + def progress_callback(status_data): + progress_calls.append(status_data) + + crawl_job = self.client.crawl( + "https://docs.firecrawl.dev", + limit=2, + poll_interval=1, + timeout=60, + progress_callback=progress_callback + ) + + # Progress callback should have been called at least once + assert len(progress_calls) > 0 + + # Check that callback received proper data + for call in progress_calls: + assert call.status in ["scraping", "completed", "failed"] + assert call.current >= 0 + assert call.total >= 0 + assert isinstance(call.data, list) + + def test_crawl_params(self): + """Test crawl_params function.""" + params_data = self.client.crawl_params( + "https://example.com", + "Extract all blog posts and documentation" + ) + + assert params_data is not None + # The LLM should return some reasonable options + assert params_data.limit is not None or params_data.include_paths is not None or params_data.max_discovery_depth is not None \ No newline at end of file diff --git a/apps/python-sdk/firecrawl/__tests__/e2e/v2/test_search.py b/apps/python-sdk/firecrawl/__tests__/e2e/v2/test_search.py index d85595397..7dbbc54f4 100644 --- a/apps/python-sdk/firecrawl/__tests__/e2e/v2/test_search.py +++ b/apps/python-sdk/firecrawl/__tests__/e2e/v2/test_search.py @@ -1,9 +1,7 @@ from firecrawl import Firecrawl import os -import pytest from dotenv import load_dotenv -# from pydantic import BaseModel -# from firecrawl.types import ScrapeOptions, JsonFormat, Location, WaitAction, SourceOptions, FormatOptions +from firecrawl.types import SearchData, SearchResult, Document, ScrapeFormats, ScrapeOptions load_dotenv() @@ -16,71 +14,198 @@ if not os.getenv("API_URL"): firecrawl = Firecrawl(api_key=os.getenv("API_KEY"), api_url=os.getenv("API_URL")) def test_search_minimal_request(): - search_response = firecrawl.search( - query="What is the capital of France?" + results = firecrawl.search( + query="What is the capital of France?" ) - print(f"Success: {search_response.success}") - print(f"Data: {search_response.data}") - print(f"Warning: {search_response.warning}") -# @pytest.mark.e2e -# def test_search_images_request(): -# search_response = firecrawl.search( -# query="What is the capital of France?", -# sources=[ "images" ], -# limit=5 -# ) + assert isinstance(results, SearchData) + assert hasattr(results, 'web') + assert hasattr(results, 'news') + assert hasattr(results, 'images') -# print(search_response) + assert results.web is not None + assert len(results.web) > 0 + + for result in results.web: + assert isinstance(result, SearchResult) + assert hasattr(result, 'url') + assert hasattr(result, 'title') + assert hasattr(result, 'description') + assert result.url.startswith('http') + assert result.title is not None + assert result.description is not None -# @pytest.mark.e2e -# def test_search_all_options(): -# class Schema(BaseModel): -# name: str -# description: str -# type: str -# required: bool + titles = [result.title.lower() for result in results.web] + descriptions = [result.description.lower() for result in results.web] + all_text = ' '.join(titles + descriptions) + + assert 'paris' in all_text + + assert results.news is None + assert results.images is None -# search_response = firecrawl.search( -# query="What is the capital of France?", -# sources=[ -# { type: "web" }, -# { type: "news" }, -# { type: "images" } -# ], -# limit=10, -# tbs="", -# location="", -# timeout=60000, -# ignore_invalid_urls=False, -# scrape_options=ScrapeOptions( -# formats=[ -# 'markdown', -# JsonFormat( -# schema={}, -# prompt="" -# ) -# ], -# only_main_content=True, -# # include_tags=[""], # TODO: check this -# # exclude_tags=[""], # TODO: check this -# max_age=0, -# headers={}, -# wait_for=0, -# mobile=False, -# skip_tls_verification=False, -# timeout=30000, -# parsers=[ 'pdf' ], -# actions=[ -# WaitAction(milliseconds=2000) -# ], -# location=Location( -# country="US", -# languages=["en-US"] -# ), -# remove_base64_images=True, -# block_ads=True, -# proxy="basic", -# store_in_cache=True, -# ) -# ) \ No newline at end of file + +def test_search_with_sources(): + """Test search with specific sources.""" + results = firecrawl.search( + query="firecrawl", + sources=["web", "news"], + limit=3 + ) + + assert isinstance(results, SearchData) + + assert results.web is not None + assert len(results.web) <= 3 + + if results.news is not None: + assert len(results.news) <= 3 + + assert results.images is None + + web_titles = [result.title.lower() for result in results.web] + web_descriptions = [result.description.lower() for result in results.web] + all_web_text = ' '.join(web_titles + web_descriptions) + + assert 'firecrawl' in all_web_text + +def test_search_result_structure(): + """Test that SearchResult objects have the correct structure.""" + results = firecrawl.search( + query="test query", + limit=1 + ) + + if results.web and len(results.web) > 0: + result = results.web[0] + + assert hasattr(result, 'url') + assert hasattr(result, 'title') + assert hasattr(result, 'description') + + assert isinstance(result.url, str) + assert isinstance(result.title, str) or result.title is None + assert isinstance(result.description, str) or result.description is None + + # Test URL format + assert result.url.startswith('http') + +def test_search_all_parameters(): + """Test search with all available parameters (comprehensive e2e test).""" + from firecrawl.types import ScrapeOptions, JsonFormat, Location, WaitAction + + # Define a schema for JSON extraction + schema = { + "type": "object", + "properties": { + "title": {"type": "string"}, + "description": {"type": "string"}, + "url": {"type": "string"} + }, + "required": ["title", "description"] + } + + results = firecrawl.search( + query="artificial intelligence", + sources=[ + {"type": "web"}, + {"type": "news"} + ], + limit=3, + tbs="qdr:m", # Last month + location="US", + ignore_invalid_urls=True, + timeout=60000, + scrape_options=ScrapeOptions( + formats=["markdown", "html"], + headers={"User-Agent": "Firecrawl-Test/1.0"}, + include_tags=["h1", "h2", "p"], + exclude_tags=["nav", "footer"], + only_main_content=True, + wait_for=2000, + mobile=False, + skip_tls_verification=False, + remove_base64_images=True, + block_ads=True, + proxy="basic", + max_age=3600000, # 1 hour cache + store_in_cache=True, + location=Location( + country="US", + languages=["en"] + ), + actions=[ + WaitAction(milliseconds=1000) + ] + # Note: raw_html and screenshot_full_page are not supported by v2 API yet + ) + ) + + # Test structure + assert isinstance(results, SearchData) + assert hasattr(results, 'web') + assert hasattr(results, 'news') + assert hasattr(results, 'images') + + # Test that web results exist + assert results.web is not None + assert len(results.web) <= 3 # Should respect limit + + # Test that results contain expected content + web_titles = [result.title.lower() for result in results.web if result.title] + web_descriptions = [result.description.lower() for result in results.web if result.description] + all_web_text = ' '.join(web_titles + web_descriptions) + + # Should contain AI-related terms (case insensitive) + ai_terms = ['artificial', 'intelligence', 'ai', 'machine', 'learning'] + assert any(term in all_web_text for term in ai_terms) + + # Test that each result has proper structure + for result in results.web: + assert isinstance(result, (SearchResult, Document)) + assert hasattr(result, 'url') + assert result.url.startswith('http') + + # If it's a Document (with scrape_options), check for additional fields + if isinstance(result, Document): + # Should have markdown or html content due to scrape_options + assert result.markdown is not None or result.html is not None + + # Test that news results exist (if API supports it) + if results.news is not None: + assert len(results.news) <= 3 + for result in results.news: + assert isinstance(result, (SearchResult, Document)) + assert result.url.startswith('http') + + # Test that unspecified sources are None + assert results.images is None + + +def test_search_formats_flexibility(): + """Test that both list and ScrapeFormats work for formats.""" + from firecrawl.types import ScrapeFormats + + # Test with list format + results1 = firecrawl.search( + query="python programming", + limit=1, + scrape_options=ScrapeOptions( + formats=["markdown"] + ) + ) + + # Test with ScrapeFormats object + results2 = firecrawl.search( + query="python programming", + limit=1, + scrape_options=ScrapeOptions( + formats=ScrapeFormats(markdown=True) + ) + ) + + # Both should work without errors + assert isinstance(results1, SearchData) + assert isinstance(results2, SearchData) + assert results1.web is not None + assert results2.web is not None \ No newline at end of file diff --git a/apps/python-sdk/firecrawl/__tests__/unit/v2/methods/crawl/test_crawl_params.py b/apps/python-sdk/firecrawl/__tests__/unit/v2/methods/crawl/test_crawl_params.py new file mode 100644 index 000000000..e665deb4c --- /dev/null +++ b/apps/python-sdk/firecrawl/__tests__/unit/v2/methods/crawl/test_crawl_params.py @@ -0,0 +1,196 @@ +import pytest +from unittest.mock import Mock, patch +from firecrawl.v2.types import CrawlParamsRequest, CrawlParamsResponse, CrawlParamsData +from firecrawl.v2.methods.crawl import crawl_params + + +class TestCrawlParams: + """Unit tests for crawl_params function.""" + + def test_crawl_params_success(self): + """Test successful crawl_params call.""" + # Mock client and response + mock_client = Mock() + mock_response = Mock() + mock_response.ok = True + mock_response.json.return_value = { + "success": True, + "data": { + "limit": 10, + "maxDiscoveryDepth": 3, + "ignoreSitemap": False + }, + "warning": None + } + mock_client.post.return_value = mock_response + + # Create request + request = CrawlParamsRequest( + url="https://example.com", + prompt="Extract all blog posts" + ) + + # Call function + result = crawl_params(mock_client, request) + + # Verify client call + mock_client.post.assert_called_once_with("/v2/crawl-params", { + "url": "https://example.com", + "prompt": "Extract all blog posts" + }) + + # Verify result + assert isinstance(result, CrawlParamsData) + assert result.limit == 10 + assert result.max_discovery_depth == 3 + assert result.ignore_sitemap is False + assert result.warning is None + + def test_crawl_params_api_error(self): + """Test crawl_params with API error.""" + # Mock client and response + mock_client = Mock() + mock_response = Mock() + mock_response.ok = False + mock_response.status_code = 400 + mock_response.text = "Bad Request" + mock_client.post.return_value = mock_response + + # Create request + request = CrawlParamsRequest( + url="https://example.com", + prompt="Extract all blog posts" + ) + + # Call function and expect exception + with pytest.raises(Exception, match="crawl params"): + crawl_params(mock_client, request) + + def test_crawl_params_success_false(self): + """Test crawl_params with success=False in response.""" + # Mock client and response + mock_client = Mock() + mock_response = Mock() + mock_response.ok = True + mock_response.json.return_value = { + "success": False, + "error": "Invalid URL provided" + } + mock_client.post.return_value = mock_response + + # Create request + request = CrawlParamsRequest( + url="https://example.com", + prompt="Extract all blog posts" + ) + + # Call function and expect exception + with pytest.raises(Exception, match="Invalid URL provided"): + crawl_params(mock_client, request) + + def test_crawl_params_empty_url(self): + """Test crawl_params with empty URL.""" + # Create request with empty URL + request = CrawlParamsRequest( + url="", + prompt="Extract all blog posts" + ) + + # Call function and expect exception + with pytest.raises(ValueError, match="URL cannot be empty"): + crawl_params(Mock(), request) + + def test_crawl_params_whitespace_url(self): + """Test crawl_params with whitespace-only URL.""" + # Create request with whitespace URL + request = CrawlParamsRequest( + url=" ", + prompt="Extract all blog posts" + ) + + # Call function and expect exception + with pytest.raises(ValueError, match="URL cannot be empty"): + crawl_params(Mock(), request) + + def test_crawl_params_empty_prompt(self): + """Test crawl_params with empty prompt.""" + # Create request with empty prompt + request = CrawlParamsRequest( + url="https://example.com", + prompt="" + ) + + # Call function and expect exception + with pytest.raises(ValueError, match="Prompt cannot be empty"): + crawl_params(Mock(), request) + + def test_crawl_params_whitespace_prompt(self): + """Test crawl_params with whitespace-only prompt.""" + # Create request with whitespace prompt + request = CrawlParamsRequest( + url="https://example.com", + prompt=" " + ) + + # Call function and expect exception + with pytest.raises(ValueError, match="Prompt cannot be empty"): + crawl_params(Mock(), request) + + def test_crawl_params_complex_options(self): + """Test crawl_params with complex options in response.""" + # Mock client and response + mock_client = Mock() + mock_response = Mock() + mock_response.ok = True + mock_response.json.return_value = { + "success": True, + "data": { + "includePaths": ["/blog/*", "/docs/*"], + "excludePaths": ["/admin/*"], + "maxDiscoveryDepth": 3, + "ignoreSitemap": False, + "limit": 50, + "crawlEntireDomain": True, + "allowExternalLinks": False, + "scrapeOptions": { + "formats": ["markdown"], + "onlyMainContent": False, + "mobile": True, + "timeout": None, + "waitFor": None, + "skipTlsVerification": False, + "removeBase64Images": True + } + } + } + mock_client.post.return_value = mock_response + + # Create request + request = CrawlParamsRequest( + url="https://example.com", + prompt="Extract all blog posts and documentation with mobile view" + ) + + # Call function + result = crawl_params(mock_client, request) + + # Verify result + assert result is not None + + # Check all fields + assert result.include_paths == ["/blog/*", "/docs/*"] + assert result.exclude_paths == ["/admin/*"] + assert result.max_discovery_depth == 3 + assert result.ignore_sitemap is False + assert result.limit == 50 + assert result.crawl_entire_domain is True + assert result.allow_external_links is False + + # Check nested scrape options + assert result.scrape_options is not None + assert result.scrape_options.formats is not None + # formats is a ScrapeFormats object, so we need to access its formats field + assert len(result.scrape_options.formats.formats) == 1 + assert result.scrape_options.formats.formats[0].type == "markdown" + assert result.scrape_options.only_main_content is False + assert result.scrape_options.mobile is True \ No newline at end of file diff --git a/apps/python-sdk/firecrawl/__tests__/unit/v2/methods/crawl/test_crawl_request_preparation.py b/apps/python-sdk/firecrawl/__tests__/unit/v2/methods/crawl/test_crawl_request_preparation.py new file mode 100644 index 000000000..849450fa8 --- /dev/null +++ b/apps/python-sdk/firecrawl/__tests__/unit/v2/methods/crawl/test_crawl_request_preparation.py @@ -0,0 +1,240 @@ +import pytest +from firecrawl.v2.types import CrawlRequest, ScrapeOptions +from firecrawl.v2.methods.crawl import _prepare_crawl_request + + +class TestCrawlRequestPreparation: + """Unit tests for crawl request preparation.""" + + def test_basic_request_preparation(self): + """Test basic request preparation with minimal fields.""" + request = CrawlRequest(url="https://example.com") + data = _prepare_crawl_request(request) + + # Check basic fields + assert data["url"] == "https://example.com" + + # Check that no options are present + assert "limit" not in data + assert "prompt" not in data + + def test_crawl_options_conversion(self): + """Test that CrawlOptions fields are converted to camelCase.""" + request = CrawlRequest( + url="https://example.com", + limit=10, + max_discovery_depth=3, + ignore_sitemap=True, + crawl_entire_domain=False, + allow_external_links=True + ) + + data = _prepare_crawl_request(request) + + # Check basic field + assert data["url"] == "https://example.com" + + # Check snake_case to camelCase conversions + assert "limit" in data + assert data["limit"] == 10 + assert "maxDiscoveryDepth" in data + assert data["maxDiscoveryDepth"] == 3 + assert "ignoreSitemap" in data + assert data["ignoreSitemap"] is True + assert "crawlEntireDomain" in data + assert data["crawlEntireDomain"] is False + assert "allowExternalLinks" in data + assert data["allowExternalLinks"] is True + + # Check that snake_case fields are not present + assert "ignore_sitemap" not in data + assert "crawl_entire_domain" not in data + assert "allow_external_links" not in data + + def test_scrape_options_conversion(self): + """Test that nested ScrapeOptions are converted to camelCase.""" + scrape_opts = ScrapeOptions( + formats=["markdown", "html"], + headers={"User-Agent": "Test"}, + include_tags=["h1", "h2"], + exclude_tags=["nav"], + only_main_content=False, + timeout=15000, + wait_for=2000, + mobile=True, + skip_tls_verification=True, + remove_base64_images=False + ) + + request = CrawlRequest( + url="https://example.com", + scrape_options=scrape_opts + ) + + data = _prepare_crawl_request(request) + + assert "scrapeOptions" in data + assert "scrape_options" not in data + + # Check nested conversions + scrape_data = data["scrapeOptions"] + assert "includeTags" in scrape_data + assert scrape_data["includeTags"] == ["h1", "h2"] + assert "excludeTags" in scrape_data + assert scrape_data["excludeTags"] == ["nav"] + assert "onlyMainContent" in scrape_data + assert scrape_data["onlyMainContent"] is False + assert "waitFor" in scrape_data + assert scrape_data["waitFor"] == 2000 + assert "skipTlsVerification" in scrape_data + assert scrape_data["skipTlsVerification"] is True + assert "removeBase64Images" in scrape_data + assert scrape_data["removeBase64Images"] is False + + def test_all_fields_conversion(self): + """Test request preparation with all possible fields.""" + scrape_opts = ScrapeOptions( + formats=["markdown"], + headers={"User-Agent": "Test"}, + only_main_content=False, + mobile=True + ) + + request = CrawlRequest( + url="https://example.com", + prompt="Extract all blog posts and documentation", + include_paths=["/blog/*", "/docs/*"], + exclude_paths=["/admin/*"], + max_discovery_depth=3, + ignore_sitemap=False, + limit=100, + crawl_entire_domain=True, + allow_external_links=False, + scrape_options=scrape_opts + ) + + data = _prepare_crawl_request(request) + + # Check basic fields + assert data["url"] == "https://example.com" + assert data["prompt"] == "Extract all blog posts and documentation" + + # Check all CrawlOptions fields + assert "includePaths" in data + assert data["includePaths"] == ["/blog/*", "/docs/*"] + assert "excludePaths" in data + assert data["excludePaths"] == ["/admin/*"] + assert "maxDiscoveryDepth" in data + assert data["maxDiscoveryDepth"] == 3 + assert "ignoreSitemap" in data + assert data["ignoreSitemap"] is False + assert "limit" in data + assert data["limit"] == 100 + assert "crawlEntireDomain" in data + assert data["crawlEntireDomain"] is True + assert "allowExternalLinks" in data + assert data["allowExternalLinks"] is False + + # Check nested scrape options + assert "scrapeOptions" in data + scrape_data = data["scrapeOptions"] + assert "onlyMainContent" in scrape_data + assert scrape_data["onlyMainContent"] is False + assert "mobile" in scrape_data + assert scrape_data["mobile"] is True + + def test_none_values_handling(self): + """Test that None values are handled correctly.""" + request = CrawlRequest( + url="https://example.com", + prompt=None, + limit=None, + scrape_options=None + ) + + data = _prepare_crawl_request(request) + + # Only the required field should be present + assert "url" in data + assert len(data) == 1 # Only url should be present + + def test_prompt_parameter(self): + """Test that prompt parameter is included when provided.""" + request = CrawlRequest( + url="https://example.com", + prompt="Extract all blog posts" + ) + + data = _prepare_crawl_request(request) + + assert "url" in data + assert "prompt" in data + assert data["prompt"] == "Extract all blog posts" + + def test_empty_options(self): + """Test that empty options are handled correctly.""" + request = CrawlRequest( + url="https://example.com" + ) + + data = _prepare_crawl_request(request) + + # Should only have the required url field + assert "url" in data + assert len(data) == 1 # Only url should be present + + def test_validation_integration(self): + """Test that validation is called during preparation.""" + # This should raise an error due to validation + with pytest.raises(ValueError, match="URL cannot be empty"): + request = CrawlRequest(url="") + _prepare_crawl_request(request) + + # This should raise an error due to validation + with pytest.raises(ValueError, match="Limit must be positive"): + request = CrawlRequest( + url="https://example.com", + limit=0 + ) + _prepare_crawl_request(request) + + def test_scrape_options_shared_function_integration(self): + """Test that the shared prepare_scrape_options function is being used.""" + # Test with all snake_case fields to ensure conversion + scrape_opts = ScrapeOptions( + include_tags=["h1", "h2"], + exclude_tags=["nav"], + only_main_content=False, + wait_for=2000, + skip_tls_verification=True, + remove_base64_images=False + ) + + request = CrawlRequest( + url="https://example.com", + scrape_options=scrape_opts + ) + + data = _prepare_crawl_request(request) + + # Check that scrapeOptions is present and converted + assert "scrapeOptions" in data + scrape_data = data["scrapeOptions"] + + # Check all conversions are working + assert "includeTags" in scrape_data + assert "excludeTags" in scrape_data + assert "onlyMainContent" in scrape_data + assert "waitFor" in scrape_data + assert "skipTlsVerification" in scrape_data + assert "removeBase64Images" in scrape_data + + # Check that snake_case fields are not present + assert "include_tags" not in scrape_data + assert "exclude_tags" not in scrape_data + assert "only_main_content" not in scrape_data + assert "wait_for" not in scrape_data + assert "skip_tls_verification" not in scrape_data + assert "remove_base64_images" not in scrape_data + assert "raw_html" not in scrape_data + assert "screenshot_full_page" not in scrape_data \ No newline at end of file diff --git a/apps/python-sdk/firecrawl/__tests__/unit/v2/methods/crawl/test_crawl_validation.py b/apps/python-sdk/firecrawl/__tests__/unit/v2/methods/crawl/test_crawl_validation.py new file mode 100644 index 000000000..518289f13 --- /dev/null +++ b/apps/python-sdk/firecrawl/__tests__/unit/v2/methods/crawl/test_crawl_validation.py @@ -0,0 +1,107 @@ +import pytest +from firecrawl.v2.types import CrawlRequest, ScrapeOptions +from firecrawl.v2.methods.crawl import _validate_crawl_request + + +class TestCrawlRequestValidation: + """Unit tests for crawl request validation.""" + + def test_validate_empty_url(self): + """Test validation with empty URL.""" + with pytest.raises(ValueError, match="URL cannot be empty"): + request = CrawlRequest(url="") + _validate_crawl_request(request) + + def test_validate_whitespace_url(self): + """Test validation with whitespace-only URL.""" + with pytest.raises(ValueError, match="URL cannot be empty"): + request = CrawlRequest(url=" ") + _validate_crawl_request(request) + + def test_validate_valid_url(self): + """Test validation with valid URL.""" + request = CrawlRequest(url="https://example.com") + _validate_crawl_request(request) # Should not raise + + def test_validate_invalid_limit(self): + """Test validation with invalid limit.""" + with pytest.raises(ValueError, match="Limit must be positive"): + request = CrawlRequest( + url="https://example.com", + limit=0 + ) + _validate_crawl_request(request) + + def test_validate_negative_limit(self): + """Test validation with negative limit.""" + with pytest.raises(ValueError, match="Limit must be positive"): + request = CrawlRequest( + url="https://example.com", + limit=-5 + ) + _validate_crawl_request(request) + + def test_validate_valid_limit(self): + """Test validation with valid limit.""" + request = CrawlRequest( + url="https://example.com", + limit=10 + ) + _validate_crawl_request(request) # Should not raise + + def test_validate_with_prompt(self): + """Test validation with prompt.""" + request = CrawlRequest( + url="https://example.com", + prompt="Extract all blog posts" + ) + _validate_crawl_request(request) # Should not raise + + def test_validate_with_prompt_and_options(self): + """Test validation with prompt and options.""" + request = CrawlRequest( + url="https://example.com", + prompt="Extract all blog posts", + limit=10 + ) + _validate_crawl_request(request) # Should not raise + + def test_validate_none_options(self): + """Test validation with None options.""" + request = CrawlRequest(url="https://example.com") + _validate_crawl_request(request) # Should not raise + + def test_validate_complex_options(self): + """Test validation with complex options.""" + scrape_opts = ScrapeOptions( + formats=["markdown"], + only_main_content=False, + mobile=True + ) + + request = CrawlRequest( + url="https://example.com", + limit=50, + max_discovery_depth=3, + scrape_options=scrape_opts + ) + _validate_crawl_request(request) # Should not raise + + def test_validate_scrape_options_integration(self): + """Test that scrape_options validation is integrated.""" + # Test with valid scrape options + scrape_opts = ScrapeOptions(formats=["markdown"], timeout=30000) + request = CrawlRequest( + url="https://example.com", + scrape_options=scrape_opts + ) + _validate_crawl_request(request) # Should not raise + + # Test with invalid scrape options (should raise error) + invalid_scrape_opts = ScrapeOptions(timeout=-1000) + request = CrawlRequest( + url="https://example.com", + scrape_options=invalid_scrape_opts + ) + with pytest.raises(ValueError, match="Timeout must be positive"): + _validate_crawl_request(request) \ No newline at end of file diff --git a/apps/python-sdk/firecrawl/__tests__/unit/v2/methods/search/test_search_request_preparation.py b/apps/python-sdk/firecrawl/__tests__/unit/v2/methods/search/test_search_request_preparation.py new file mode 100644 index 000000000..7d7e37b7d --- /dev/null +++ b/apps/python-sdk/firecrawl/__tests__/unit/v2/methods/search/test_search_request_preparation.py @@ -0,0 +1,171 @@ +import pytest +from firecrawl.v2.types import SearchRequest, ScrapeOptions, Source +from firecrawl.v2.methods.search import _prepare_search_request + + +class TestSearchRequestPreparation: + """Unit tests for search request preparation.""" + + def test_basic_request_preparation(self): + """Test basic request preparation with minimal fields.""" + request = SearchRequest(query="test query") + data = _prepare_search_request(request) + + # Check basic fields + assert data["query"] == "test query" + assert data["limit"] == 5 + assert data["timeout"] == 60000 + + # Check that snake_case fields are not present + assert "ignore_invalid_urls" not in data + assert "scrape_options" not in data + + def test_all_fields_conversion(self): + """Test request preparation with all possible fields.""" + scrape_opts = ScrapeOptions( + formats=["markdown"], + headers={"User-Agent": "Test"}, + include_tags=["h1", "h2"], + exclude_tags=["nav"], + only_main_content=False, + timeout=15000, + wait_for=2000, + mobile=True, + skip_tls_verification=True, + remove_base64_images=False + ) + + request = SearchRequest( + query="test query", + sources=["web", "news"], + limit=10, + tbs="qdr:w", + location="US", + ignore_invalid_urls=False, + timeout=30000, + scrape_options=scrape_opts + ) + + data = _prepare_search_request(request) + + # Check all basic fields + assert data["query"] == "test query" + assert data["limit"] == 10 + assert data["tbs"] == "qdr:w" + assert data["location"] == "US" + assert data["timeout"] == 30000 + + # Check snake_case to camelCase conversions + assert "ignoreInvalidURLs" in data + assert data["ignoreInvalidURLs"] is False + assert "ignore_invalid_urls" not in data + + assert "scrapeOptions" in data + assert "scrape_options" not in data + + # Check sources + assert "sources" in data + assert len(data["sources"]) == 2 + assert data["sources"][0]["type"] == "web" + assert data["sources"][1]["type"] == "news" + + # Check nested scrape options conversions + scrape_data = data["scrapeOptions"] + assert "includeTags" in scrape_data + assert scrape_data["includeTags"] == ["h1", "h2"] + assert "excludeTags" in scrape_data + assert scrape_data["excludeTags"] == ["nav"] + assert "onlyMainContent" in scrape_data + assert scrape_data["onlyMainContent"] is False + assert "waitFor" in scrape_data + assert scrape_data["waitFor"] == 2000 + assert "skipTlsVerification" in scrape_data + assert scrape_data["skipTlsVerification"] is True + assert "removeBase64Images" in scrape_data + assert scrape_data["removeBase64Images"] is False + + def test_exclude_none_behavior(self): + """Test that exclude_none=True behavior is working.""" + request = SearchRequest( + query="test", + sources=None, + limit=None, + tbs=None, + location=None, + ignore_invalid_urls=None, + timeout=None, + scrape_options=None + ) + + data = _prepare_search_request(request) + + # Default values should be included + assert "query" in data + assert "limit" in data # limit has default value 5 + assert "timeout" in data # timeout has default value 60000 + assert len(data) == 3 # query, limit, and timeout should be present + + def test_empty_scrape_options(self): + """Test that empty scrape options are handled correctly.""" + scrape_opts = ScrapeOptions() # All defaults + + request = SearchRequest( + query="test", + scrape_options=scrape_opts + ) + + data = _prepare_search_request(request) + + assert "scrapeOptions" in data + scrape_data = data["scrapeOptions"] + + # Should have default values + assert "onlyMainContent" in scrape_data + assert scrape_data["onlyMainContent"] is True + assert "mobile" in scrape_data + assert scrape_data["mobile"] is False + + def test_scrape_options_shared_function_integration(self): + """Test that the shared prepare_scrape_options function is being used.""" + # Test with all snake_case fields to ensure conversion + scrape_opts = ScrapeOptions( + include_tags=["h1", "h2"], + exclude_tags=["nav"], + only_main_content=False, + wait_for=2000, + skip_tls_verification=True, + remove_base64_images=False, + raw_html=True, + screenshot_full_page=True + ) + + request = SearchRequest( + query="test", + scrape_options=scrape_opts + ) + + data = _prepare_search_request(request) + + # Check that scrapeOptions is present and converted + assert "scrapeOptions" in data + scrape_data = data["scrapeOptions"] + + # Check all conversions are working + assert "includeTags" in scrape_data + assert "excludeTags" in scrape_data + assert "onlyMainContent" in scrape_data + assert "waitFor" in scrape_data + assert "skipTlsVerification" in scrape_data + assert "removeBase64Images" in scrape_data + assert "rawHtml" in scrape_data + assert "screenshot@fullPage" in scrape_data + + # Check that snake_case fields are not present + assert "include_tags" not in scrape_data + assert "exclude_tags" not in scrape_data + assert "only_main_content" not in scrape_data + assert "wait_for" not in scrape_data + assert "skip_tls_verification" not in scrape_data + assert "remove_base64_images" not in scrape_data + assert "raw_html" not in scrape_data + assert "screenshot_full_page" not in scrape_data \ No newline at end of file diff --git a/apps/python-sdk/firecrawl/__tests__/unit/v2/methods/search/test_search_validation.py b/apps/python-sdk/firecrawl/__tests__/unit/v2/methods/search/test_search_validation.py new file mode 100644 index 000000000..d41c50409 --- /dev/null +++ b/apps/python-sdk/firecrawl/__tests__/unit/v2/methods/search/test_search_validation.py @@ -0,0 +1,206 @@ +import pytest +from firecrawl.v2.types import SearchRequest, Source, ScrapeOptions, ScrapeFormats +from firecrawl.v2.methods.search import _validate_search_request + + +class TestSearchValidation: + """Unit tests for search request validation.""" + + def test_validate_empty_query(self): + """Test validation of empty query.""" + request = SearchRequest(query="") + with pytest.raises(ValueError, match="Query cannot be empty"): + _validate_search_request(request) + + request = SearchRequest(query=" ") + with pytest.raises(ValueError, match="Query cannot be empty"): + _validate_search_request(request) + + def test_validate_invalid_limit(self): + """Test validation of invalid limits.""" + # Zero limit + request = SearchRequest(query="test", limit=0) + with pytest.raises(ValueError, match="Limit must be positive"): + _validate_search_request(request) + + # Negative limit + request = SearchRequest(query="test", limit=-1) + with pytest.raises(ValueError, match="Limit must be positive"): + _validate_search_request(request) + + # Too high limit + request = SearchRequest(query="test", limit=101) + with pytest.raises(ValueError, match="Limit cannot exceed 100"): + _validate_search_request(request) + + def test_validate_invalid_timeout(self): + """Test validation of invalid timeouts.""" + # Zero timeout + request = SearchRequest(query="test", timeout=0) + with pytest.raises(ValueError, match="Timeout must be positive"): + _validate_search_request(request) + + # Negative timeout + request = SearchRequest(query="test", timeout=-1000) + with pytest.raises(ValueError, match="Timeout must be positive"): + _validate_search_request(request) + + # Too high timeout + request = SearchRequest(query="test", timeout=300001) + with pytest.raises(ValueError, match="Timeout cannot exceed 300000ms"): + _validate_search_request(request) + + def test_validate_invalid_sources(self): + """Test validation of invalid sources.""" + # Invalid string source + request = SearchRequest(query="test", sources=["invalid_source"]) + with pytest.raises(ValueError, match="Invalid source type"): + _validate_search_request(request) + + # Invalid object source + request = SearchRequest(query="test", sources=[Source(type="invalid_source")]) + with pytest.raises(ValueError, match="Invalid source type"): + _validate_search_request(request) + + # Mixed valid/invalid sources + request = SearchRequest(query="test", sources=["web", "invalid_source"]) + with pytest.raises(ValueError, match="Invalid source type"): + _validate_search_request(request) + + def test_validate_invalid_location(self): + """Test validation of invalid location.""" + # Empty location + request = SearchRequest(query="test", location="") + with pytest.raises(ValueError, match="Location must be a non-empty string"): + _validate_search_request(request) + + # Whitespace location + request = SearchRequest(query="test", location=" ") + with pytest.raises(ValueError, match="Location must be a non-empty string"): + _validate_search_request(request) + + def test_validate_invalid_tbs(self): + """Test validation of invalid tbs values.""" + invalid_tbs_values = ["invalid", "qdr:x", "yesterday", "last_week"] + + for invalid_tbs in invalid_tbs_values: + request = SearchRequest(query="test", tbs=invalid_tbs) + with pytest.raises(ValueError, match="Invalid tbs value"): + _validate_search_request(request) + + def test_validate_valid_requests(self): + """Test that valid requests pass validation.""" + # Minimal valid request + request = SearchRequest(query="test") + validated = _validate_search_request(request) + assert validated == request + + # Request with all optional parameters + request = SearchRequest( + query="test query", + sources=["web", "news"], + limit=10, + tbs="qdr:w", + location="US", + ignore_invalid_urls=False, + timeout=30000 + ) + validated = _validate_search_request(request) + assert validated == request + + # Request with object sources + request = SearchRequest( + query="test", + sources=[Source(type="web"), Source(type="images")] + ) + validated = _validate_search_request(request) + assert validated == request + + def test_validate_edge_cases(self): + """Test edge cases and boundary values.""" + # Maximum valid limit + request = SearchRequest(query="test", limit=100) + validated = _validate_search_request(request) + assert validated == request + + # Maximum valid timeout + request = SearchRequest(query="test", timeout=300000) + validated = _validate_search_request(request) + assert validated == request + + # Minimum valid limit + request = SearchRequest(query="test", limit=1) + validated = _validate_search_request(request) + assert validated == request + + # Minimum valid timeout + request = SearchRequest(query="test", timeout=1) + validated = _validate_search_request(request) + assert validated == request + + def test_validate_none_values(self): + """Test that None values for optional fields are handled correctly.""" + request = SearchRequest( + query="test", + sources=None, + limit=None, + tbs=None, + location=None, + ignore_invalid_urls=None, + timeout=None + ) + validated = _validate_search_request(request) + assert validated == request + + def test_validate_scrape_options_integration(self): + """Test that scrape_options validation is integrated.""" + # Test with valid scrape options + scrape_opts = ScrapeOptions(formats=["markdown"], timeout=30000) + request = SearchRequest(query="test", scrape_options=scrape_opts) + validated = _validate_search_request(request) + assert validated == request + + # Test with invalid scrape options (should raise error) + invalid_scrape_opts = ScrapeOptions(timeout=-1000) + request = SearchRequest(query="test", scrape_options=invalid_scrape_opts) + with pytest.raises(ValueError, match="Timeout must be positive"): + _validate_search_request(request) + + + + + +class TestSearchRequestModel: + """Unit tests for SearchRequest model behavior.""" + + def test_default_values(self): + """Test that default values are set correctly.""" + request = SearchRequest(query="test") + assert request.limit == 5 + assert request.ignore_invalid_urls is None # No default in model + assert request.timeout == 60000 + assert request.sources is None + assert request.tbs is None + assert request.location is None + assert request.scrape_options is None + + def test_field_aliases(self): + """Test that field aliases work correctly for API serialization.""" + # Test with None value (no default) + request1 = SearchRequest(query="test") + data1 = request1.model_dump(by_alias=True) + assert "ignore_invalid_urls" in data1 # No alias, uses snake_case + assert data1["ignore_invalid_urls"] is None + + # Test with explicit False value + request2 = SearchRequest( + query="test", + ignore_invalid_urls=False, + scrape_options=ScrapeOptions(formats=["markdown"]) + ) + + # Check that aliases are used in model_dump with by_alias=True + data2 = request2.model_dump(by_alias=True) + assert "ignore_invalid_urls" in data2 # No alias, uses snake_case + assert "scrape_options" in data2 # No alias, uses snake_case + assert data2["ignore_invalid_urls"] is False \ No newline at end of file diff --git a/apps/python-sdk/firecrawl/__tests__/unit/v2/utils/test_validation.py b/apps/python-sdk/firecrawl/__tests__/unit/v2/utils/test_validation.py new file mode 100644 index 000000000..6048b7153 --- /dev/null +++ b/apps/python-sdk/firecrawl/__tests__/unit/v2/utils/test_validation.py @@ -0,0 +1,232 @@ +import pytest +from firecrawl.v2.types import ScrapeOptions +from firecrawl.v2.utils.validation import validate_scrape_options, prepare_scrape_options + + +class TestValidateScrapeOptions: + """Unit tests for validate_scrape_options function.""" + + def test_validate_none_options(self): + """Test validation with None options.""" + result = validate_scrape_options(None) + assert result is None + + def test_validate_valid_options(self): + """Test validation with valid options.""" + options = ScrapeOptions( + formats=["markdown"], + timeout=30000, + wait_for=2000 + ) + result = validate_scrape_options(options) + assert result == options + + def test_validate_invalid_timeout(self): + """Test validation with invalid timeout.""" + options = ScrapeOptions(timeout=0) + with pytest.raises(ValueError, match="Timeout must be positive"): + validate_scrape_options(options) + + def test_validate_negative_timeout(self): + """Test validation with negative timeout.""" + options = ScrapeOptions(timeout=-1000) + with pytest.raises(ValueError, match="Timeout must be positive"): + validate_scrape_options(options) + + def test_validate_invalid_wait_for(self): + """Test validation with invalid wait_for.""" + options = ScrapeOptions(wait_for=-500) + with pytest.raises(ValueError, match="wait_for must be non-negative"): + validate_scrape_options(options) + + def test_validate_zero_wait_for(self): + """Test validation with zero wait_for (should be valid).""" + options = ScrapeOptions(wait_for=0) + result = validate_scrape_options(options) + assert result == options + + def test_validate_complex_options(self): + """Test validation with complex options.""" + options = ScrapeOptions( + formats=["markdown", "html"], + headers={"User-Agent": "Test"}, + include_tags=["h1", "h2"], + exclude_tags=["nav"], + only_main_content=False, + timeout=15000, + wait_for=2000, + mobile=True, + skip_tls_verification=True, + remove_base64_images=False, + raw_html=True, + screenshot_full_page=True + ) + result = validate_scrape_options(options) + assert result == options + + def test_validate_multiple_invalid_fields(self): + """Test validation with multiple invalid fields.""" + options = ScrapeOptions(timeout=-1000, wait_for=-500) + with pytest.raises(ValueError, match="Timeout must be positive"): + validate_scrape_options(options) + # Should fail on first invalid field (timeout) + + def test_validate_edge_cases(self): + """Test validation with edge case values.""" + # Test with very large timeout + options = ScrapeOptions(timeout=999999) + result = validate_scrape_options(options) + assert result == options + + # Test with very large wait_for + options = ScrapeOptions(wait_for=999999) + result = validate_scrape_options(options) + assert result == options + + +class TestPrepareScrapeOptions: + """Unit tests for prepare_scrape_options function.""" + + def test_prepare_none_options(self): + """Test preparation with None options.""" + result = prepare_scrape_options(None) + assert result is None + + def test_prepare_basic_options(self): + """Test preparation with basic options.""" + options = ScrapeOptions( + formats=["markdown"], + timeout=30000, + wait_for=2000 + ) + result = prepare_scrape_options(options) + + assert isinstance(result, dict) + assert "formats" in result + assert "timeout" in result + assert "waitFor" in result + assert result["timeout"] == 30000 + assert result["waitFor"] == 2000 + + def test_prepare_snake_case_conversion(self): + """Test snake_case to camelCase conversion.""" + options = ScrapeOptions( + include_tags=["h1", "h2"], + exclude_tags=["nav"], + only_main_content=False, + wait_for=2000, + skip_tls_verification=True, + remove_base64_images=False, + raw_html=True, + screenshot_full_page=True + ) + result = prepare_scrape_options(options) + + # Check conversions + assert "includeTags" in result + assert result["includeTags"] == ["h1", "h2"] + assert "excludeTags" in result + assert result["excludeTags"] == ["nav"] + assert "onlyMainContent" in result + assert result["onlyMainContent"] is False + assert "waitFor" in result + assert result["waitFor"] == 2000 + assert "skipTlsVerification" in result + assert result["skipTlsVerification"] is True + assert "removeBase64Images" in result + assert result["removeBase64Images"] is False + assert "rawHtml" in result + assert result["rawHtml"] is True + assert "screenshot@fullPage" in result + assert result["screenshot@fullPage"] is True + + # Check that snake_case fields are not present + assert "include_tags" not in result + assert "exclude_tags" not in result + assert "only_main_content" not in result + assert "wait_for" not in result + assert "skip_tls_verification" not in result + assert "remove_base64_images" not in result + assert "raw_html" not in result + assert "screenshot_full_page" not in result + + def test_prepare_complex_options(self): + """Test preparation with complex options.""" + options = ScrapeOptions( + formats=["markdown", "html"], + headers={"User-Agent": "Test Bot"}, + include_tags=["h1", "h2", "h3"], + exclude_tags=["nav", "footer"], + only_main_content=False, + timeout=15000, + wait_for=2000, + mobile=True, + skip_tls_verification=True, + remove_base64_images=False, + raw_html=True, + screenshot_full_page=True + ) + result = prepare_scrape_options(options) + + # Check all fields are present and converted + assert "formats" in result + assert "headers" in result + assert "includeTags" in result + assert "excludeTags" in result + assert "onlyMainContent" in result + assert "timeout" in result + assert "waitFor" in result + assert "mobile" in result + assert "skipTlsVerification" in result + assert "removeBase64Images" in result + assert "rawHtml" in result + assert "screenshot@fullPage" in result + + # Check values + assert result["formats"] == ["markdown", "html"] + assert result["headers"] == {"User-Agent": "Test Bot"} + assert result["includeTags"] == ["h1", "h2", "h3"] + assert result["excludeTags"] == ["nav", "footer"] + assert result["onlyMainContent"] is False + assert result["timeout"] == 15000 + assert result["waitFor"] == 2000 + assert result["mobile"] is True + assert result["skipTlsVerification"] is True + assert result["removeBase64Images"] is False + assert result["rawHtml"] is True + assert result["screenshot@fullPage"] is True + + def test_prepare_invalid_options(self): + """Test preparation with invalid options (should raise error).""" + options = ScrapeOptions(timeout=-1000) + with pytest.raises(ValueError, match="Timeout must be positive"): + prepare_scrape_options(options) + + def test_prepare_empty_options(self): + """Test preparation with empty options.""" + options = ScrapeOptions() # All defaults + result = prepare_scrape_options(options) + + # Should return dict with default values + assert isinstance(result, dict) + assert "onlyMainContent" in result + assert result["onlyMainContent"] is True + assert "mobile" in result + assert result["mobile"] is False + + def test_prepare_none_values(self): + """Test preparation with None values in options.""" + options = ScrapeOptions( + formats=None, + timeout=None, + wait_for=None, + include_tags=None, + exclude_tags=None + ) + result = prepare_scrape_options(options) + + # Should only include non-None values + assert isinstance(result, dict) + # Should have default values for required fields + assert "onlyMainContent" in result + assert "mobile" in result \ No newline at end of file diff --git a/apps/python-sdk/firecrawl/client.py b/apps/python-sdk/firecrawl/client.py index 9e415fc15..20f78545f 100644 --- a/apps/python-sdk/firecrawl/client.py +++ b/apps/python-sdk/firecrawl/client.py @@ -53,14 +53,16 @@ class V2Proxy: def __init__(self, client_instance: Optional[V2FirecrawlClient]): self._client = client_instance - # TODO: Not implemented yet if client_instance: - self.scrape = client_instance.scrape + # self.scrape = client_instance.scrape self.search = client_instance.search self.crawl = client_instance.crawl - self.batch_scrape = client_instance.batch_scrape self.get_crawl_status = client_instance.get_crawl_status self.cancel_crawl = client_instance.cancel_crawl + self.start_crawl = client_instance.start_crawl + self.crawl_params = client_instance.crawl_params + # self.batch_scrape = client_instance.batch_scrape + # self.map = client_instance.map def __getattr__(self, name): """Forward attribute access to the underlying client.""" @@ -128,13 +130,15 @@ class Firecrawl: self.v1 = V1Proxy(self._v1_client) if self._v1_client else None self.v2 = V2Proxy(self._v2_client) - # Methods - # TODO: Not implemented yet + # self.scrape = self._v2_client.scrape - # self.crawl = self._v2_client.crawl + self.crawl = self._v2_client.crawl + self.start_crawl = self._v2_client.start_crawl + self.crawl_params = self._v2_client.crawl_params + self.get_crawl_status = self._v2_client.get_crawl_status + self.cancel_crawl = self._v2_client.cancel_crawl # self.batch_scrape = self._v2_client.batch_scrape - # self.get_crawl_status = self._v2_client.get_crawl_status - # self.cancel_crawl = self._v2_client.cancel_crawl + # self.map = self._v2_client.map self.search = self._v2_client.search class AsyncFirecrawl: diff --git a/apps/python-sdk/firecrawl/types.py b/apps/python-sdk/firecrawl/types.py index 66538f84d..f82d6dea4 100644 --- a/apps/python-sdk/firecrawl/types.py +++ b/apps/python-sdk/firecrawl/types.py @@ -17,22 +17,24 @@ from .v2.types import ( ScrapeFormats, ScrapeOptions, ScrapeRequest, + ScrapeData, ScrapeResponse, # Crawl types - CrawlOptions, CrawlRequest, CrawlJob, + CrawlJobData, + CrawlData, CrawlResponse, - CrawlStatusData, - CrawlStatusResponse, + CrawlParamsRequest, + CrawlParamsData, + CrawlParamsResponse, # Batch scrape types BatchScrapeRequest, BatchScrapeJob, + BatchScrapeData, BatchScrapeResponse, - BatchScrapeStatusData, - BatchScrapeStatusResponse, # Map types MapOptions, @@ -47,6 +49,7 @@ from .v2.types import ( FormatOption, SearchRequest, SearchResult, + SearchData, SearchResponse, # Action types @@ -98,22 +101,24 @@ __all__ = [ 'ScrapeFormats', 'ScrapeOptions', 'ScrapeRequest', + 'ScrapeData', 'ScrapeResponse', # Crawl types - 'CrawlOptions', 'CrawlRequest', 'CrawlJob', + 'CrawlJobData', + 'CrawlData', 'CrawlResponse', - 'CrawlStatusData', - 'CrawlStatusResponse', + 'CrawlParamsRequest', + 'CrawlParamsData', + 'CrawlParamsResponse', # Batch scrape types 'BatchScrapeRequest', 'BatchScrapeJob', + 'BatchScrapeData', 'BatchScrapeResponse', - 'BatchScrapeStatusData', - 'BatchScrapeStatusResponse', # Map types 'MapOptions', @@ -128,6 +133,7 @@ __all__ = [ 'FormatOption', 'SearchRequest', 'SearchResult', + 'SearchData', 'SearchResponse', # Action types diff --git a/apps/python-sdk/firecrawl/v2/__init__.py b/apps/python-sdk/firecrawl/v2/__init__.py index 5bedce8a8..98769c735 100644 --- a/apps/python-sdk/firecrawl/v2/__init__.py +++ b/apps/python-sdk/firecrawl/v2/__init__.py @@ -1,11 +1,22 @@ -from typing import Optional, List, Union -from ..types import SearchResponse, ScrapeOptions, SearchRequest, SourceOption, FormatOption +from typing import Optional, List, Union, Dict, Callable, Literal, Any +from ..types import ( + SearchData, SearchResult, Document, ScrapeOptions, SearchRequest, + SourceOption, FormatOption, CrawlRequest, CrawlJobData, CrawlData, + CrawlParamsRequest, CrawlParamsData +) from .methods.search import search as search_method +from .methods.crawl import ( + crawl as crawl_method, + start_crawl as start_crawl_method, + cancel_crawl as cancel_crawl_method, + get_crawl_status as get_crawl_status_method, + crawl_params as crawl_params_method +) from .utils.http_client import HttpClient class FirecrawlClient: """ - + Firecrawl v2 API client. """ def __init__(self, api_key: str = None, api_url: str = "https://api.firecrawl.dev"): @@ -25,8 +36,8 @@ class FirecrawlClient: ): pass - # async-batch-scrape - def async_batch_scrape( + # start-batch-scrape + def start_batch_scrape( self ): pass @@ -50,40 +61,121 @@ class FirecrawlClient: # crawl def crawl( - self + self, + url: str, + prompt: Optional[str] = None, + include_paths: Optional[List[str]] = None, + exclude_paths: Optional[List[str]] = None, + max_discovery_depth: Optional[int] = None, + ignore_sitemap: bool = False, + ignore_query_parameters: bool = False, + limit: Optional[int] = None, + crawl_entire_domain: bool = False, + allow_external_links: bool = False, + allow_subdomains: bool = False, + delay: Optional[int] = None, + max_concurrency: Optional[int] = None, + webhook: Optional[Dict[str, Any]] = None, + scrape_options: Optional[ScrapeOptions] = None, + zero_data_retention: bool = False, + poll_interval: int = 2, + timeout: Optional[int] = None, + progress_callback: Optional[Callable[[CrawlJobData], None]] = None ): - pass + """Start a crawl job and wait for it to complete.""" + request = CrawlRequest( + url=url, + prompt=prompt, + include_paths=include_paths, + exclude_paths=exclude_paths, + max_discovery_depth=max_discovery_depth, + ignore_sitemap=ignore_sitemap, + ignore_query_parameters=ignore_query_parameters, + limit=limit, + crawl_entire_domain=crawl_entire_domain, + allow_external_links=allow_external_links, + allow_subdomains=allow_subdomains, + delay=delay, + max_concurrency=max_concurrency, + webhook=webhook, + scrape_options=scrape_options, + zero_data_retention=zero_data_retention + ) + return crawl_method(self._client, request, poll_interval, timeout, progress_callback) - # async-crawl - def async_crawl( - self - ): - pass - - - # crawl-params - def crawl_params( - self - ): - pass + # start-crawl + def start_crawl( + self, + url: str, + prompt: Optional[str] = None, + include_paths: Optional[List[str]] = None, + exclude_paths: Optional[List[str]] = None, + max_discovery_depth: Optional[int] = None, + ignore_sitemap: bool = False, + ignore_query_parameters: bool = False, + limit: Optional[int] = None, + crawl_entire_domain: bool = False, + allow_external_links: bool = False, + allow_subdomains: bool = False, + delay: Optional[int] = None, + max_concurrency: Optional[int] = None, + webhook: Optional[Dict[str, Any]] = None, + scrape_options: Optional[ScrapeOptions] = None, + zero_data_retention: bool = False + ) -> CrawlJobData: + """Start a crawl job and return immediately.""" + request = CrawlRequest( + url=url, + prompt=prompt, + include_paths=include_paths, + exclude_paths=exclude_paths, + max_discovery_depth=max_discovery_depth, + ignore_sitemap=ignore_sitemap, + ignore_query_parameters=ignore_query_parameters, + limit=limit, + crawl_entire_domain=crawl_entire_domain, + allow_external_links=allow_external_links, + allow_subdomains=allow_subdomains, + delay=delay, + max_concurrency=max_concurrency, + webhook=webhook, + scrape_options=scrape_options, + zero_data_retention=zero_data_retention + ) + return start_crawl_method(self._client, request) # get-crawl-status def get_crawl_status( - self + self, + job_id: str ): - pass + """Get the status of a crawl job.""" + return get_crawl_status_method(self._client, job_id) # cancel-crawl def cancel_crawl( - self + self, + job_id: str ): - pass + """Cancel a running crawl job.""" + return cancel_crawl_method(self._client, job_id) + + # crawl-params + def crawl_params( + self, + url: str, + prompt: str + ) -> CrawlParamsData: + """Get crawl parameters from LLM based on URL and prompt.""" + request = CrawlParamsRequest(url=url, prompt=prompt) + return crawl_params_method(self._client, request) # get-active-crawls def get_active_crawls( self ): pass + # get-crawl-errors def get_crawl_errors( self @@ -106,7 +198,7 @@ class FirecrawlClient: ignore_invalid_urls: Optional[bool] = True, timeout: Optional[int] = 60000, scrape_options: Optional[ScrapeOptions] = None, - ) -> SearchResponse: + ) -> SearchData: """Search for documents.""" request = SearchRequest( query=query, diff --git a/apps/python-sdk/firecrawl/v2/methods/batch.py b/apps/python-sdk/firecrawl/v2/methods/batch.py index f74524aae..d0d6bdc4a 100644 --- a/apps/python-sdk/firecrawl/v2/methods/batch.py +++ b/apps/python-sdk/firecrawl/v2/methods/batch.py @@ -5,11 +5,10 @@ Batch scraping functionality for Firecrawl v2 API. import time from typing import Optional, List, Callable from .types import ( - BatchScrapeRequest, BatchScrapeResponse, BatchScrapeStatusResponse, - BatchScrapeJob, BatchScrapeStatusData, ScrapeOptions, Document + BatchScrapeRequest, BatchScrapeResponse, + BatchScrapeJob, BatchScrapeData, ScrapeOptions, Document ) -from .utils.http_client import HttpClient -from .utils.error_handler import handle_response_error +from .utils import HttpClient, handle_response_error, validate_scrape_options, prepare_scrape_options def start_batch_scrape( @@ -32,10 +31,7 @@ def start_batch_scrape( FirecrawlError: If the batch scrape operation fails to start """ # Prepare request data - request_data = {"urls": urls} - - if options: - request_data["pageOptions"] = options.dict(exclude_none=True, by_alias=True) + request_data = prepare_batch_request(urls, options) # Make the API request response = client.post("/v1/batch/scrape", request_data) @@ -66,7 +62,7 @@ def start_batch_scrape( def get_batch_scrape_status( client: HttpClient, job_id: str -) -> BatchScrapeStatusResponse: +) -> BatchScrapeResponse: """ Get the status of a batch scrape job. @@ -99,20 +95,20 @@ def get_batch_scrape_status( for doc_data in status_data["data"]: documents.append(Document(**doc_data)) - batch_status = BatchScrapeStatusData( + batch_data = BatchScrapeData( status=status_data.get("status"), current=status_data.get("current", 0), total=status_data.get("total", 0), data=documents ) - return BatchScrapeStatusResponse( + return BatchScrapeResponse( success=True, - data=batch_status, + data=batch_data, warning=response_data.get("warning") ) else: - return BatchScrapeStatusResponse( + return BatchScrapeResponse( success=False, error=response_data.get("error", "Unknown error occurred") ) @@ -121,7 +117,7 @@ def get_batch_scrape_status( def cancel_batch_scrape( client: HttpClient, job_id: str -) -> BatchScrapeStatusResponse: +) -> BatchScrapeResponse: """ Cancel a running batch scrape job. @@ -148,20 +144,20 @@ def cancel_batch_scrape( if response_data.get("success"): status_data = response_data.get("data", {}) - batch_status = BatchScrapeStatusData( + batch_data = BatchScrapeData( status=status_data.get("status", "cancelled"), current=status_data.get("current", 0), total=status_data.get("total", 0), data=[] ) - return BatchScrapeStatusResponse( + return BatchScrapeResponse( success=True, - data=batch_status, + data=batch_data, warning=response_data.get("warning") ) else: - return BatchScrapeStatusResponse( + return BatchScrapeResponse( success=False, error=response_data.get("error", "Unknown error occurred") ) @@ -172,8 +168,8 @@ def wait_for_batch_completion( job_id: str, poll_interval: int = 2, timeout: Optional[int] = None, - progress_callback: Optional[Callable[[BatchScrapeStatusData], None]] = None -) -> BatchScrapeStatusResponse: + progress_callback: Optional[Callable[[BatchScrapeData], None]] = None +) -> BatchScrapeResponse: """ Wait for a batch scrape job to complete, polling for status updates. @@ -223,8 +219,8 @@ def batch_scrape_and_wait( options: Optional[ScrapeOptions] = None, poll_interval: int = 2, timeout: Optional[int] = None, - progress_callback: Optional[Callable[[BatchScrapeStatusData], None]] = None -) -> BatchScrapeStatusResponse: + progress_callback: Optional[Callable[[BatchScrapeData], None]] = None +) -> BatchScrapeResponse: """ Start a batch scrape job and wait for it to complete. @@ -308,7 +304,10 @@ def prepare_batch_request(urls: List[str], options: Optional[ScrapeOptions] = No request_data = {"urls": validated_urls} if options: - request_data["pageOptions"] = options.dict(exclude_none=True, by_alias=True) + # Use shared function for ScrapeOptions preparation + scrape_data = prepare_scrape_options(options) + if scrape_data: + request_data["pageOptions"] = scrape_data return request_data diff --git a/apps/python-sdk/firecrawl/v2/methods/crawl.py b/apps/python-sdk/firecrawl/v2/methods/crawl.py index 922e28843..8312f9824 100644 --- a/apps/python-sdk/firecrawl/v2/methods/crawl.py +++ b/apps/python-sdk/firecrawl/v2/methods/crawl.py @@ -3,42 +3,116 @@ Crawling functionality for Firecrawl v2 API. """ import time -from typing import Optional, Callable, Generator -from .types import ( - CrawlRequest, CrawlResponse, CrawlStatusResponse, CrawlOptions, - CrawlJob, CrawlStatusData, Document +from typing import Optional, Callable +from ..types import ( + CrawlRequest, CrawlResponse, CrawlStartResponse, + CrawlJob, CrawlJobData, CrawlData, Document, CrawlParamsRequest, CrawlParamsResponse, CrawlParamsData ) -from .utils.http_client import HttpClient -from .utils.error_handler import handle_response_error +from ..utils import HttpClient, handle_response_error, validate_scrape_options, prepare_scrape_options -def start_crawl( - client: HttpClient, - url: str, - options: Optional[CrawlOptions] = None -) -> CrawlResponse: +def _validate_crawl_request(request: CrawlRequest) -> None: + """ + Validate crawl request parameters. + + Args: + request: CrawlRequest to validate + + Raises: + ValueError: If request is invalid + """ + if not request.url or not request.url.strip(): + raise ValueError("URL cannot be empty") + + if request.limit is not None and request.limit <= 0: + raise ValueError("Limit must be positive") + + # Validate scrape_options (if provided) + if request.scrape_options is not None: + validate_scrape_options(request.scrape_options) + + +def _prepare_crawl_request(request: CrawlRequest) -> dict: + """ + Prepare crawl request for API submission. + + Args: + request: CrawlRequest to prepare + + Returns: + Dictionary ready for API submission + """ + # Validate request + _validate_crawl_request(request) + + # Start with basic data + data = {"url": request.url} + + # Add prompt if present + if request.prompt: + data["prompt"] = request.prompt + + # Handle scrape_options conversion first (before model_dump) + if request.scrape_options is not None: + scrape_data = prepare_scrape_options(request.scrape_options) + if scrape_data: + data["scrapeOptions"] = scrape_data + + # Convert request to dict + request_data = request.model_dump(exclude_none=True, exclude_unset=True) + + # Remove url, prompt, and scrape_options (already handled) + request_data.pop("url", None) + request_data.pop("prompt", None) + request_data.pop("scrape_options", None) + + # Convert other snake_case fields to camelCase + field_mappings = { + "include_paths": "includePaths", + "exclude_paths": "excludePaths", + "max_discovery_depth": "maxDiscoveryDepth", + "ignore_sitemap": "ignoreSitemap", + "ignore_query_parameters": "ignoreQueryParameters", + "crawl_entire_domain": "crawlEntireDomain", + "allow_external_links": "allowExternalLinks", + "allow_subdomains": "allowSubdomains", + "delay": "delay", + "max_concurrency": "maxConcurrency", + "webhook": "webhook", + "zero_data_retention": "zeroDataRetention" + } + + # Apply field mappings + for snake_case, camel_case in field_mappings.items(): + if snake_case in request_data: + data[camel_case] = request_data.pop(snake_case) + + # Add any remaining fields that don't need conversion (like limit) + data.update(request_data) + + return data + + +def start_crawl(client: HttpClient, request: CrawlRequest) -> CrawlJob: """ Start a crawl job for a website. Args: client: HTTP client instance - url: URL to crawl - options: Crawling options + request: CrawlRequest containing URL and options Returns: - CrawlResponse containing job information + CrawlJob with job information Raises: - FirecrawlError: If the crawl operation fails to start + ValueError: If request is invalid + Exception: If the crawl operation fails to start """ # Prepare request data - request_data = {"url": url} - - if options: - request_data.update(options.dict(exclude_none=True, by_alias=True)) + request_data = _prepare_crawl_request(request) # Make the API request - response = client.post("/v1/crawl", request_data) + response = client.post("/v2/crawl", request_data) # Handle errors if not response.ok: @@ -48,25 +122,18 @@ def start_crawl( response_data = response.json() if response_data.get("success"): - job_data = response_data.get("data", {}) - job = CrawlJob(**job_data) - - return CrawlResponse( - success=True, - data=job, - warning=response_data.get("warning") - ) + # The API returns id and url at the top level, not in a data field + job_data = { + "id": response_data.get("id"), + "url": request.url, # Use the original request URL + "status": "scraping" # Default status for new jobs + } + return CrawlJob(**job_data) else: - return CrawlResponse( - success=False, - error=response_data.get("error", "Unknown error occurred") - ) + raise Exception(response_data.get("error", "Unknown error occurred")) -def get_crawl_status( - client: HttpClient, - job_id: str -) -> CrawlStatusResponse: +def get_crawl_status(client: HttpClient, job_id: str) -> CrawlJob: """ Get the status of a crawl job. @@ -75,13 +142,13 @@ def get_crawl_status( job_id: ID of the crawl job Returns: - CrawlStatusResponse containing job status and data + CrawlJob with current status and data Raises: - FirecrawlError: If the status check fails + Exception: If the status check fails """ # Make the API request - response = client.get(f"/v1/crawl/{job_id}") + response = client.get(f"/v2/crawl/{job_id}") # Handle errors if not response.ok: @@ -91,44 +158,47 @@ def get_crawl_status( response_data = response.json() if response_data.get("success"): - status_data = response_data.get("data", {}) + # The API returns status fields at the top level, not in a data field # Convert documents documents = [] - if "data" in status_data: - for doc_data in status_data["data"]: + data_list = response_data.get("data", []) + for doc_data in data_list: + if isinstance(doc_data, str): + # Handle case where API returns just URLs - this shouldn't happen for crawl + # but we'll handle it gracefully + continue + else: documents.append(Document(**doc_data)) # Convert partial data if present partial_documents = [] - if "partialData" in status_data: - for doc_data in status_data["partialData"]: + partial_data_list = response_data.get("partialData", []) + for doc_data in partial_data_list: + if isinstance(doc_data, str): + # Handle case where API returns just URLs - this shouldn't happen for crawl + # but we'll handle it gracefully + continue + else: partial_documents.append(Document(**doc_data)) - crawl_status = CrawlStatusData( - status=status_data.get("status"), - current=status_data.get("current", 0), - total=status_data.get("total", 0), + # Create CrawlJob with current status and data + return CrawlJob( + id=job_id, + url=response_data.get("url", ""), # URL might not be in status response + status=response_data.get("status"), + current=response_data.get("completed", 0), # API uses "completed" instead of "current" + total=response_data.get("total", 0), + created_at=response_data.get("createdAt"), + completed_at=response_data.get("completedAt"), data=documents, partial_data=partial_documents if partial_documents else None ) - - return CrawlStatusResponse( - success=True, - data=crawl_status, - warning=response_data.get("warning") - ) else: - return CrawlStatusResponse( - success=False, - error=response_data.get("error", "Unknown error occurred") - ) + raise Exception(response_data.get("error", "Unknown error occurred")) -def cancel_crawl( - client: HttpClient, - job_id: str -) -> CrawlStatusResponse: +def cancel_crawl(client: HttpClient, job_id: str) -> CrawlResponse: """ Cancel a running crawl job. @@ -137,13 +207,13 @@ def cancel_crawl( job_id: ID of the crawl job to cancel Returns: - CrawlStatusResponse with updated status + CrawlResponse with updated status Raises: - FirecrawlError: If the cancellation fails + Exception: If the cancellation fails """ # Make the API request - response = client.delete(f"/v1/crawl/{job_id}") + response = client.delete(f"/v2/crawl/{job_id}") # Handle errors if not response.ok: @@ -155,20 +225,21 @@ def cancel_crawl( if response_data.get("success"): status_data = response_data.get("data", {}) - crawl_status = CrawlStatusData( - status=status_data.get("status", "cancelled"), + crawl_job_data = CrawlJobData( + id=job_id, # Use the job_id parameter + status=status_data.get("status", "failed"), current=status_data.get("current", 0), total=status_data.get("total", 0), data=[] ) - return CrawlStatusResponse( + return CrawlResponse( success=True, - data=crawl_status, + data=crawl_job_data, warning=response_data.get("warning") ) else: - return CrawlStatusResponse( + return CrawlResponse( success=False, error=response_data.get("error", "Unknown error occurred") ) @@ -179,8 +250,8 @@ def wait_for_crawl_completion( job_id: str, poll_interval: int = 2, timeout: Optional[int] = None, - progress_callback: Optional[Callable[[CrawlStatusData], None]] = None -) -> CrawlStatusResponse: + progress_callback: Optional[Callable[[CrawlJob], None]] = None +) -> CrawlJob: """ Wait for a crawl job to complete, polling for status updates. @@ -192,29 +263,24 @@ def wait_for_crawl_completion( progress_callback: Optional callback for progress updates Returns: - CrawlStatusResponse when job completes + CrawlJob when job completes Raises: - FirecrawlError: If the job fails or timeout is reached + Exception: If the job fails TimeoutError: If timeout is reached """ start_time = time.time() while True: - status_response = get_crawl_status(client, job_id) - - if not status_response.success: - return status_response - - status_data = status_response.data + crawl_job = get_crawl_status(client, job_id) # Call progress callback if provided - if progress_callback and status_data: - progress_callback(status_data) + if progress_callback: + progress_callback(crawl_job) # Check if job is complete - if status_data and status_data.status in ["completed", "failed", "cancelled"]: - return status_response + if crawl_job.status in ["completed", "failed"]: + return crawl_job # Check timeout if timeout and (time.time() - start_time) > timeout: @@ -224,42 +290,34 @@ def wait_for_crawl_completion( time.sleep(poll_interval) -def crawl_and_wait( +def crawl( client: HttpClient, - url: str, - options: Optional[CrawlOptions] = None, + request: CrawlRequest, poll_interval: int = 2, timeout: Optional[int] = None, - progress_callback: Optional[Callable[[CrawlStatusData], None]] = None -) -> CrawlStatusResponse: + progress_callback: Optional[Callable[[CrawlJob], None]] = None +) -> CrawlJob: """ Start a crawl job and wait for it to complete. Args: client: HTTP client instance - url: URL to crawl - options: Crawling options + request: CrawlRequest containing URL and options poll_interval: Seconds between status checks timeout: Maximum seconds to wait (None for no timeout) progress_callback: Optional callback for progress updates Returns: - CrawlStatusResponse when job completes + CrawlJob when job completes Raises: - FirecrawlError: If the crawl fails to start or complete + ValueError: If request is invalid + Exception: If the crawl fails to start or complete TimeoutError: If timeout is reached """ # Start the crawl - crawl_response = start_crawl(client, url, options) - - if not crawl_response.success or not crawl_response.data: - return CrawlStatusResponse( - success=False, - error=crawl_response.error or "Failed to start crawl" - ) - - job_id = crawl_response.data.id + crawl_job = start_crawl(client, request) + job_id = crawl_job.id # Wait for completion return wait_for_crawl_completion( @@ -267,94 +325,106 @@ def crawl_and_wait( ) -def stream_crawl_results( - client: HttpClient, - job_id: str, - poll_interval: int = 2 -) -> Generator[Document, None, None]: +def crawl_params(client: HttpClient, request: CrawlParamsRequest) -> CrawlParamsData: """ - Stream crawl results as they become available. + Get crawl parameters from LLM based on URL and prompt. Args: client: HTTP client instance - job_id: ID of the crawl job - poll_interval: Seconds between status checks - - Yields: - Document objects as they are crawled - - Raises: - FirecrawlError: If the job fails - """ - seen_count = 0 - - while True: - status_response = get_crawl_status(client, job_id) - - if not status_response.success: - raise Exception(f"Failed to get crawl status: {status_response.error}") - - status_data = status_response.data - if not status_data: - break - - # Yield new documents - if status_data.data: - for i in range(seen_count, len(status_data.data)): - yield status_data.data[i] - seen_count = len(status_data.data) - - # Check if job is complete - if status_data.status in ["completed", "failed", "cancelled"]: - break - - # Wait before next poll - time.sleep(poll_interval) - - -def validate_crawl_options(options: Optional[CrawlOptions]) -> Optional[CrawlOptions]: - """ - Validate and normalize crawl options. - - Args: - options: Crawling options to validate + request: CrawlParamsRequest containing URL and prompt Returns: - Validated options or None + CrawlParamsData containing suggested crawl options Raises: - ValueError: If options are invalid + ValueError: If request is invalid + Exception: If the operation fails """ - if options is None: - return None + # Validate request + if not request.url or not request.url.strip(): + raise ValueError("URL cannot be empty") - # Validate limit - if options.limit is not None and options.limit <= 0: - raise ValueError("Limit must be positive") + if not request.prompt or not request.prompt.strip(): + raise ValueError("Prompt cannot be empty") - # Validate max_depth - if options.max_depth is not None and options.max_depth < 0: - raise ValueError("max_depth must be non-negative") + # Prepare request data + request_data = { + "url": request.url, + "prompt": request.prompt + } - return options - - -def prepare_crawl_request(url: str, options: Optional[CrawlOptions] = None) -> dict: - """ - Prepare a crawl request payload. + # Make the API request + response = client.post("/v2/crawl-params", request_data) - Args: - url: URL to crawl - options: Crawling options + # Handle errors + if not response.ok: + handle_response_error(response, "crawl params") + + # Parse response + response_data = response.json() + + if response_data.get("success"): + params_data = response_data.get("data", {}) - Returns: - Request payload dictionary - """ - request_data = {"url": url} - - if options: - validated_options = validate_crawl_options(options) - if validated_options: - request_data.update(validated_options.dict(exclude_none=True, by_alias=True)) - - return request_data \ No newline at end of file + # Convert camelCase to snake_case for CrawlParamsData + converted_params = {} + field_mappings = { + "includePaths": "include_paths", + "excludePaths": "exclude_paths", + "maxDiscoveryDepth": "max_discovery_depth", + "ignoreSitemap": "ignore_sitemap", + "ignoreQueryParameters": "ignore_query_parameters", + "crawlEntireDomain": "crawl_entire_domain", + "allowExternalLinks": "allow_external_links", + "allowSubdomains": "allow_subdomains", + "maxConcurrency": "max_concurrency", + "scrapeOptions": "scrape_options", + "zeroDataRetention": "zero_data_retention" + } + + for camel_case, snake_case in field_mappings.items(): + if camel_case in params_data: + if camel_case == "scrapeOptions" and params_data[camel_case] is not None: + # Handle nested scrapeOptions conversion + scrape_opts_data = params_data[camel_case] + converted_scrape_opts = {} + scrape_field_mappings = { + "includeTags": "include_tags", + "excludeTags": "exclude_tags", + "onlyMainContent": "only_main_content", + "waitFor": "wait_for", + "skipTlsVerification": "skip_tls_verification", + "removeBase64Images": "remove_base64_images" + } + + for scrape_camel, scrape_snake in scrape_field_mappings.items(): + if scrape_camel in scrape_opts_data: + converted_scrape_opts[scrape_snake] = scrape_opts_data[scrape_camel] + + # Handle formats field - if it's a list, convert to ScrapeFormats + if "formats" in scrape_opts_data: + formats_data = scrape_opts_data["formats"] + if isinstance(formats_data, list): + # Convert list to ScrapeFormats object + from ..types import ScrapeFormats + converted_scrape_opts["formats"] = ScrapeFormats(formats=formats_data) + else: + converted_scrape_opts["formats"] = formats_data + + # Add fields that don't need conversion + for key, value in scrape_opts_data.items(): + if key not in scrape_field_mappings and key != "formats": + converted_scrape_opts[key] = value + + converted_params[snake_case] = converted_scrape_opts + else: + converted_params[snake_case] = params_data[camel_case] + + # Add fields that don't need conversion + for key, value in params_data.items(): + if key not in field_mappings: + converted_params[key] = value + + return CrawlParamsData(**converted_params) + else: + raise Exception(response_data.get("error", "Unknown error occurred")) \ No newline at end of file diff --git a/apps/python-sdk/firecrawl/v2/methods/scrape.py b/apps/python-sdk/firecrawl/v2/methods/scrape.py index eb6e6eaef..eacaf8bf0 100644 --- a/apps/python-sdk/firecrawl/v2/methods/scrape.py +++ b/apps/python-sdk/firecrawl/v2/methods/scrape.py @@ -5,8 +5,7 @@ Scraping functionality for Firecrawl v2 API. import json from typing import Optional, Dict, Any from .types import ScrapeRequest, ScrapeResponse, ScrapeOptions, Document -from .utils.http_client import HttpClient -from .utils.error_handler import handle_response_error +from .utils import HttpClient, handle_response_error, validate_scrape_options def scrape( @@ -167,31 +166,7 @@ def scrape_multiple( return results -def validate_scrape_options(options: Optional[ScrapeOptions]) -> Optional[ScrapeOptions]: - """ - Validate and normalize scrape options. - - Args: - options: Scraping options to validate - - Returns: - Validated options or None - - Raises: - ValueError: If options are invalid - """ - if options is None: - return None - - # Validate timeout - if options.timeout is not None and options.timeout <= 0: - raise ValueError("Timeout must be positive") - - # Validate wait_for - if options.wait_for is not None and options.wait_for < 0: - raise ValueError("wait_for must be non-negative") - - return options + def prepare_scrape_request(url: str, options: Optional[ScrapeOptions] = None) -> Dict[str, Any]: diff --git a/apps/python-sdk/firecrawl/v2/methods/search.py b/apps/python-sdk/firecrawl/v2/methods/search.py index ea56a6e19..c60ed7a01 100644 --- a/apps/python-sdk/firecrawl/v2/methods/search.py +++ b/apps/python-sdk/firecrawl/v2/methods/search.py @@ -3,15 +3,14 @@ Search functionality for Firecrawl v2 API. """ from typing import Optional, Dict, Any, Union -from ...types import SearchRequest, SearchResponse, SearchResult, Document -from ..utils.http_client import HttpClient -from ..utils.error_handler import handle_response_error +from ...types import SearchRequest, SearchData, SearchResult, Document +from ..utils import HttpClient, handle_response_error, validate_scrape_options, prepare_scrape_options def search( client: HttpClient, request: SearchRequest -) -> SearchResponse: +) -> SearchData: """ Search for documents. @@ -20,7 +19,7 @@ def search( request: Search request Returns: - SearchResponse containing the search results + SearchData with search results grouped by source type Raises: FirecrawlError: If the search operation fails @@ -34,37 +33,35 @@ def search( response_data = response.json() - if response_data.get("success"): - data = response_data.get("data", {}) - grouped_results = {} - - for source_type, source_documents in data.items(): - if isinstance(source_documents, list): - results = [] - for doc_data in source_documents: - if isinstance(doc_data, dict): - if any(key in doc_data for key in ['markdown', 'html', 'content', 'screenshot']): - results.append(Document(**doc_data)) - else: - results.append(SearchResult( - url=doc_data.get('url', ''), - title=doc_data.get('title'), - description=doc_data.get('description') - )) - elif isinstance(doc_data, str): - results.append(SearchResult(url=doc_data)) - grouped_results[source_type] = results - - return SearchResponse( - success=True, - data=grouped_results, - warning=response_data.get("warning") - ) - else: - return SearchResponse( - success=False, - error=response_data.get("error", "Unknown error occurred") - ) + if not response_data.get("success"): + # Handle error case + error_msg = response_data.get("error", "Unknown error occurred") + raise Exception(f"Search failed: {error_msg}") + + data = response_data.get("data", {}) + search_data = SearchData() + + for source_type, source_documents in data.items(): + if isinstance(source_documents, list): + results = [] + for doc_data in source_documents: + if isinstance(doc_data, dict): + if any(key in doc_data for key in ['markdown', 'html', 'content', 'screenshot']): + results.append(Document(**doc_data)) + else: + results.append(SearchResult( + url=doc_data.get('url', ''), + title=doc_data.get('title'), + description=doc_data.get('description') + )) + elif isinstance(doc_data, str): + results.append(SearchResult(url=doc_data)) + + # Set the appropriate field based on source type + if hasattr(search_data, source_type): + setattr(search_data, source_type, results) + + return search_data def _validate_search_request(request: SearchRequest) -> SearchRequest: @@ -80,9 +77,52 @@ def _validate_search_request(request: SearchRequest) -> SearchRequest: Raises: ValueError: If request is invalid """ - # Validate timeout if present - if request.timeout is not None and request.timeout <= 0: - raise ValueError("Timeout must be positive") + # Validate query + if not request.query or not request.query.strip(): + raise ValueError("Query cannot be empty") + + # Validate limit + if request.limit is not None: + if request.limit <= 0: + raise ValueError("Limit must be positive") + if request.limit > 100: + raise ValueError("Limit cannot exceed 100") + + # Validate timeout + if request.timeout is not None: + if request.timeout <= 0: + raise ValueError("Timeout must be positive") + if request.timeout > 300000: # 5 minutes max + raise ValueError("Timeout cannot exceed 300000ms (5 minutes)") + + # Validate sources (if provided) + if request.sources is not None: + valid_sources = {"web", "news", "images"} + for source in request.sources: + if isinstance(source, str): + if source not in valid_sources: + raise ValueError(f"Invalid source type: {source}. Valid types: {valid_sources}") + elif hasattr(source, 'type'): + if source.type not in valid_sources: + raise ValueError(f"Invalid source type: {source.type}. Valid types: {valid_sources}") + + # Validate location (if provided) + if request.location is not None: + if not isinstance(request.location, str) or len(request.location.strip()) == 0: + raise ValueError("Location must be a non-empty string") + + # Validate tbs (time-based search, if provided) + if request.tbs is not None: + valid_tbs_values = { + "qdr:d", "qdr:w", "qdr:m", "qdr:y", # Google time filters + "d", "w", "m", "y" # Short forms + } + if request.tbs not in valid_tbs_values: + raise ValueError(f"Invalid tbs value: {request.tbs}. Valid values: {valid_tbs_values}") + + # Validate scrape_options (if provided) + if request.scrape_options is not None: + validate_scrape_options(request.scrape_options) return request @@ -98,4 +138,21 @@ def _prepare_search_request(request: SearchRequest) -> Dict[str, Any]: Request payload dictionary """ validated_request = _validate_search_request(request) - return validated_request.model_dump(exclude_none=True, by_alias=True) \ No newline at end of file + data = validated_request.model_dump(exclude_none=True, by_alias=True) + + # Handle snake_case to camelCase conversions manually + # (Pydantic Field() aliases interfere with value assignment) + + # ignore_invalid_urls → ignoreInvalidURLs + if validated_request.ignore_invalid_urls is not None: + data["ignoreInvalidURLs"] = validated_request.ignore_invalid_urls + data.pop("ignore_invalid_urls", None) + + # scrape_options → scrapeOptions + if validated_request.scrape_options is not None: + scrape_data = prepare_scrape_options(validated_request.scrape_options) + if scrape_data: + data["scrapeOptions"] = scrape_data + data.pop("scrape_options", None) + + return data \ No newline at end of file diff --git a/apps/python-sdk/firecrawl/v2/types.py b/apps/python-sdk/firecrawl/v2/types.py index 9884a1981..64fd8d97a 100644 --- a/apps/python-sdk/firecrawl/v2/types.py +++ b/apps/python-sdk/firecrawl/v2/types.py @@ -24,7 +24,7 @@ class DocumentMetadata(BaseModel): title: Optional[str] = None description: Optional[str] = None language: Optional[str] = None - keywords: Optional[str] = None + keywords: Optional[Union[str, List[str]]] = None robots: Optional[str] = None og_title: Optional[str] = Field(None, alias="ogTitle") og_description: Optional[str] = Field(None, alias="ogDescription") @@ -36,15 +36,15 @@ class DocumentMetadata(BaseModel): class Document(BaseModel): """A scraped document.""" - url: str markdown: Optional[str] = None html: Optional[str] = None raw_html: Optional[str] = Field(None, alias="rawHtml") - content: Optional[str] = None metadata: Optional[DocumentMetadata] = None links: Optional[List[str]] = None screenshot: Optional[str] = None actions: Optional[Dict[str, Any]] = None + warning: Optional[str] = None + change_tracking: Optional[Dict[str, Any]] = Field(None, alias="changeTracking") class Source(BaseModel): """Configuration for a search source.""" @@ -92,70 +92,144 @@ class ScrapeFormats(BaseModel): class ScrapeOptions(BaseModel): """Options for scraping operations.""" - formats: Optional[ScrapeFormats] = None + formats: Optional[Union[ScrapeFormats, List[FormatOption]]] = None headers: Optional[Dict[str, str]] = None - include_tags: Optional[List[str]] = Field(None, alias="includeTags") - exclude_tags: Optional[List[str]] = Field(None, alias="excludeTags") - only_main_content: bool = Field(True, alias="onlyMainContent") + include_tags: Optional[List[str]] = None + exclude_tags: Optional[List[str]] = None + only_main_content: bool = True timeout: Optional[int] = None - wait_for: Optional[int] = Field(None, alias="waitFor") + wait_for: Optional[int] = None mobile: bool = False - skip_tls_verification: bool = Field(False, alias="skipTlsVerification") - remove_base64_images: bool = Field(True, alias="removeBase64Images") + skip_tls_verification: bool = False + remove_base64_images: bool = True + # Note: raw_html and screenshot_full_page are not supported by v2 API yet + # raw_html: bool = False + # screenshot_full_page: bool = False + block_ads: bool = False + proxy: Optional[str] = None + max_age: Optional[int] = None + store_in_cache: bool = False + location: Optional['Location'] = None + actions: Optional[List[Union['WaitAction', 'ScreenshotAction', 'ClickAction', 'WriteAction', 'PressAction', 'ScrollAction', 'ScrapeAction', 'ExecuteJavascriptAction', 'PDFAction']]] = None + + @field_validator('formats') + @classmethod + def validate_formats(cls, v): + """Validate and normalize formats input.""" + if v is None: + return v + + # If it's already a ScrapeFormats object, return as is + if isinstance(v, ScrapeFormats): + return v + + # If it's a list, keep it as a list (don't convert to ScrapeFormats) + if isinstance(v, list): + return v + + raise ValueError(f"Invalid formats type: {type(v)}. Expected ScrapeFormats or List[FormatOption]") class ScrapeRequest(BaseModel): """Request for scraping a single URL.""" url: str options: Optional[ScrapeOptions] = None -class ScrapeResponse(BaseResponse[Document]): - """Response from scraping operation.""" +class ScrapeData(Document): + """Scrape results data.""" + pass + +class ScrapeResponse(BaseResponse[ScrapeData]): + """Response for scrape operations.""" pass # Crawl types -class CrawlOptions(BaseModel): - """Options for crawling operations.""" - includes: Optional[List[str]] = None - excludes: Optional[List[str]] = None - generate_img_alt_text: bool = Field(False, alias="generateImgAltText") - return_only_urls: bool = Field(False, alias="returnOnlyUrls") - max_depth: Optional[int] = Field(None, alias="maxDepth") - mode: Literal["default", "fast"] = "default" - ignore_sitemap: bool = Field(False, alias="ignoreSitemap") - limit: Optional[int] = None - allow_backward_crawling: bool = Field(False, alias="allowBackwardCrawling") - allow_external_content_links: bool = Field(False, alias="allowExternalContentLinks") - scrape_options: Optional[ScrapeOptions] = Field(None, alias="scrapeOptions") - class CrawlRequest(BaseModel): """Request for crawling a website.""" url: str - options: Optional[CrawlOptions] = None + prompt: Optional[str] = None + exclude_paths: Optional[List[str]] = None + include_paths: Optional[List[str]] = None + max_discovery_depth: Optional[int] = None + ignore_sitemap: bool = False + ignore_query_parameters: bool = False + limit: Optional[int] = None + crawl_entire_domain: bool = False + allow_external_links: bool = False + allow_subdomains: bool = False + delay: Optional[int] = None + max_concurrency: Optional[int] = None + webhook: Optional[Dict[str, Any]] = None + scrape_options: Optional[ScrapeOptions] = None + zero_data_retention: bool = False class CrawlJob(BaseModel): """Information about a crawl job.""" id: str url: str - status: Literal["scraping", "completed", "failed", "cancelled"] + status: Literal["scraping", "completed", "failed"] current: Optional[int] = None total: Optional[int] = None created_at: Optional[datetime] = Field(None, alias="createdAt") completed_at: Optional[datetime] = Field(None, alias="completedAt") + data: Optional[List[Document]] = None + partial_data: Optional[List[Document]] = Field(None, alias="partialData") -class CrawlResponse(BaseResponse[CrawlJob]): - """Response from starting a crawl.""" - pass - -class CrawlStatusData(BaseModel): - """Data for crawl status response.""" - status: Literal["scraping", "completed", "failed", "cancelled"] +class CrawlJobData(BaseModel): + """Crawl job status and progress data.""" + id: str + status: Literal["scraping", "completed", "failed"] current: int total: int data: List[Document] partial_data: Optional[List[Document]] = Field(None, alias="partialData") -class CrawlStatusResponse(BaseResponse[CrawlStatusData]): - """Response from checking crawl status.""" +class CrawlData(List[Document]): + """Crawl results - just the documents.""" + pass + +class SearchDocument(Document): + """A document from a search operation with URL and description.""" + url: str + description: Optional[str] = None + +class MapDocument(Document): + """A document from a map operation with URL and description.""" + url: str + description: Optional[str] = None + +class CrawlStartResponse(BaseResponse[CrawlJob]): + """Response for starting a crawl job.""" + pass + +class CrawlResponse(BaseResponse[CrawlJobData]): + """Response for crawl operations.""" + pass + +# Crawl params types +class CrawlParamsRequest(BaseModel): + """Request for getting crawl parameters from LLM.""" + url: str + prompt: str + +class CrawlParamsData(BaseModel): + """Data returned from crawl params endpoint.""" + include_paths: Optional[List[str]] = None + exclude_paths: Optional[List[str]] = None + max_discovery_depth: Optional[int] = None + ignore_sitemap: bool = False + ignore_query_parameters: bool = False + limit: Optional[int] = None + crawl_entire_domain: bool = False + allow_external_links: bool = False + allow_subdomains: bool = False + delay: Optional[int] = None + max_concurrency: Optional[int] = None + webhook: Optional[Dict[str, Any]] = None + scrape_options: Optional[ScrapeOptions] = None + zero_data_retention: bool = False + +class CrawlParamsResponse(BaseResponse[CrawlParamsData]): + """Response from crawl params endpoint.""" pass # Batch scrape types @@ -167,25 +241,21 @@ class BatchScrapeRequest(BaseModel): class BatchScrapeJob(BaseModel): """Information about a batch scrape job.""" id: str - status: Literal["scraping", "completed", "failed", "cancelled"] + status: Literal["scraping", "completed", "failed"] current: Optional[int] = None total: Optional[int] = None created_at: Optional[datetime] = Field(None, alias="createdAt") completed_at: Optional[datetime] = Field(None, alias="completedAt") -class BatchScrapeResponse(BaseResponse[BatchScrapeJob]): - """Response from starting a batch scrape.""" - pass - -class BatchScrapeStatusData(BaseModel): - """Data for batch scrape status response.""" - status: Literal["scraping", "completed", "failed", "cancelled"] +class BatchScrapeData(BaseModel): + """Batch scrape results data.""" + status: Literal["scraping", "completed", "failed"] current: int total: int data: List[Document] -class BatchScrapeStatusResponse(BaseResponse[BatchScrapeStatusData]): - """Response from checking batch scrape status.""" +class BatchScrapeResponse(BaseResponse[BatchScrapeData]): + """Response for batch scrape operations.""" pass # Map types @@ -202,11 +272,11 @@ class MapRequest(BaseModel): options: Optional[MapOptions] = None class MapData(BaseModel): - """Data for map response.""" + """Map results data.""" links: List[str] class MapResponse(BaseResponse[MapData]): - """Response from mapping operation.""" + """Response for map operations.""" pass # Action types @@ -270,7 +340,6 @@ class JsonFormat(BaseModel): """Configuration for JSON extraction.""" prompt: Optional[str] = None schema_field: Optional[Dict[str, Any]] = Field(None, alias="schema") - system_prompt: Optional[str] = Field(None, alias="systemPrompt") class SearchRequest(BaseModel): """Request for search operations.""" @@ -279,9 +348,9 @@ class SearchRequest(BaseModel): limit: Optional[int] = 5 tbs: Optional[str] = None location: Optional[str] = None - ignore_invalid_urls: Optional[bool] = Field(True, alias="ignoreInvalidURLs") + ignore_invalid_urls: Optional[bool] = None timeout: Optional[int] = 60000 - scrape_options: Optional[ScrapeOptions] = Field(None, alias="scrapeOptions") + scrape_options: Optional[ScrapeOptions] = None @field_validator('sources') @classmethod @@ -309,7 +378,13 @@ class SearchResult(BaseModel): title: Optional[str] = None description: Optional[str] = None -class SearchResponse(BaseResponse[Dict[str, List[Union[SearchResult, Document]]]]): +class SearchData(BaseModel): + """Search results grouped by source type.""" + web: Optional[List[Union[SearchResult, SearchDocument]]] = None + news: Optional[List[Union[SearchResult, SearchDocument]]] = None + images: Optional[List[Union[SearchResult, SearchDocument]]] = None + +class SearchResponse(BaseResponse[SearchData]): """Response from search operation.""" pass @@ -330,7 +405,7 @@ class ErrorResponse(BaseModel): class JobStatus(BaseModel): """Generic job status information.""" id: str - status: Literal["pending", "scraping", "completed", "failed", "cancelled"] + status: Literal["pending", "scraping", "completed", "failed"] current: Optional[int] = None total: Optional[int] = None created_at: Optional[datetime] = Field(None, alias="createdAt") @@ -358,18 +433,16 @@ class ClientConfig(BaseModel): # Union types for convenience ScrapeResult = Union[Document, List[Document]] -CrawlResult = Union[CrawlJob, CrawlStatusData] -BatchResult = Union[BatchScrapeJob, BatchScrapeStatusData] +CrawlResult = Union[CrawlJob, CrawlJobData] +BatchResult = Union[BatchScrapeJob, BatchScrapeData] JobResult = Union[CrawlJob, BatchScrapeJob] -StatusResult = Union[CrawlStatusData, BatchScrapeStatusData] +StatusResult = Union[CrawlJobData, BatchScrapeData] # Response union types AnyResponse = Union[ ScrapeResponse, CrawlResponse, - CrawlStatusResponse, BatchScrapeResponse, - BatchScrapeStatusResponse, MapResponse, SearchResponse, ErrorResponse diff --git a/apps/python-sdk/firecrawl/v2/utils/__init__.py b/apps/python-sdk/firecrawl/v2/utils/__init__.py index 37457ba29..58c18b7bf 100644 --- a/apps/python-sdk/firecrawl/v2/utils/__init__.py +++ b/apps/python-sdk/firecrawl/v2/utils/__init__.py @@ -4,5 +4,6 @@ Utility modules for v2 API client. from .http_client import HttpClient from .error_handler import FirecrawlError, handle_response_error +from .validation import validate_scrape_options, prepare_scrape_options -__all__ = ['HttpClient', 'FirecrawlError', 'handle_response_error'] \ No newline at end of file +__all__ = ['HttpClient', 'FirecrawlError', 'handle_response_error', 'validate_scrape_options', 'prepare_scrape_options'] \ No newline at end of file diff --git a/apps/python-sdk/firecrawl/v2/utils/validation.py b/apps/python-sdk/firecrawl/v2/utils/validation.py new file mode 100644 index 000000000..4782be013 --- /dev/null +++ b/apps/python-sdk/firecrawl/v2/utils/validation.py @@ -0,0 +1,115 @@ +""" +Shared validation functions for Firecrawl v2 API. +""" + +from typing import Optional, Dict, Any +from ..types import ScrapeOptions + + +def validate_scrape_options(options: Optional[ScrapeOptions]) -> Optional[ScrapeOptions]: + """ + Validate and normalize scrape options. + + Args: + options: Scraping options to validate + + Returns: + Validated options or None + + Raises: + ValueError: If options are invalid + """ + if options is None: + return None + + # Validate timeout + if options.timeout is not None and options.timeout <= 0: + raise ValueError("Timeout must be positive") + + # Validate wait_for + if options.wait_for is not None and options.wait_for < 0: + raise ValueError("wait_for must be non-negative") + + return options + + +def prepare_scrape_options(options: Optional[ScrapeOptions]) -> Optional[Dict[str, Any]]: + """ + Prepare ScrapeOptions for API submission with manual snake_case to camelCase conversion. + + Args: + options: ScrapeOptions to prepare + + Returns: + Dictionary ready for API submission or None if options is None + """ + if options is None: + return None + + # Validate options first + validated_options = validate_scrape_options(options) + if validated_options is None: + return None + + # Convert to dict and handle manual snake_case to camelCase conversion + options_data = validated_options.model_dump(exclude_none=True) + scrape_data = {} + + for key, value in options_data.items(): + if value is not None: + if key == "formats": + # Handle formats - if it's a ScrapeFormats object, convert boolean flags to format strings + if hasattr(value, 'formats') and value.formats is not None: + scrape_data["formats"] = value.formats + elif hasattr(value, 'markdown') or hasattr(value, 'html') or hasattr(value, 'raw_html') or hasattr(value, 'content') or hasattr(value, 'links') or hasattr(value, 'screenshot') or hasattr(value, 'screenshot_full_page'): + # Convert ScrapeFormats boolean flags to format strings + formats = [] + if getattr(value, 'markdown', False): + formats.append("markdown") + if getattr(value, 'html', False): + formats.append("html") + if getattr(value, 'raw_html', False): + formats.append("rawHtml") + if getattr(value, 'content', False): + formats.append("content") + if getattr(value, 'links', False): + formats.append("links") + if getattr(value, 'screenshot', False): + formats.append("screenshot") + if getattr(value, 'screenshot_full_page', False): + formats.append("screenshot@fullPage") + scrape_data["formats"] = formats + elif isinstance(value, list): + # If it's already a list, use it directly + scrape_data["formats"] = value + else: + # Fallback - use the value as-is + scrape_data["formats"] = value + elif key == "include_tags": + scrape_data["includeTags"] = value + elif key == "exclude_tags": + scrape_data["excludeTags"] = value + elif key == "only_main_content": + scrape_data["onlyMainContent"] = value + elif key == "wait_for": + scrape_data["waitFor"] = value + elif key == "skip_tls_verification": + scrape_data["skipTlsVerification"] = value + elif key == "remove_base64_images": + scrape_data["removeBase64Images"] = value + elif key == "block_ads": + scrape_data["blockAds"] = value + elif key == "store_in_cache": + scrape_data["storeInCache"] = value + elif key == "max_age": + scrape_data["maxAge"] = value + # Note: raw_html and screenshot_full_page are not supported by v2 API yet + # elif key == "raw_html": + # scrape_data["rawHtml"] = value + # elif key == "screenshot_full_page": + # scrape_data["screenshot@fullPage"] = value + else: + # For fields that don't need conversion, use as-is + scrape_data[key] = value + + return scrape_data \ No newline at end of file