Coverage for haystack/components/converters/utils.py: 100%

1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai> 

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5from copy import deepcopy 

6from enum import Enum 

7from pathlib import Path 

8from typing import Any 

9 

10from haystack.dataclasses import ByteStream 

11 

12 

13class LinkFormat(Enum): 

14 """ 

15 Supported formats for storing link information in a Document. 

16 """ 

17 

18 MARKDOWN = "markdown" 

19 PLAIN = "plain" 

20 NONE = "none" 

21 

22 def __str__(self) -> str: 

23 return self.value 

24 

25 @staticmethod 

26 def from_str(string: str) -> "LinkFormat": 

27 """ 

28 Convert a string to a LinkFormat enum. 

29 """ 

30 enum_map = {e.value: e for e in LinkFormat} 

31 link_format = enum_map.get(string.lower()) 

32 if link_format is None: 

33 msg = f"Unknown link format '{string}'. Supported formats are: {list(enum_map.keys())}" 

34 raise ValueError(msg) 

35 return link_format 

36 

37 

38def get_bytestream_from_source(source: str | Path | ByteStream, guess_mime_type: bool = False) -> ByteStream: 

39 """ 

40 Creates a ByteStream object from a source. 

41 

42 :param source: 

43 A source to convert to a ByteStream. Can be a string (path to a file), a Path object, or a ByteStream. 

44 :param guess_mime_type: 

45 Whether to guess the mime type from the file. 

46 :return: 

47 A ByteStream object. 

48 """ 

49 

50 if isinstance(source, ByteStream): 

51 return source 

52 if isinstance(source, (str, Path)): 

53 bs = ByteStream.from_file_path(Path(source), guess_mime_type=guess_mime_type) 

54 bs.meta["file_path"] = str(source) 

55 return bs 

56 raise ValueError(f"Unsupported source type {type(source)}") 

57 

58 

59def normalize_metadata(meta: dict[str, Any] | list[dict[str, Any]] | None, sources_count: int) -> list[dict[str, Any]]: 

60 """ 

61 Normalize the metadata input for a converter. 

62 

63 Given all the possible value of the meta input for a converter (None, dictionary or list of dicts), 

64 makes sure to return a list of dictionaries of the correct length for the converter to use. 

65 

66 :param meta: the meta input of the converter, as-is 

67 :param sources_count: the number of sources the converter received 

68 :returns: a list of dictionaries of the make length as the sources list 

69 

70 Each source always gets its own independent dictionary. When ``meta`` is ``None`` or a single 

71 dictionary, a separate copy is returned for every source so that mutating one source's metadata 

72 downstream does not leak into the others. 

73 """ 

74 if meta is None: 

75 return [{} for _ in range(sources_count)] 

76 if isinstance(meta, dict): 

77 return [deepcopy(meta) for _ in range(sources_count)] 

78 if isinstance(meta, list): 

79 if sources_count != len(meta): 

80 raise ValueError("The length of the metadata list must match the number of sources.") 

81 return meta 

82 raise ValueError("meta must be either None, a dictionary or a list of dictionaries.")