Execute the root graph. Args: input: Graph input payload. attributes: Optional attribute overrides for this invocation. Returns: tuple[dict, dict[str, object]]: `(output, attributes_snapshot)`.
(self, input: dict, attributes: dict[str, object] | None = None)
| 28 | self._output = {} |
| 29 | |
| 30 | def invoke(self, input: dict, attributes: dict[str, object] | None = None): |
| 31 | """ |
| 32 | Execute the root graph. |
| 33 | |
| 34 | Args: |
| 35 | input: Graph input payload. |
| 36 | attributes: Optional attribute overrides for this invocation. |
| 37 | |
| 38 | Returns: |
| 39 | tuple[dict, dict[str, object]]: `(output, attributes_snapshot)`. |
| 40 | """ |
| 41 | if not self.check_built(): |
| 42 | raise RuntimeError("Graph is not built yet. Please build the graph first.") |
| 43 | if attributes is None: |
| 44 | attributes = {} |
| 45 | self._attributes_store = {**self._attributes_store, **attributes} |
| 46 | try: |
| 47 | from masfactory.visualizer import get_bridge # noqa: WPS433 |
| 48 | except Exception: |
| 49 | get_bridge = None # type: ignore[assignment] |
| 50 | runtime = get_bridge() if get_bridge is not None else None |
| 51 | if runtime is not None: |
| 52 | runtime.begin_run(self, input=input) |
| 53 | self._input = input |
| 54 | self.execute(self.attributes) |
| 55 | if runtime is not None: |
| 56 | runtime.end_run(self, output=self._output if isinstance(self._output, dict) else None) |
| 57 | return self._output,self.attributes.copy() |
| 58 | |
| 59 | def build(self): |
| 60 | if self._is_built: |
no test coverage detected