Coverage for haystack/core/pipeline/utils.py: 99%
71 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 heapq
6from collections.abc import Callable
7from copy import deepcopy
8from functools import wraps
9from itertools import count
10from typing import Any
12from haystack import logging
13from haystack.core.component import Component
15logger = logging.getLogger(__name__)
18def _deepcopy_with_exceptions(obj: Any) -> Any:
19 """
20 Attempts to perform a deep copy of the given object.
22 This function recursively handles common container types (lists, tuples, sets, and dicts) to ensure deep copies
23 of nested structures. For specific object types that are known to be problematic for deepcopying-such as
24 instances of `Component`, `Tool`, or `Toolset` - the original object is returned as-is.
25 If `deepcopy` fails for any other reason, the original object is returned and a log message is recorded.
27 :param obj: The object to be deep-copied.
29 :returns:
30 A deep-copied version of the object, or the original object if deepcopying fails.
31 """
32 # Import here to avoid circular imports
33 from haystack.tools.tool import Tool
34 from haystack.tools.toolset import Toolset
36 # namedtuples are tuple subclasses whose __new__ takes the fields positionally,
37 # so they must be rebuilt by unpacking rather than from a single iterable.
38 if isinstance(obj, tuple) and hasattr(obj, "_fields"):
39 return type(obj)(*(_deepcopy_with_exceptions(v) for v in obj))
41 if isinstance(obj, (list, tuple, set)):
42 return type(obj)(_deepcopy_with_exceptions(v) for v in obj)
44 if isinstance(obj, dict):
45 return {k: _deepcopy_with_exceptions(v) for k, v in obj.items()}
47 # Components and Tools often contain objects that we do not want to deepcopy or are not deepcopyable
48 # (e.g. models, clients, etc.). In this case we return the object as-is.
49 if isinstance(obj, (Component, Tool, Toolset)):
50 return obj
52 try:
53 return deepcopy(obj)
54 except Exception as e:
55 logger.info(
56 "Deepcopy failed for object of type '{obj_type}'. Error: {error}. Returning original object instead.",
57 obj_type=type(obj).__name__,
58 error=e,
59 )
60 return obj
63def parse_connect_string(connection: str) -> tuple[str, str | None]:
64 """
65 Returns component-connection pairs from a connect_to/from string.
67 :param connection:
68 The connection string.
69 :returns:
70 A tuple containing the component name and the connection name.
71 """
72 if "." in connection:
73 split_str = connection.split(".", maxsplit=1)
74 return (split_str[0], split_str[1])
75 return connection, None
78class FIFOPriorityQueue:
79 """
80 A priority queue that maintains FIFO order for items of equal priority.
82 Items with the same priority are processed in the order they were added.
83 This queue ensures that when multiple items share the same priority level,
84 they are dequeued in the same order they were enqueued (First-In-First-Out).
85 """
87 def __init__(self) -> None:
88 """
89 Initialize a new FIFO priority queue.
90 """
91 # List of tuples (priority, count, item) where count ensures FIFO order
92 self._queue: list[tuple[int, int, Any]] = []
93 # Counter to maintain insertion order for equal priorities
94 self._counter = count()
96 def push(self, item: Any, priority: int) -> None:
97 """
98 Push an item into the queue with a given priority.
100 Items with equal priority maintain FIFO ordering based on insertion time.
101 Lower priority numbers are dequeued first.
103 :param item:
104 The item to insert into the queue.
105 :param priority:
106 Priority level for the item. Lower numbers indicate higher priority.
107 """
108 next_count = next(self._counter)
109 entry = (priority, next_count, item)
110 heapq.heappush(self._queue, entry)
112 def pop(self) -> tuple[int, Any]:
113 """
114 Remove and return the highest priority item from the queue.
116 For items with equal priority, returns the one that was inserted first.
118 :returns:
119 A tuple containing (priority, item) with the lowest priority number.
120 :raises IndexError:
121 If the queue is empty.
122 """
123 if not self._queue:
124 raise IndexError("pop from empty queue")
125 priority, _, item = heapq.heappop(self._queue)
126 return priority, item
128 def peek(self) -> tuple[int, Any]:
129 """
130 Return but don't remove the highest priority item from the queue.
132 For items with equal priority, returns the one that was inserted first.
134 :returns:
135 A tuple containing (priority, item) with the lowest priority number.
136 :raises IndexError:
137 If the queue is empty.
138 """
139 if not self._queue:
140 raise IndexError("peek at empty queue")
141 priority, _, item = self._queue[0]
142 return priority, item
144 def get(self) -> tuple[int, Any] | None:
145 """
146 Remove and return the highest priority item from the queue.
148 For items with equal priority, returns the one that was inserted first.
149 Unlike pop(), returns None if the queue is empty instead of raising an exception.
151 :returns:
152 A tuple containing (priority, item), or None if the queue is empty.
153 """
154 if not self._queue:
155 return None
156 priority, _, item = heapq.heappop(self._queue)
157 return priority, item
159 def __len__(self) -> int:
160 """
161 Return the number of items in the queue.
163 :returns:
164 The number of items currently in the queue.
165 """
166 return len(self._queue)
168 def __bool__(self) -> bool:
169 """
170 Return True if the queue has items, False if empty.
172 :returns:
173 True if the queue contains items, False otherwise.
174 """
175 return bool(self._queue)
178def args_deprecated(func: Callable[..., Any]) -> Callable[..., Any]:
179 """
180 Decorator to warn about the use of positional arguments in a function.
182 Adapted from https://stackoverflow.com/questions/68432070/
183 :param func:
184 """
186 def _positional_arg_warning() -> None:
187 """
188 Triggers a warning message if positional arguments are used in a function
189 """
190 import warnings
192 msg = (
193 "Warning: In an upcoming release, this method will require keyword arguments for all parameters. "
194 "Please update your code to use keyword arguments to ensure future compatibility. "
195 )
196 warnings.warn(msg, DeprecationWarning, stacklevel=2)
198 @wraps(func)
199 def wrapper(*args: Any, **kwargs: Any) -> Any:
200 # call the function first, to make sure the signature matches
201 ret_value = func(*args, **kwargs)
203 # A Pipeline instance is always the first argument - remove it from the args to check for positional arguments
204 # We check the class name as strings to avoid circular imports
205 if args and isinstance(args, tuple) and args[0].__class__.__name__ in ["Pipeline", "PipelineBase"]:
206 args = args[1:]
208 if args:
209 _positional_arg_warning()
210 return ret_value
212 return wrapper