Coverage for haystack/components/fetchers/link_content.py: 91%
202 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-12 09:42 +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 = dict(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(
195 url, headers=self._get_headers(client=self._client, user_agent=self.user_agents[user_agent_idx])
196 )
197 response.raise_for_status()
198 return response
200 return get_response(url)
202 def _build_client_kwargs(self) -> dict[str, Any]:
203 """
204 Build the keyword arguments used to construct the httpx clients.
206 Resolves optional HTTP/2 support, downgrading to HTTP/1.1 if the 'h2' package is not installed.
207 """
208 client_kwargs = {**self.client_kwargs}
210 # Optional HTTP/2 support
211 if self.http2:
212 try:
213 h2_import.check()
214 client_kwargs["http2"] = True
215 except ImportError:
216 logger.warning(
217 "HTTP/2 support requested but 'h2' package is not installed. "
218 "Falling back to HTTP/1.1. Install with `pip install httpx[http2]` to enable HTTP/2 support."
219 )
220 self.http2 = False # Update the setting to match actual capability
222 return client_kwargs
224 def warm_up(self) -> None:
225 """
226 Initializes the synchronous httpx client.
227 """
228 if self._client is None:
229 self._client = httpx.Client(**self._build_client_kwargs())
231 async def warm_up_async(self) -> None: # noqa: RUF029
232 """
233 Initializes the asynchronous httpx client on the serving event loop.
234 """
235 if self._async_client is None:
236 self._async_client = httpx.AsyncClient(**self._build_client_kwargs())
238 def close(self) -> None:
239 """
240 Releases the synchronous httpx client.
241 """
242 if self._client is not None:
243 self._client.close()
244 self._client = None
246 async def close_async(self) -> None:
247 """
248 Releases the asynchronous httpx client.
249 """
250 if self._async_client is not None:
251 await self._async_client.aclose()
252 self._async_client = None
254 def _get_headers(self, client: httpx.Client | httpx.AsyncClient, user_agent: str) -> dict[str, str]:
255 """
256 Build headers with precedence
258 client defaults -> component defaults -> user-provided -> rotating UA
260 :param client: The HTTP client used for the request.
261 :param user_agent: The user agent for this attempt, taken from the caller's own rotation.
262 """
263 base = dict(client.headers)
264 return _merge_headers(base, REQUEST_HEADERS, self.request_headers, {"User-Agent": user_agent})
266 @component.output_types(streams=list[ByteStream])
267 def run(self, urls: list[str]) -> dict[str, Any]:
268 """
269 Fetches content from a list of URLs and returns a list of extracted content streams.
271 Each content stream is a `ByteStream` object containing the extracted content as binary data.
272 Each ByteStream object in the returned list corresponds to the contents of a single URL.
273 The content type of each stream is stored in the metadata of the ByteStream object under
274 the key "content_type". The URL of the fetched content is stored under the key "url".
276 :param urls: A list of URLs to fetch content from.
277 :returns: `ByteStream` objects representing the extracted content.
279 :raises Exception: If the provided list of URLs contains only a single URL, and `raise_on_failure` is set to
280 `True`, an exception will be raised in case of an error during content retrieval.
281 In all other scenarios, any retrieval errors are logged, and a list of successfully retrieved `ByteStream`
282 objects is returned.
283 """
284 self.warm_up()
286 streams: list[ByteStream] = []
287 if not urls:
288 return {"streams": streams}
290 # don't use multithreading if there's only one URL
291 if len(urls) == 1:
292 stream_metadata, stream = self._fetch(urls[0])
293 stream.meta.update(stream_metadata)
294 stream = replace(stream, mime_type=stream.meta.get("content_type", None))
295 streams.append(stream)
296 else:
297 with ThreadPoolExecutor() as executor:
298 results = executor.map(self._fetch_with_exception_suppression, urls)
300 for stream_metadata, stream in results: # type: ignore
301 if stream_metadata is not None and stream is not None:
302 stream.meta.update(stream_metadata)
303 stream = replace(stream, mime_type=stream.meta.get("content_type", None))
304 streams.append(stream)
306 return {"streams": streams}
308 @component.output_types(streams=list[ByteStream])
309 async def run_async(self, urls: list[str]) -> dict[str, Any]:
310 """
311 Asynchronously fetches content from a list of URLs and returns a list of extracted content streams.
313 This is the asynchronous version of the `run` method with the same parameters and return values.
315 :param urls: A list of URLs to fetch content from.
316 :returns: `ByteStream` objects representing the extracted content.
317 """
318 await self.warm_up_async()
320 streams: list[ByteStream] = []
321 if not urls:
322 return {"streams": streams}
324 assert self._async_client is not None # mypy: async_client is built by warm_up_async above
325 # Create tasks for all URLs using _fetch_async directly
326 tasks = [self._fetch_async(url, self._async_client) for url in urls]
328 # Only capture exceptions when we have multiple URLs or raise_on_failure=False
329 # This ensures errors propagate appropriately for single URLs with raise_on_failure=True
330 return_exceptions = not (len(urls) == 1 and self.raise_on_failure)
331 results = await asyncio.gather(*tasks, return_exceptions=return_exceptions)
333 # Process results
334 for i, result in enumerate(results):
335 # Handle exception results (only happens when return_exceptions=True)
336 if isinstance(result, Exception):
337 logger.warning("Error fetching {url}: {error}", url=urls[i], error=str(result))
338 # Add an empty result for failed URLs when raise_on_failure=False
339 if not self.raise_on_failure:
340 streams.append(ByteStream(data=b"", meta={"content_type": "Unknown", "url": urls[i]}))
341 continue
343 # Process successful results
344 # At this point, result is not an exception, so we need to cast it to the correct type for mypy
345 if not isinstance(result, Exception): # Runtime check
346 # Use cast to tell mypy that result is the tuple type returned by _fetch_async
347 result_tuple = cast(tuple[dict[str, str] | None, ByteStream | None], result)
348 stream_metadata, stream = result_tuple
349 if stream_metadata is not None and stream is not None:
350 stream.meta.update(stream_metadata)
351 stream = replace(stream, mime_type=stream.meta.get("content_type", None))
352 streams.append(stream)
354 return {"streams": streams}
356 def _fetch(self, url: str) -> tuple[dict[str, str], ByteStream]:
357 """
358 Fetches content from a URL and returns it as a ByteStream.
360 :param url: The URL to fetch content from.
361 :returns: A tuple containing the ByteStream metadata dict and the corresponding ByteStream.
362 ByteStream metadata contains the URL and the content type of the fetched content.
363 The content type is a string indicating the type of content fetched (for example, "text/html",
364 "application/pdf"). The ByteStream object contains the fetched content as binary data.
366 :raises: If an error occurs during content retrieval and `raise_on_failure` is set to True, this method will
367 raise an exception. Otherwise, all fetching errors are logged, and an empty ByteStream is returned.
369 """
370 content_type: str = "text/html"
371 stream: ByteStream = ByteStream(data=b"")
372 try:
373 response = self._get_response(url)
374 content_type = self._get_content_type(response)
375 handler: Callable = self._resolve_handler(content_type)
376 stream = handler(response)
377 except Exception as e:
378 if self.raise_on_failure:
379 raise e
380 # less verbose log as this is expected to happen often (requests failing, blocked, etc.)
381 logger.debug("Couldn't retrieve content from {url} because {error}", url=url, error=str(e))
383 return {"content_type": content_type, "url": url}, stream
385 async def _fetch_async(
386 self, url: str, client: httpx.AsyncClient
387 ) -> tuple[dict[str, str] | None, ByteStream | None]:
388 """
389 Asynchronously fetches content from a URL and returns it as a ByteStream.
391 :param url: The URL to fetch content from.
392 :param client: The async httpx client to use for making requests.
393 :returns: A tuple containing the ByteStream metadata dict and the corresponding ByteStream.
394 """
395 content_type: str = "text/html"
396 stream: ByteStream | None = None
397 metadata: dict[str, str] | None = None
399 try:
400 response = await self._get_response_async(url, client)
401 content_type = self._get_content_type(response)
402 handler: Callable = self._resolve_handler(content_type)
403 stream = handler(response)
404 metadata = {"content_type": content_type, "url": url}
405 except Exception as e:
406 if self.raise_on_failure:
407 raise e
408 logger.debug("Couldn't retrieve content from {url} because {error}", url=url, error=str(e))
409 # Create an empty ByteStream for failed requests when raise_on_failure is False
410 stream = ByteStream(data=b"")
411 metadata = {"content_type": content_type, "url": url}
413 return metadata, stream
415 def _fetch_with_exception_suppression(self, url: str) -> tuple[dict[str, str] | None, ByteStream | None]:
416 """
417 Fetches content from a URL and returns it as a ByteStream.
419 If `raise_on_failure` is set to True, this method will wrap the fetch() method and catch any exceptions.
420 Otherwise, it will simply call the fetch() method.
421 :param url: The URL to fetch content from.
422 :returns: A tuple containing the ByteStream metadata dict and the corresponding ByteStream.
424 """
425 if self.raise_on_failure:
426 try:
427 return self._fetch(url)
428 except Exception as e:
429 logger.warning("Error fetching {url}: {error}", url=url, error=str(e))
430 return {"content_type": "Unknown", "url": url}, None
431 else:
432 return self._fetch(url)
434 async def _get_response_async(self, url: str, client: httpx.AsyncClient) -> httpx.Response:
435 """
436 Asynchronously gets a response from a URL with retry logic.
438 :param url: The URL to fetch.
439 :param client: The async httpx client to use for making requests.
440 :returns: The httpx Response object.
441 """
442 attempt = 0
443 last_exception = None
444 # Local to this call: `run_async` gathers URLs concurrently, so a cursor on the component
445 # would be shared by every request in flight.
446 user_agent_idx = 0
448 while attempt <= self.retry_attempts:
449 try:
450 response = await client.get(
451 url, headers=self._get_headers(client=client, user_agent=self.user_agents[user_agent_idx])
452 )
453 response.raise_for_status()
454 return response
455 except (httpx.HTTPStatusError, httpx.RequestError) as e:
456 last_exception = e
457 attempt += 1
458 if attempt <= self.retry_attempts:
459 # Switch user agent for next retry
460 user_agent_idx = (user_agent_idx + 1) % len(self.user_agents)
461 # Wait before retry using exponential backoff
462 await asyncio.sleep(min(2 * 2 ** (attempt - 1), 10))
463 else:
464 break
466 # If we've exhausted all retries, raise the last exception
467 if last_exception:
468 raise last_exception
470 # This should never happen, but just in case
471 raise httpx.RequestError("Failed to get response after retries", request=None)
473 def _get_content_type(self, response: httpx.Response) -> str:
474 """
475 Get the content type of the response.
477 :param response: The response object.
478 :returns: The content type of the response.
479 """
480 content_type = response.headers.get("Content-Type", "")
481 return content_type.split(";")[0]
483 def _resolve_handler(self, content_type: str) -> Callable[[httpx.Response], ByteStream]:
484 """
485 Resolves the handler for the given content type.
487 First, it tries to find a direct match for the content type in the handlers dictionary.
488 If no direct match is found, it tries to find a pattern match using the fnmatch function.
489 If no pattern match is found, it returns the default handler for text/plain.
491 :param content_type: The content type to resolve the handler for.
492 :returns: The handler for the given content type, if found. Otherwise, the default handler for text/plain.
493 """
494 # direct match
495 if content_type in self.handlers:
496 return self.handlers[content_type]
498 # pattern matches
499 for pattern, handler in self.handlers.items():
500 if fnmatch(content_type, pattern):
501 return handler
503 # default handler
504 return self.handlers["text/plain"]