Coverage for haystack/components/fetchers/link_content.py: 91%
202 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 13:53 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5import asyncio
6from collections import defaultdict
7from collections.abc import Callable
8from concurrent.futures import ThreadPoolExecutor
9from dataclasses import replace
10from fnmatch import fnmatch
11from typing import Any, cast
13import httpx
14from tenacity import RetryCallState, retry, retry_if_exception_type, stop_after_attempt, wait_exponential
16from haystack import component, logging
17from haystack.dataclasses import ByteStream
18from haystack.lazy_imports import LazyImport
19from haystack.version import __version__
21# HTTP/2 support via lazy import
22with LazyImport("Run 'pip install httpx[http2]' to use HTTP/2 support") as h2_import:
23 pass # nothing to import as we simply set the http2 attribute, library handles the rest
25logger = logging.getLogger(__name__)
27DEFAULT_USER_AGENT = f"haystack/LinkContentFetcher/{__version__}"
29REQUEST_HEADERS = {
30 "accept": "*/*",
31 "User-Agent": DEFAULT_USER_AGENT,
32 "Accept-Language": "en-US,en;q=0.9,it;q=0.8,es;q=0.7",
33 "referer": "https://www.google.com/",
34}
37def _merge_headers(*args: dict[str, str]) -> dict[str, str]:
38 """
39 Merge a list of dict using case-insensitively
41 :param args: a list of dict to merge
42 :returns: The merged dict
43 """
44 merged = {}
45 keymap = {}
47 for d in args:
48 for k, v in d.items():
49 kl = k.lower()
50 keymap[kl] = k
51 merged[kl] = v
53 return {keymap[kl]: v for kl, v in merged.items()}
56def _text_content_handler(response: httpx.Response) -> ByteStream:
57 """
58 Handles text content.
60 :param response: Response object from the request.
61 :returns: The extracted text.
62 """
63 return ByteStream.from_string(response.text)
66def _binary_content_handler(response: httpx.Response) -> ByteStream:
67 """
68 Handles binary content.
70 :param response: Response object from the request.
71 :returns: The extracted binary file-like object.
72 """
73 return ByteStream(data=response.content)
76@component
77class LinkContentFetcher:
78 """
79 Fetches and extracts content from URLs.
81 It supports various content types, retries on failures, and automatic user-agent rotation for failed web
82 requests. Use it as the data-fetching step in your pipelines.
84 You may need to convert LinkContentFetcher's output into a list of documents. Use HTMLToDocument
85 converter to do this.
87 ### Usage example
89 ```python
90 from haystack.components.fetchers.link_content import LinkContentFetcher
92 fetcher = LinkContentFetcher()
93 streams = fetcher.run(urls=["https://www.google.com"])["streams"]
95 assert len(streams) == 1
96 assert streams[0].meta == {'content_type': 'text/html', 'url': 'https://www.google.com'}
97 assert streams[0].data
98 ```
100 For async usage:
102 ```python
103 import asyncio
104 from haystack.components.fetchers import LinkContentFetcher
106 async def fetch_async():
107 fetcher = LinkContentFetcher()
108 result = await fetcher.run_async(urls=["https://www.google.com"])
109 return result["streams"]
111 streams = asyncio.run(fetch_async())
112 ```
113 """
115 def __init__(
116 self,
117 raise_on_failure: bool = True,
118 user_agents: list[str] | None = None,
119 retry_attempts: int = 2,
120 timeout: int = 3,
121 http2: bool = False,
122 client_kwargs: dict | None = None,
123 request_headers: dict[str, str] | None = None,
124 ) -> None:
125 """
126 Initializes the component.
128 :param raise_on_failure: If `True`, raises an exception if it fails to fetch a single URL.
129 For multiple URLs, it logs errors and returns the content it successfully fetched.
130 :param user_agents: [User agents](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/User-Agent)
131 for fetching content. If `None`, a default user agent is used.
132 :param retry_attempts: The number of times to retry to fetch the URL's content.
133 :param timeout: Timeout in seconds for the request.
134 :param http2: Whether to enable HTTP/2 support for requests. Defaults to False.
135 Requires the 'h2' package to be installed (via `pip install httpx[http2]`).
136 :param client_kwargs: Additional keyword arguments to pass to the httpx client.
137 If `None`, default values are used.
138 :param request_headers: Additional headers to send with every request. These take precedence over the
139 component's default headers but not over the rotating `User-Agent`.
140 """
141 self.raise_on_failure = raise_on_failure
142 self.user_agents = user_agents or [DEFAULT_USER_AGENT]
143 self.retry_attempts = retry_attempts
144 self.timeout = timeout
145 self.http2 = http2
146 self.client_kwargs = client_kwargs or {}
147 self.request_headers = request_headers or {}
149 # Configure default client settings
150 self.client_kwargs.setdefault("timeout", timeout)
151 self.client_kwargs.setdefault("follow_redirects", True)
153 # httpx clients are built lazily in warm_up / warm_up_async (resource lifecycle)
154 self._client: httpx.Client | None = None
155 self._async_client: httpx.AsyncClient | None = None
157 # register default content handlers that extract data from the response
158 self.handlers: dict[str, Callable[[httpx.Response], ByteStream]] = defaultdict(lambda: _text_content_handler)
159 self.handlers["text/*"] = _text_content_handler
160 self.handlers["text/html"] = _binary_content_handler
161 self.handlers["application/json"] = _text_content_handler
162 self.handlers["application/*"] = _binary_content_handler
163 self.handlers["image/*"] = _binary_content_handler
164 self.handlers["audio/*"] = _binary_content_handler
165 self.handlers["video/*"] = _binary_content_handler
167 def _get_response(self, url: str) -> httpx.Response:
168 """
169 Gets a response from a URL, rotating the user agent on every failed attempt.
171 The rotation cursor is local to this call: `run` fetches URLs concurrently, so a cursor kept on the
172 component would be advanced and reset by whichever fetches happen to be in flight at the same time.
174 :param url: The URL to fetch.
175 :returns: The httpx Response object.
176 """
177 user_agent_idx = 0
179 def rotate_user_agent(retry_state: RetryCallState) -> None: # noqa: ARG001
180 nonlocal user_agent_idx
181 user_agent_idx = (user_agent_idx + 1) % len(self.user_agents)
182 logger.debug("Switched user agent to {user_agent}", user_agent=self.user_agents[user_agent_idx])
184 @retry(
185 reraise=True,
186 stop=stop_after_attempt(self.retry_attempts + 1),
187 wait=wait_exponential(multiplier=1, min=2, max=10),
188 retry=(retry_if_exception_type((httpx.HTTPStatusError, httpx.RequestError))),
189 # This callback is invoked only after failed requests (exception raised)
190 after=rotate_user_agent,
191 )
192 def get_response(url: str) -> httpx.Response:
193 assert self._client is not None # mypy: client is built by warm_up before run
194 response = self._client.get(url, headers=self._get_headers(self.user_agents[user_agent_idx]))
195 response.raise_for_status()
196 return response
198 return get_response(url)
200 def _build_client_kwargs(self) -> dict[str, Any]:
201 """
202 Build the keyword arguments used to construct the httpx clients.
204 Resolves optional HTTP/2 support, downgrading to HTTP/1.1 if the 'h2' package is not installed.
205 """
206 client_kwargs = {**self.client_kwargs}
208 # Optional HTTP/2 support
209 if self.http2:
210 try:
211 h2_import.check()
212 client_kwargs["http2"] = True
213 except ImportError:
214 logger.warning(
215 "HTTP/2 support requested but 'h2' package is not installed. "
216 "Falling back to HTTP/1.1. Install with `pip install httpx[http2]` to enable HTTP/2 support."
217 )
218 self.http2 = False # Update the setting to match actual capability
220 return client_kwargs
222 def warm_up(self) -> None:
223 """
224 Initializes the synchronous httpx client.
225 """
226 if self._client is None:
227 self._client = httpx.Client(**self._build_client_kwargs())
229 async def warm_up_async(self) -> None: # noqa: RUF029
230 """
231 Initializes the asynchronous httpx client on the serving event loop.
232 """
233 if self._async_client is None:
234 self._async_client = httpx.AsyncClient(**self._build_client_kwargs())
236 def close(self) -> None:
237 """
238 Releases the synchronous httpx client.
239 """
240 if self._client is not None:
241 self._client.close()
242 self._client = None
244 async def close_async(self) -> None:
245 """
246 Releases the asynchronous httpx client.
247 """
248 if self._async_client is not None:
249 await self._async_client.aclose()
250 self._async_client = None
252 def _get_headers(self, user_agent: str) -> dict[str, str]:
253 """
254 Build headers with precedence
256 client defaults -> component defaults -> user-provided -> rotating UA
258 :param user_agent: The user agent for this attempt, taken from the caller's own rotation.
259 """
260 base = dict(self._client.headers) if self._client is not None else {}
261 return _merge_headers(base, REQUEST_HEADERS, self.request_headers, {"User-Agent": user_agent})
263 @component.output_types(streams=list[ByteStream])
264 def run(self, urls: list[str]) -> dict[str, Any]:
265 """
266 Fetches content from a list of URLs and returns a list of extracted content streams.
268 Each content stream is a `ByteStream` object containing the extracted content as binary data.
269 Each ByteStream object in the returned list corresponds to the contents of a single URL.
270 The content type of each stream is stored in the metadata of the ByteStream object under
271 the key "content_type". The URL of the fetched content is stored under the key "url".
273 :param urls: A list of URLs to fetch content from.
274 :returns: `ByteStream` objects representing the extracted content.
276 :raises Exception: If the provided list of URLs contains only a single URL, and `raise_on_failure` is set to
277 `True`, an exception will be raised in case of an error during content retrieval.
278 In all other scenarios, any retrieval errors are logged, and a list of successfully retrieved `ByteStream`
279 objects is returned.
280 """
281 self.warm_up()
283 streams: list[ByteStream] = []
284 if not urls:
285 return {"streams": streams}
287 # don't use multithreading if there's only one URL
288 if len(urls) == 1:
289 stream_metadata, stream = self._fetch(urls[0])
290 stream.meta.update(stream_metadata)
291 stream = replace(stream, mime_type=stream.meta.get("content_type", None))
292 streams.append(stream)
293 else:
294 with ThreadPoolExecutor() as executor:
295 results = executor.map(self._fetch_with_exception_suppression, urls)
297 for stream_metadata, stream in results: # type: ignore
298 if stream_metadata is not None and stream is not None:
299 stream.meta.update(stream_metadata)
300 stream = replace(stream, mime_type=stream.meta.get("content_type", None))
301 streams.append(stream)
303 return {"streams": streams}
305 @component.output_types(streams=list[ByteStream])
306 async def run_async(self, urls: list[str]) -> dict[str, Any]:
307 """
308 Asynchronously fetches content from a list of URLs and returns a list of extracted content streams.
310 This is the asynchronous version of the `run` method with the same parameters and return values.
312 :param urls: A list of URLs to fetch content from.
313 :returns: `ByteStream` objects representing the extracted content.
314 """
315 await self.warm_up_async()
317 streams: list[ByteStream] = []
318 if not urls:
319 return {"streams": streams}
321 assert self._async_client is not None # mypy: async_client is built by warm_up_async above
322 # Create tasks for all URLs using _fetch_async directly
323 tasks = [self._fetch_async(url, self._async_client) for url in urls]
325 # Only capture exceptions when we have multiple URLs or raise_on_failure=False
326 # This ensures errors propagate appropriately for single URLs with raise_on_failure=True
327 return_exceptions = not (len(urls) == 1 and self.raise_on_failure)
328 results = await asyncio.gather(*tasks, return_exceptions=return_exceptions)
330 # Process results
331 for i, result in enumerate(results):
332 # Handle exception results (only happens when return_exceptions=True)
333 if isinstance(result, Exception):
334 logger.warning("Error fetching {url}: {error}", url=urls[i], error=str(result))
335 # Add an empty result for failed URLs when raise_on_failure=False
336 if not self.raise_on_failure:
337 streams.append(ByteStream(data=b"", meta={"content_type": "Unknown", "url": urls[i]}))
338 continue
340 # Process successful results
341 # At this point, result is not an exception, so we need to cast it to the correct type for mypy
342 if not isinstance(result, Exception): # Runtime check
343 # Use cast to tell mypy that result is the tuple type returned by _fetch_async
344 result_tuple = cast(tuple[dict[str, str] | None, ByteStream | None], result)
345 stream_metadata, stream = result_tuple
346 if stream_metadata is not None and stream is not None:
347 stream.meta.update(stream_metadata)
348 stream = replace(stream, mime_type=stream.meta.get("content_type", None))
349 streams.append(stream)
351 return {"streams": streams}
353 def _fetch(self, url: str) -> tuple[dict[str, str], ByteStream]:
354 """
355 Fetches content from a URL and returns it as a ByteStream.
357 :param url: The URL to fetch content from.
358 :returns: A tuple containing the ByteStream metadata dict and the corresponding ByteStream.
359 ByteStream metadata contains the URL and the content type of the fetched content.
360 The content type is a string indicating the type of content fetched (for example, "text/html",
361 "application/pdf"). The ByteStream object contains the fetched content as binary data.
363 :raises: If an error occurs during content retrieval and `raise_on_failure` is set to True, this method will
364 raise an exception. Otherwise, all fetching errors are logged, and an empty ByteStream is returned.
366 """
367 content_type: str = "text/html"
368 stream: ByteStream = ByteStream(data=b"")
369 try:
370 response = self._get_response(url)
371 content_type = self._get_content_type(response)
372 handler: Callable = self._resolve_handler(content_type)
373 stream = handler(response)
374 except Exception as e:
375 if self.raise_on_failure:
376 raise e
377 # less verbose log as this is expected to happen often (requests failing, blocked, etc.)
378 logger.debug("Couldn't retrieve content from {url} because {error}", url=url, error=str(e))
380 return {"content_type": content_type, "url": url}, stream
382 async def _fetch_async(
383 self, url: str, client: httpx.AsyncClient
384 ) -> tuple[dict[str, str] | None, ByteStream | None]:
385 """
386 Asynchronously fetches content from a URL and returns it as a ByteStream.
388 :param url: The URL to fetch content from.
389 :param client: The async httpx client to use for making requests.
390 :returns: A tuple containing the ByteStream metadata dict and the corresponding ByteStream.
391 """
392 content_type: str = "text/html"
393 stream: ByteStream | None = None
394 metadata: dict[str, str] | None = None
396 try:
397 response = await self._get_response_async(url, client)
398 content_type = self._get_content_type(response)
399 handler: Callable = self._resolve_handler(content_type)
400 stream = handler(response)
401 metadata = {"content_type": content_type, "url": url}
402 except Exception as e:
403 if self.raise_on_failure:
404 raise e
405 logger.debug("Couldn't retrieve content from {url} because {error}", url=url, error=str(e))
406 # Create an empty ByteStream for failed requests when raise_on_failure is False
407 stream = ByteStream(data=b"")
408 metadata = {"content_type": content_type, "url": url}
410 return metadata, stream
412 def _fetch_with_exception_suppression(self, url: str) -> tuple[dict[str, str] | None, ByteStream | None]:
413 """
414 Fetches content from a URL and returns it as a ByteStream.
416 If `raise_on_failure` is set to True, this method will wrap the fetch() method and catch any exceptions.
417 Otherwise, it will simply call the fetch() method.
418 :param url: The URL to fetch content from.
419 :returns: A tuple containing the ByteStream metadata dict and the corresponding ByteStream.
421 """
422 if self.raise_on_failure:
423 try:
424 return self._fetch(url)
425 except Exception as e:
426 logger.warning("Error fetching {url}: {error}", url=url, error=str(e))
427 return {"content_type": "Unknown", "url": url}, None
428 else:
429 return self._fetch(url)
431 async def _get_response_async(self, url: str, client: httpx.AsyncClient) -> httpx.Response:
432 """
433 Asynchronously gets a response from a URL with retry logic.
435 :param url: The URL to fetch.
436 :param client: The async httpx client to use for making requests.
437 :returns: The httpx Response object.
438 """
439 attempt = 0
440 last_exception = None
441 # Local to this call: `run_async` gathers URLs concurrently, so a cursor on the component
442 # would be shared by every request in flight.
443 user_agent_idx = 0
445 while attempt <= self.retry_attempts:
446 try:
447 response = await client.get(url, headers=self._get_headers(self.user_agents[user_agent_idx]))
448 response.raise_for_status()
449 return response
450 except (httpx.HTTPStatusError, httpx.RequestError) as e:
451 last_exception = e
452 attempt += 1
453 if attempt <= self.retry_attempts:
454 # Switch user agent for next retry
455 user_agent_idx = (user_agent_idx + 1) % len(self.user_agents)
456 # Wait before retry using exponential backoff
457 await asyncio.sleep(min(2 * 2 ** (attempt - 1), 10))
458 else:
459 break
461 # If we've exhausted all retries, raise the last exception
462 if last_exception:
463 raise last_exception
465 # This should never happen, but just in case
466 raise httpx.RequestError("Failed to get response after retries", request=None)
468 def _get_content_type(self, response: httpx.Response) -> str:
469 """
470 Get the content type of the response.
472 :param response: The response object.
473 :returns: The content type of the response.
474 """
475 content_type = response.headers.get("Content-Type", "")
476 return content_type.split(";")[0]
478 def _resolve_handler(self, content_type: str) -> Callable[[httpx.Response], ByteStream]:
479 """
480 Resolves the handler for the given content type.
482 First, it tries to find a direct match for the content type in the handlers dictionary.
483 If no direct match is found, it tries to find a pattern match using the fnmatch function.
484 If no pattern match is found, it returns the default handler for text/plain.
486 :param content_type: The content type to resolve the handler for.
487 :returns: The handler for the given content type, if found. Otherwise, the default handler for text/plain.
488 """
489 # direct match
490 if content_type in self.handlers:
491 return self.handlers[content_type]
493 # pattern matches
494 for pattern, handler in self.handlers.items():
495 if fnmatch(content_type, pattern):
496 return handler
498 # default handler
499 return self.handlers["text/plain"]