Coverage for haystack/core/super_component/super_component.py: 96%

194 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 

4 

5import functools 

6from pathlib import Path 

7from types import new_class 

8from typing import Any, TypeVar 

9 

10from haystack import logging 

11from haystack.core.component.component import component 

12from haystack.core.pipeline.pipeline import Pipeline 

13from haystack.core.pipeline.utils import parse_connect_string 

14from haystack.core.serialization import default_from_dict, default_to_dict, generate_qualified_class_name 

15from haystack.core.super_component.utils import _delegate_default, _is_compatible 

16 

17logger = logging.getLogger(__name__) 

18 

19T = TypeVar("T") 

20 

21 

22class InvalidMappingTypeError(Exception): 

23 """Raised when input or output mappings have invalid types or type conflicts.""" 

24 

25 pass 

26 

27 

28class InvalidMappingValueError(Exception): 

29 """Raised when input or output mappings have invalid values or missing components/sockets.""" 

30 

31 pass 

32 

33 

34@component 

35class _SuperComponent: 

36 def __init__( 

37 self, 

38 pipeline: Pipeline, 

39 input_mapping: dict[str, list[str]] | None = None, 

40 output_mapping: dict[str, str] | None = None, 

41 ) -> None: 

42 """ 

43 Creates a SuperComponent with optional input and output mappings. 

44 

45 :param pipeline: The pipeline instance to be wrapped 

46 :param input_mapping: A dictionary mapping component input names to pipeline input socket paths. 

47 If not provided, a default input mapping will be created based on all pipeline inputs. 

48 Example: 

49 ```python 

50 input_mapping={ 

51 "query": ["retriever.query", "prompt_builder.query"], 

52 } 

53 ``` 

54 :param output_mapping: A dictionary mapping pipeline output socket paths to component output names. 

55 If not provided, a default output mapping will be created based on all pipeline outputs. 

56 Example: 

57 ```python 

58 output_mapping={ 

59 "retriever.documents": "documents", 

60 "generator.replies": "replies", 

61 } 

62 ``` 

63 :raises InvalidMappingError: Raised if any mapping is invalid or type conflicts occur 

64 :raises ValueError: Raised if no pipeline is provided 

65 """ 

66 if pipeline is None: 

67 raise ValueError("Pipeline must be provided to SuperComponent.") 

68 

69 self.pipeline: Pipeline = pipeline 

70 

71 # Determine input types based on pipeline and mapping 

72 pipeline_inputs = self.pipeline.inputs() 

73 resolved_input_mapping = ( 

74 input_mapping if input_mapping is not None else self._create_input_mapping(pipeline_inputs) 

75 ) 

76 self._validate_input_mapping(pipeline_inputs, resolved_input_mapping) 

77 input_types = self._resolve_input_types_from_mapping(pipeline_inputs, resolved_input_mapping) 

78 # Set input types on the component 

79 for input_name, info in input_types.items(): 

80 component.set_input_type(self, name=input_name, **info) 

81 

82 self.input_mapping: dict[str, list[str]] = resolved_input_mapping 

83 self._original_input_mapping = input_mapping 

84 

85 # Set output types based on pipeline and mapping 

86 leaf_pipeline_outputs = self.pipeline.outputs() 

87 all_possible_pipeline_outputs = self.pipeline.outputs(include_components_with_connected_outputs=True) 

88 

89 resolved_output_mapping = ( 

90 output_mapping if output_mapping is not None else self._create_output_mapping(leaf_pipeline_outputs) 

91 ) 

92 self._validate_output_mapping(all_possible_pipeline_outputs, resolved_output_mapping) 

93 output_types = self._resolve_output_types_from_mapping(all_possible_pipeline_outputs, resolved_output_mapping) 

94 # Set output types on the component 

95 component.set_output_types(self, **output_types) 

96 self.output_mapping: dict[str, str] = resolved_output_mapping 

97 self._original_output_mapping = output_mapping 

98 

99 def warm_up(self) -> None: 

100 """ 

101 Warms up the SuperComponent by warming up the wrapped pipeline. 

102 """ 

103 self.pipeline.warm_up() 

104 

105 async def warm_up_async(self) -> None: 

