(code: str, variables: dict[str, Any])
| 135 | |
| 136 | |
| 137 | def run_dify_code(code: str, variables: dict[str, Any]) -> dict[str, Any]: |
| 138 | namespace: dict[str, Any] = {} |
| 139 | try: |
| 140 | exec(textwrap.dedent(code), {"__builtins__": __builtins__}, namespace) # noqa: S102 |
| 141 | except ModuleNotFoundError as exc: |
| 142 | name = getattr(exc, "name", None) or str(exc) |
| 143 | raise CompatibilityImportError( |
| 144 | f"Dify code node requires Python package {name!r}. " |
| 145 | "Install extras with: uv sync --extra dify-preview" |
| 146 | ) from exc |
| 147 | main = namespace.get("main") |
| 148 | if not callable(main): |
| 149 | raise CompatibilityImportError("Dify code node must define a callable `main(...)`.") |
| 150 | result = main(**variables) |
| 151 | if not isinstance(result, dict): |
| 152 | raise CompatibilityImportError("Dify code node `main` must return a dict.") |
| 153 | return result |
| 154 | |
| 155 | |
| 156 | def init_dify_state(metadata: dict[str, Any] | None, outer: dict[str, Any]) -> DifyRuntimeState: |
no test coverage detected