106 """ 

107 Warms up the SuperComponent by warming up the wrapped pipeline on the serving event loop. 

108 """ 

109 await self.pipeline.warm_up_async() 

110 

111 def close(self) -> None: 

112 """ 

113 Releases the synchronous resources held by the wrapped pipeline's components. 

114 """ 

115 self.pipeline.close() 

116 

117 async def close_async(self) -> None: 

118 """ 

119 Releases the async resources held by the wrapped pipeline's components. 

120 """ 

121 await self.pipeline.close_async() 

122 

123 def run(self, **kwargs: Any) -> dict[str, Any]: 

124 """ 

125 Runs the wrapped pipeline with the provided inputs. 

126 

127 Steps: 

128 1. Maps the inputs from kwargs to pipeline component inputs 

129 2. Runs the pipeline 

130 3. Maps the pipeline outputs to the SuperComponent's outputs 

131 

132 :param kwargs: Keyword arguments matching the SuperComponent's input names 

133 :returns: 

134 Dictionary containing the SuperComponent's output values 

135 """ 

136 # `is not`, not `!=`: numpy/pandas/torch override `__ne__` element-wise and would crash here. 

137 filtered_inputs = {param: value for param, value in kwargs.items() if value is not _delegate_default} 

138 pipeline_inputs = self._map_explicit_inputs(input_mapping=self.input_mapping, inputs=filtered_inputs) 

139 include_outputs_from = self._get_include_outputs_from() 

140 pipeline_outputs = self.pipeline.run(data=pipeline_inputs, include_outputs_from=include_outputs_from) 

141 return self._map_explicit_outputs(pipeline_outputs, self.output_mapping) 

142 

143 def _get_include_outputs_from(self) -> set[str]: 

144 # Collecting the component names from output_mapping 

145 return {self._split_component_path(path)[0] for path in self.output_mapping.keys()} 

146 

147 async def run_async(self, **kwargs: Any) -> dict[str, Any]: 

148 """ 

149 Runs the wrapped pipeline with the provided inputs async. 

150 

151 Steps: 

152 1. Maps the inputs from kwargs to pipeline component inputs 

153 2. Runs the pipeline async 

154 3. Maps the pipeline outputs to the SuperComponent's outputs 

155 

156 :param kwargs: Keyword arguments matching the SuperComponent's input names 

157 :returns: 

158 Dictionary containing the SuperComponent's output values 

159 """ 

160 # `is not`, not `!=`: numpy/pandas/torch override `__ne__` element-wise and would crash here. 

161 filtered_inputs = {param: value for param, value in kwargs.items() if value is not _delegate_default} 

162 pipeline_inputs = self._map_explicit_inputs(input_mapping=self.input_mapping, inputs=filtered_inputs) 

163 pipeline_outputs = await self.pipeline.run_async(data=pipeline_inputs) 

164 return self._map_explicit_outputs(pipeline_outputs, self.output_mapping) 

165 

166 @staticmethod 

167 def _split_component_path(path: str) -> tuple[str, str]: 

168 """ 

169 Splits a component path into a component name and a socket name. 

170 

171 :param path: A string in the format "component_name.socket_name". 

172 :returns: 

173 A tuple containing (component_name, socket_name). 

174 :raises InvalidMappingValueError: 

175 If the path format is incorrect. 

176 """ 

177 comp_name, socket_name = parse_connect_string(path) 

178 if socket_name is None: 

179 raise InvalidMappingValueError(f"Invalid path format: '{path}'. Expected 'component_name.socket_name'.") 

180 return comp_name, socket_name 

181 

182 def _validate_input_mapping( 

183 self, pipeline_inputs: dict[str, dict[str, Any]], input_mapping: dict[str, list[str]] 

184 ) -> None: 

185 """ 

186 Validates the input mapping to ensure that specified components and sockets exist in the pipeline. 

187 

188 :param pipeline_inputs: A dictionary containing pipeline input specifications. 

189 :param input_mapping: A dictionary mapping wrapper input names to pipeline socket paths. 

190 :raises InvalidMappingTypeError: 

191 If the input mapping is of invalid type or contains invalid types. 

192 :raises InvalidMappingValueError: 

193 If the input mapping contains nonexistent components or sockets. 

194 """ 

195 if not isinstance(input_mapping, dict): 

196 raise InvalidMappingTypeError("input_mapping must be a dictionary") 

197 

198 for wrapper_input_name, pipeline_input_paths in input_mapping.items(): 

199 if not isinstance(pipeline_input_paths, list): 

200 raise InvalidMappingTypeError(f"Input paths for '{wrapper_input_name}' must be a list of strings.") 

201 for path in pipeline_input_paths: 

202 comp_name, socket_name = self._split_component_path(path) 

203 if comp_name not in pipeline_inputs: 

204 raise InvalidMappingValueError( 

205 f"Component '{comp_name}' not found in pipeline inputs.\n" 

206 f"Available components: {list(pipeline_inputs.keys())}" 

207 ) 

208 if socket_name not in pipeline_inputs[comp_name]: 

209 raise InvalidMappingValueError( 

210 f"Input socket '{socket_name}' not found in component '{comp_name}'.\n" 

211 f"Available inputs for '{comp_name}': {list(pipeline_inputs[comp_name].keys())}" 

212 ) 

213 

214 def _resolve_input_types_from_mapping( 

215 self, pipeline_inputs: dict[str, dict[str, Any]], input_mapping: dict[str, list[str]] 

216 ) -> dict[str, dict[str, Any]]: 

217 """ 

218 Resolves and validates input types based on the provided input mapping. 

219 

220 This function ensures that all mapped pipeline inputs are compatible, consolidating types 

221 when multiple mappings exist. It also determines whether an input is mandatory or has a default value. 

222 

223 :param pipeline_inputs: A dictionary containing pipeline input specifications. 

224 :param input_mapping: A dictionary mapping SuperComponent inputs to pipeline socket paths. 

225 :returns: 

226 A dictionary specifying the resolved input types and their properties. 

227 :raises InvalidMappingTypeError: 

228 If the input mapping contains incompatible types. 

229 """ 

230 aggregated_inputs: dict[str, dict[str, Any]] = {} 

231 for wrapper_input_name, pipeline_input_paths in input_mapping.items(): 

232 for path in pipeline_input_paths: 

233 comp_name, socket_name = self._split_component_path(path) 

234 socket_info = pipeline_inputs[comp_name][socket_name] 

235 

236 # Add to aggregated inputs 

237 existing_socket_info = aggregated_inputs.get(wrapper_input_name) 

238 if existing_socket_info is None: 

239 aggregated_inputs[wrapper_input_name] = {"type": socket_info["type"]} 

240 if not socket_info["is_mandatory"]: 

241 aggregated_inputs[wrapper_input_name]["default"] = _delegate_default 

242 continue 

243 

244 is_compatible, common_type = _is_compatible(existing_socket_info["type"], socket_info["type"]) 

245 

246 if not is_compatible: 

247 raise InvalidMappingTypeError( 

248 f"Type conflict for input '{socket_name}' from component '{comp_name}'. " 

249 f"Existing type: {existing_socket_info['type']}, new type: {socket_info['type']}." 

250 ) 

251 

252 # Use the common type for the aggregated input 

253 aggregated_inputs[wrapper_input_name]["type"] = common_type 

254 

255 # If any socket requires mandatory inputs then the aggregated input is also considered mandatory. 

256 # So we use the type of the mandatory input and remove the default value if it exists. 

257 if socket_info["is_mandatory"]: 

258 aggregated_inputs[wrapper_input_name].pop("default", None) 

259 

260 return aggregated_inputs 

261 

262 @staticmethod 

263 def _create_input_mapping(pipeline_inputs: dict[str, dict[str, Any]]) -> dict[str, list[str]]: 

264 """ 

265 Create an input mapping from pipeline inputs. 

266 

267 :param pipeline_inputs: Dictionary of pipeline input specifications 

268 :returns: 

269 Dictionary mapping SuperComponent input names to pipeline socket paths 

270 """ 

271 input_mapping: dict[str, list[str]] = {} 

272 for comp_name, inputs_dict in pipeline_inputs.items(): 

273 for socket_name in inputs_dict.keys(): 

274 existing_socket_info = input_mapping.get(socket_name) 

275 if existing_socket_info is None: 

276 input_mapping[socket_name] = [f"{comp_name}.{socket_name}"] 

277 continue 

278 input_mapping[socket_name].append(f"{comp_name}.{socket_name}") 

279 return input_mapping 

280 

281 def _validate_output_mapping( 

282 self, pipeline_outputs: dict[str, dict[str, Any]], output_mapping: dict[str, str] 

283 ) -> None: 

284 """ 

285 Validates the output mapping to ensure that specified components and sockets exist in the pipeline. 

286 

287 :param pipeline_outputs: A dictionary containing pipeline output specifications. 

288 :param output_mapping: A dictionary mapping pipeline socket paths to wrapper output names. 

289 :raises InvalidMappingTypeError: 

290 If the output mapping is of invalid type or contains invalid types. 

291 :raises InvalidMappingValueError: 

292 If the output mapping contains nonexistent components or sockets. 

293 """ 

294 for pipeline_output_path, wrapper_output_name in output_mapping.items(): 

295 if not isinstance(wrapper_output_name, str): 

296 raise InvalidMappingTypeError("Output names in output_mapping must be strings.") 

297 comp_name, socket_name = self._split_component_path(pipeline_output_path) 

298 if comp_name not in pipeline_outputs: 

299 raise InvalidMappingValueError(f"Component '{comp_name}' not found among pipeline outputs.") 

300 if socket_name not in pipeline_outputs[comp_name]: 

301 raise InvalidMappingValueError(f"Output socket '{socket_name}' not found in component '{comp_name}'.") 

302 

303 def _resolve_output_types_from_mapping( 

304 self, pipeline_outputs: dict[str, dict[str, Any]], output_mapping: dict[str, str] 

305 ) -> dict[str, Any]: 

306 """ 

307 Resolves and validates output types based on the provided output mapping. 

308 

309 This function ensures that all mapped pipeline outputs are correctly assigned to 

310 the corresponding SuperComponent outputs while preventing duplicate output names. 

311 

312 :param pipeline_outputs: A dictionary containing pipeline output specifications. 

313 :param output_mapping: A dictionary mapping pipeline output socket paths to SuperComponent output names. 

314 :returns: 

315 A dictionary mapping SuperComponent output names to their resolved types. 

316 :raises InvalidMappingValueError: 

317 If the output mapping contains duplicate output names. 

318 """ 

319 resolved_outputs = {} 

320 for pipeline_output_path, wrapper_output_name in output_mapping.items(): 

321 comp_name, socket_name = self._split_component_path(pipeline_output_path) 

322 if wrapper_output_name in resolved_outputs: 

323 raise InvalidMappingValueError(f"Duplicate output name '{wrapper_output_name}' in output_mapping.") 

324 resolved_outputs[wrapper_output_name] = pipeline_outputs[comp_name][socket_name]["type"] 

325 return resolved_outputs 

326 

327 @staticmethod 

328 def _create_output_mapping(pipeline_outputs: dict[str, dict[str, Any]]) -> dict[str, str]: 

329 """ 

330 Create an output mapping from pipeline outputs. 

331 

332 :param pipeline_outputs: Dictionary of pipeline output specifications 

333 :returns: 

334 Dictionary mapping pipeline socket paths to SuperComponent output names 

335 :raises InvalidMappingValueError: 

336 If there are output name conflicts between components 

337 """ 

338 output_mapping = {} 

339 used_output_names: set[str] = set() 

340 for comp_name, outputs_dict in pipeline_outputs.items(): 

341 for socket_name in outputs_dict.keys(): 

342 if socket_name in used_output_names: 

343 raise InvalidMappingValueError( 

344 f"Output name conflict: '{socket_name}' is produced by multiple components. " 

345 "Please provide an output_mapping to resolve this conflict." 

346 ) 

347 used_output_names.add(socket_name) 

348 output_mapping[f"{comp_name}.{socket_name}"] = socket_name 

349 return output_mapping 

350 

351 def _map_explicit_inputs( 

352 self, input_mapping: dict[str, list[str]], inputs: dict[str, Any] 

353 ) -> dict[str, dict[str, Any]]: 

354 """ 

355 Map inputs according to explicit input mapping. 

356 

357 :param input_mapping: Mapping configuration for inputs 

358 :param inputs: Input arguments provided to wrapper 

359 :return: Dictionary of mapped pipeline inputs 

360 """ 

361 pipeline_inputs: dict[str, dict[str, Any]] = {} 

362 for wrapper_input_name, pipeline_input_paths in input_mapping.items(): 

363 if wrapper_input_name not in inputs: 

364 continue 

365 

366 for socket_path in pipeline_input_paths: 

367 comp_name, input_name = self._split_component_path(socket_path) 

368 if comp_name not in pipeline_inputs: 

369 pipeline_inputs[comp_name] = {} 

370 pipeline_inputs[comp_name][input_name] = inputs[wrapper_input_name] 

371 

372 return pipeline_inputs 

373 

374 def _map_explicit_outputs( 

375 self, pipeline_outputs: dict[str, dict[str, Any]], output_mapping: dict[str, str] 

376 ) -> dict[str, Any]: 

377 """ 

378 Map outputs according to explicit output mapping. 

379 

380 :param pipeline_outputs: Raw outputs from pipeline execution 

381 :param output_mapping: Output mapping configuration 

382 :return: Dictionary of mapped outputs 

383 """ 

384 outputs: dict[str, Any] = {} 

385 for pipeline_output_path, wrapper_output_name in output_mapping.items(): 

386 comp_name, socket_name = self._split_component_path(pipeline_output_path) 

387 if comp_name in pipeline_outputs and socket_name in pipeline_outputs[comp_name]: 

388 outputs[wrapper_output_name] = pipeline_outputs[comp_name][socket_name] 

389 return outputs 

390 

391 def _to_super_component_dict(self) -> dict[str, Any]: 

392 """ 

393 Convert to a SuperComponent dictionary representation. 

394 

395 :return: Dictionary containing serialized SuperComponent data 

396 """ 

397 serialized_pipeline = self.pipeline.to_dict() 

398 serialized = default_to_dict( 

399 self, 

400 pipeline=serialized_pipeline, 

401 input_mapping=self._original_input_mapping, 

402 output_mapping=self._original_output_mapping, 

403 ) 

404 serialized["type"] = generate_qualified_class_name(SuperComponent) 

405 return serialized 

406 

407 

408@component 

409class SuperComponent(_SuperComponent): 

410 """ 

411 A class for creating super components that wrap around a Pipeline. 

412 

413 This component allows for remapping of input and output socket names between the wrapped pipeline and the 

414 SuperComponent's input and output names. This is useful for creating higher-level components that abstract 

415 away the details of the wrapped pipeline. 

416 

417 ### Usage example 

418 

419 ```python 

420 from haystack import Pipeline, SuperComponent 

421 from haystack.components.generators.chat import OpenAIChatGenerator 

422 from haystack.components.builders import ChatPromptBuilder 

423 from haystack.components.retrievers import InMemoryBM25Retriever 

424 from haystack.dataclasses.chat_message import ChatMessage 

425 from haystack.document_stores.in_memory import InMemoryDocumentStore 

426 from haystack.dataclasses import Document 

427 

428 document_store = InMemoryDocumentStore() 

429 documents = [ 

430 Document(content="Paris is the capital of France."), 

431 Document(content="London is the capital of England."), 

432 ] 

433 document_store.write_documents(documents) 

434 

435 prompt_template = [ 

436 ChatMessage.from_user( 

437 ''' 

438 According to the following documents: 

439 {% for document in documents %} 

440 {{document.content}} 

441 {% endfor %} 

442 Answer the given question: {{query}} 

443 Answer: 

444 ''' 

445 ) 

446 ] 

447 

448 prompt_builder = ChatPromptBuilder(template=prompt_template, required_variables="*") 

449 

450 pipeline = Pipeline() 

451 pipeline.add_component("retriever", InMemoryBM25Retriever(document_store=document_store)) 

452 pipeline.add_component("prompt_builder", prompt_builder) 

453 pipeline.add_component("llm", OpenAIChatGenerator()) 

454 pipeline.connect("retriever.documents", "prompt_builder.documents") 

455 pipeline.connect("prompt_builder.prompt", "llm.messages") 

456 

457 # Create a super component with simplified input/output mapping 

458 wrapper = SuperComponent( 

459 pipeline=pipeline, 

460 input_mapping={ 

461 "query": ["retriever.query", "prompt_builder.query"], 

462 }, 

463 output_mapping={"llm.replies": "replies"} 

464 ) 

465 

466 # Run the pipeline with simplified interface 

467 result = wrapper.run(query="What is the capital of France?") 

468 print(result) 

469 {'replies': [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>, 

470 _content=[TextContent(text='The capital of France is Paris.')],...) 

471 ``` 

472 

473 """ 

474 

475 def to_dict(self) -> dict[str, Any]: 

476 """ 

477 Serializes the SuperComponent into a dictionary. 

478 

479 :returns: 

480 Dictionary with serialized data. 

481 """ 

482 return self._to_super_component_dict() 

483 

484 @classmethod 

485 def from_dict(cls, data: dict[str, Any]) -> "SuperComponent": 

486 """ 

487 Deserializes the SuperComponent from a dictionary. 

488 

489 :param data: The dictionary to deserialize from. 

490 :returns: 

491 The deserialized SuperComponent. 

492 """ 

493 # `is_pipeline_async` is a legacy key kept only for backward compatibility. 

494 data["init_parameters"].pop("is_pipeline_async", None) 

495 pipeline = Pipeline.from_dict(data["init_parameters"]["pipeline"]) 

496 data["init_parameters"]["pipeline"] = pipeline 

497 return default_from_dict(cls, data) 

498 

499 def show(self, server_url: str = "https://mermaid.ink", params: dict | None = None, timeout: int = 30) -> None: 

500 """ 

501 Display an image representing this SuperComponent's underlying pipeline in a Jupyter notebook. 

502 

503 This function generates a diagram of the Pipeline using a Mermaid server and displays it directly in 

504 the notebook. 

505 

506 :param server_url: 

507 The base URL of the Mermaid server used for rendering (default: 'https://mermaid.ink'). 

508 See https://github.com/jihchi/mermaid.ink and https://github.com/mermaid-js/mermaid-live-editor for more 

509 info on how to set up your own Mermaid server. 

510 

511 :param params: 

512 Dictionary of customization parameters to modify the output. Refer to Mermaid documentation for more details 

513 Supported keys: 

514 - format: Output format ('img', 'svg', or 'pdf'). Default: 'img'. 

515 - type: Image type for /img endpoint ('jpeg', 'png', 'webp'). Default: 'png'. 

516 - theme: Mermaid theme ('default', 'neutral', 'dark', 'forest'). Default: 'neutral'. 

517 - bgColor: Background color in hexadecimal (e.g., 'FFFFFF') or named format (e.g., '!white'). 

518 - width: Width of the output image (integer). 

519 - height: Height of the output image (integer). 

520 - scale: Scaling factor (1–3). Only applicable if 'width' or 'height' is specified. 

521 - fit: Whether to fit the diagram size to the page (PDF only, boolean). 

522 - paper: Paper size for PDFs (e.g., 'a4', 'a3'). Ignored if 'fit' is true. 

523 - landscape: Landscape orientation for PDFs (boolean). Ignored if 'fit' is true. 

524 

525 :param timeout: 

526 Timeout in seconds for the request to the Mermaid server. 

527 

528 :raises PipelineDrawingError: 

529 If the function is called outside of a Jupyter notebook or if there is an issue with rendering. 

530 """ 

531 self.pipeline.show(server_url=server_url, params=params, timeout=timeout) 

532 

533 def draw( 

534 self, path: Path, server_url: str = "https://mermaid.ink", params: dict | None = None, timeout: int = 30 

535 ) -> None: 

536 """ 

537 Save an image representing this SuperComponent's underlying pipeline to the specified file path. 

538 

539 This function generates a diagram of the Pipeline using the Mermaid server and saves it to the provided path. 

540 

541 :param path: 

542 The file path where the generated image will be saved. 

543 :param server_url: 

544 The base URL of the Mermaid server used for rendering (default: 'https://mermaid.ink'). 

545 See https://github.com/jihchi/mermaid.ink and https://github.com/mermaid-js/mermaid-live-editor for more 

546 info on how to set up your own Mermaid server. 

547 :param params: 

548 Dictionary of customization parameters to modify the output. Refer to Mermaid documentation for more details 

549 Supported keys: 

550 - format: Output format ('img', 'svg', or 'pdf'). Default: 'img'. 

551 - type: Image type for /img endpoint ('jpeg', 'png', 'webp'). Default: 'png'. 

552 - theme: Mermaid theme ('default', 'neutral', 'dark', 'forest'). Default: 'neutral'. 

553 - bgColor: Background color in hexadecimal (e.g., 'FFFFFF') or named format (e.g., '!white'). 

554 - width: Width of the output image (integer). 

555 - height: Height of the output image (integer). 

556 - scale: Scaling factor (1–3). Only applicable if 'width' or 'height' is specified. 

557 - fit: Whether to fit the diagram size to the page (PDF only, boolean). 

558 - paper: Paper size for PDFs (e.g., 'a4', 'a3'). Ignored if 'fit' is true. 

559 - landscape: Landscape orientation for PDFs (boolean). Ignored if 'fit' is true. 

560 

561 :param timeout: 

562 Timeout in seconds for the request to the Mermaid server. 

563 

564 :raises PipelineDrawingError: 

565 If there is an issue with rendering or saving the image. 

566 """ 

567 self.pipeline.draw(path=path, server_url=server_url, params=params, timeout=timeout) 

568 

569 

570def super_component(cls: type[T]) -> type[T]: 

571 """ 

572 Decorator that converts a class into a SuperComponent. 

573 

574 This decorator: 

575 1. Creates a new class that inherits from SuperComponent 

576 2. Copies all methods and attributes from the original class 

577 3. Adds initialization logic to properly set up the SuperComponent 

578 

579 The decorated class should define: 

580 - pipeline: A Pipeline instance in the __init__ method 

581 - input_mapping: Dictionary mapping component inputs to pipeline inputs (optional) 

582 - output_mapping: Dictionary mapping pipeline outputs to component outputs (optional) 

583 """ 

584 logger.debug("Registering {cls} as a super_component", cls=cls) 

585 

586 # Store the original __init__ method 

587 original_init = cls.__init__ 

588 

589 # Create a new __init__ method that will initialize both the original class and SuperComponent 

590 def init_wrapper(self: Any, *args: Any, **kwargs: Any) -> None: 

591 # Call the original __init__ to set up pipeline and mappings 

592 original_init(self, *args, **kwargs) 

593 

594 # Verify required attributes 

595 if not hasattr(self, "pipeline"): 

596 raise ValueError(f"Class {cls.__name__} decorated with @super_component must define a 'pipeline' attribute") 

597 

598 # Initialize SuperComponent 

599 _SuperComponent.__init__( 

600 self, 

601 pipeline=self.pipeline, 

602 input_mapping=getattr(self, "input_mapping", None), 

603 output_mapping=getattr(self, "output_mapping", None), 

604 ) 

605 

606 # Preserve original init's signature for IDEs/docs/tools 

607 init_wrapper = functools.wraps(original_init)(init_wrapper) 

608 

609 # Function to copy namespace from the original class 

610 def copy_class_namespace(namespace: dict[str, Any]) -> None: 

611 """Copy all attributes from the original class except special ones.""" 

612 for key, val in dict(cls.__dict__).items(): 

613 # Skip special attributes that should be recreated 

614 if key in ("__dict__", "__weakref__"): 

615 continue 

616 

617 # Override __init__ with our wrapper 

618 if key == "__init__": 

619 namespace["__init__"] = init_wrapper 

620 continue 

621 

622 namespace[key] = val 

623 

624 # Create a new class inheriting from SuperComponent with the original methods 

625 # We use (SuperComponent,) + cls.__bases__ to make the new class inherit from 

626 # SuperComponent and all the original class's bases 

627 new_cls = new_class(cls.__name__, (_SuperComponent,) + cls.__bases__, {}, copy_class_namespace) 

628 

629 # Copy other class attributes 

630 new_cls.__module__ = cls.__module__ 

631 new_cls.__qualname__ = cls.__qualname__ 

632 new_cls.__doc__ = cls.__doc__ 

633 

634 # Apply the component decorator to the new class 

635 return component(new_cls)