This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 59ada4ec02 [Improve][Core] Support transform and value constraint
option rules in CLI (#11109)
59ada4ec02 is described below
commit 59ada4ec02f821d97491507c20ab3575be4aa86b
Author: yzeng1618 <[email protected]>
AuthorDate: Fri Jun 19 18:25:52 2026 +0800
[Improve][Core] Support transform and value constraint option rules in CLI
(#11109)
Co-authored-by: zengyi <[email protected]>
---
seatunnel-cli/README.md | 9 +-
seatunnel-cli/README.zh-CN.md | 9 +-
seatunnel-cli/seatunnel_cli/agents.py | 13 +-
seatunnel-cli/seatunnel_cli/cli.py | 3 +-
seatunnel-cli/seatunnel_cli/connectors.py | 99 +++++++-
seatunnel-cli/seatunnel_cli/skills.py | 32 ++-
seatunnel-cli/tests/test_connector_metadata.py | 304 +++++++++++++++++++++++++
7 files changed, 446 insertions(+), 23 deletions(-)
diff --git a/seatunnel-cli/README.md b/seatunnel-cli/README.md
index f268e3ff12..17f81c02af 100644
--- a/seatunnel-cli/README.md
+++ b/seatunnel-cli/README.md
@@ -10,6 +10,7 @@ Describe your data synchronization task in English or
Chinese, and the CLI gener
- **Multi-Provider LLM** -- AWS Bedrock, Anthropic API, OpenAI (and compatible
APIs like Azure OpenAI)
- **Multi-Agent Pipeline** -- Planner -> Generator -> Validator -> Auto-fix,
up to 3 correction rounds
- **100+ Connectors** -- Full coverage of SeaTunnel's connector ecosystem with
runtime metadata reflection
+- **Transform Metadata** -- Source, sink, and transform plugins use full
option rules and value constraints during generation
- **Skill Framework** -- Three-layer generation: Skill SOP -> Golden Example
-> Connector Metadata
- **Auto-Save** -- Generated configs automatically saved to
`.data/last_job.conf` (co-located with CLI)
- **Auto-Fix** -- `/check` and `/run` failures trigger automatic LLM-powered
diagnosis and config repair
@@ -241,7 +242,7 @@ Options:
| `/save <path>` | Save config to custom path (auto-saved to
`.data/last_job.conf` on generation) |
| `/check` | Dry-run validate last config; auto-diagnoses and fixes on failure
|
| `/run` | Execute last config via REST API or `seatunnel.sh`; auto-diagnoses
on failure |
-| `/connectors` | List all available sources, sinks, and transforms |
+| `/connectors` | List all available sources, sinks, and transforms; transform
option rules and constraints are supported during generation |
| `/sessions` | List recent conversation sessions |
| `/resume [id]` | Resume a previous session |
| `/new` | Start a fresh session |
@@ -371,7 +372,9 @@ User Input (natural language)
Two-tier resolution with intelligent fallback:
1. **Runtime API** -- Live metadata from running SeaTunnel engine
(`/option-rules` endpoint). Always accurate, zero maintenance.
-2. **Bundled Metadata** -- `connector_metadata.json` ships with the CLI
package. 150 connectors with full option rules, exported from SeaTunnel engine
via reflection. Zero LLM token cost.
+2. **Bundled Metadata** -- `connector_metadata.json` ships with the CLI
package. Source, sink, and transform plugins include full option rules and
value constraints exported from SeaTunnel engine via reflection. Zero LLM token
cost.
+
+Transform metadata is resolved through the same path as source and sink
metadata, so prompts can include transform-specific required options and
constraints such as non-blank SQL queries.
### Memory System
@@ -394,7 +397,7 @@ Memory is stored locally at `.data/memory.json` (co-located
with the CLI package
## Connector Metadata
-The CLI ships with `connector_metadata.json` (150 connectors), exported from
the SeaTunnel engine via runtime reflection. No extra steps needed.
+The CLI ships with `connector_metadata.json`, exported from the SeaTunnel
engine via runtime reflection. It includes source, sink, and transform plugin
option rules, conditional options, and value constraints. No extra steps needed.
To re-export for a different SeaTunnel version (requires a running engine):
diff --git a/seatunnel-cli/README.zh-CN.md b/seatunnel-cli/README.zh-CN.md
index f59ea644ff..0b0f97f57b 100644
--- a/seatunnel-cli/README.zh-CN.md
+++ b/seatunnel-cli/README.zh-CN.md
@@ -10,6 +10,7 @@
- **多 LLM 提供商** -- 支持 AWS Bedrock、Anthropic API、OpenAI（及兼容 API，如 Azure OpenAI）
- **多智能体流水线** -- 规划器 -> 生成器 -> 校验器 -> 自动修复，最多 3 轮纠错
- **100+ 连接器** -- 全面覆盖 SeaTunnel 连接器生态，支持运行时元数据反射
+- **Transform 元数据** -- Source、Sink 和 Transform 插件在生成配置时都支持完整选项规则和值约束
- **技能框架** -- 三层生成：技能 SOP -> 黄金示例 -> 连接器元数据
- **自动保存** -- 生成的配置自动保存到 `.data/last_job.conf`（与 CLI 同目录）
- **自动修复** -- `/check` 和 `/run` 失败时自动触发 LLM 诊断和配置修复
@@ -241,7 +242,7 @@ seatunnel [request] [options]
| `/save <path>` | 将配置保存到自定义路径（生成时自动保存到 `.data/last_job.conf`） |
| `/check` | 试运行校验最近的配置；失败时自动诊断并修复 |
| `/run` | 通过 REST API 或 `seatunnel.sh` 执行最近的配置；失败时自动诊断 |
-| `/connectors` | 列出所有可用的 Source、Sink 和 Transform |
+| `/connectors` | 列出所有可用的 Source、Sink 和 Transform；生成配置时支持 Transform 选项规则和值约束 |
| `/sessions` | 列出最近的对话会话 |
| `/resume [id]` | 恢复之前的会话 |
| `/new` | 开始新会话 |
@@ -371,7 +372,9 @@ seatunnel [request] [options]
两级解析，智能回退：
1. **运行时 API** -- 从运行中的 SeaTunnel 引擎获取实时元数据（`/option-rules` 端点）。始终准确，零维护成本。
-2. **内置元数据** -- `connector_metadata.json` 随 CLI 包分发。包含 150 个连接器的完整选项规则，通过
SeaTunnel 引擎反射导出。零 LLM Token 消耗。
+2. **内置元数据** -- `connector_metadata.json` 随 CLI 包分发。Source、Sink 和 Transform
插件都包含通过 SeaTunnel 引擎反射导出的完整选项规则和值约束。零 LLM Token 消耗。
+
+Transform 元数据与 Source、Sink 元数据走同一条解析路径，因此提示词可以包含 Transform 专属必填项和值约束，例如 SQL
查询不能为空。
### 记忆系统
@@ -394,7 +397,7 @@ CLI 跨会话记忆信息，以提高配置准确性：
## 连接器元数据
-CLI 内置 `connector_metadata.json`（150 个连接器），通过运行时反射从 SeaTunnel 引擎导出，无需额外步骤。
+CLI 内置 `connector_metadata.json`，通过运行时反射从 SeaTunnel 引擎导出。它包含 Source、Sink 和
Transform 插件的选项规则、条件选项和值约束，无需额外步骤。
如需为不同 SeaTunnel 版本重新导出（需要运行中的引擎）：
diff --git a/seatunnel-cli/seatunnel_cli/agents.py
b/seatunnel-cli/seatunnel_cli/agents.py
index 833f2d0f22..89c5010a4c 100644
--- a/seatunnel-cli/seatunnel_cli/agents.py
+++ b/seatunnel-cli/seatunnel_cli/agents.py
@@ -68,8 +68,10 @@ TOOLS = [
"toolSpec": {
"name": "get_connector_info",
"description": "Get detailed info about a specific connector
including parameters and examples. "
- "IMPORTANT: Always specify connector_type ('source'
or 'sink') to get the correct "
- "type-specific options. Source and sink connectors
have different required/optional parameters.",
+ "IMPORTANT: Always specify connector_type "
+ "('source', 'sink', or 'transform') to get the
correct "
+ "type-specific options. Source, sink, and transform
plugins can "
+ "have different required/optional parameters.",
"inputSchema": {
"json": {
"type": "object",
@@ -80,9 +82,10 @@ TOOLS = [
},
"connector_type": {
"type": "string",
- "enum": ["source", "sink"],
- "description": "Whether this connector is used as
'source' or 'sink'. "
- "Source and sink have different
options — always specify this.",
+ "enum": ["source", "sink", "transform"],
+ "description": "Whether this plugin is used as
'source', 'sink', "
+ "or 'transform'. These plugin types
have different "
+ "options — always specify this.",
},
},
"required": ["connector_name"],
diff --git a/seatunnel-cli/seatunnel_cli/cli.py
b/seatunnel-cli/seatunnel_cli/cli.py
index e357d99783..6be82c7f70 100644
--- a/seatunnel-cli/seatunnel_cli/cli.py
+++ b/seatunnel-cli/seatunnel_cli/cli.py
@@ -70,7 +70,7 @@ Generate Apache SeaTunnel configs with natural language.
[bold]/save <path>[/bold] — Save config to custom path (auto-saved to
.data/last_job.conf)
[bold]/check[/bold] — Dry-run validate last config (auto-fixes on
failure)
[bold]/run[/bold] — Execute last config with SeaTunnel
- [bold]/connectors[/bold] — List available connectors
+ [bold]/connectors[/bold] — List available sources, sinks, and transforms
[bold]/config[/bold] — Show/change LLM provider settings
[bold]/sessions[/bold] — List recent sessions
[bold]/resume [id][/bold] — Resume a previous session
@@ -1066,6 +1066,7 @@ class SeaTunnelCLI:
self.console.print(f" [bold]Sources:[/bold] {',
'.join(names['sources'])}")
self.console.print(f" [bold]Sinks:[/bold] {',
'.join(names['sinks'])}")
self.console.print(f" [bold]Transforms:[/bold] {',
'.join(names['transforms'])}")
+ self.console.print(" Transform option rules and value constraints
are supported during generation.")
elif command == "/config":
self._cmd_config(arg.strip())
diff --git a/seatunnel-cli/seatunnel_cli/connectors.py
b/seatunnel-cli/seatunnel_cli/connectors.py
index 614df5d91c..1fa2e73769 100644
--- a/seatunnel-cli/seatunnel_cli/connectors.py
+++ b/seatunnel-cli/seatunnel_cli/connectors.py
@@ -180,6 +180,11 @@ def _api_response_to_detail(resp: dict) -> dict:
"then_options": nested_opts,
})
+ value_constraints = [
+ _value_constraint_to_dict(constraint)
+ for constraint in rule.get("valueConstraints", [])
+ ]
+
return {
"name": resp.get("pluginName", ""),
"types": [resp.get("pluginType", "unknown")],
@@ -187,6 +192,7 @@ def _api_response_to_detail(resp: dict) -> dict:
"optional": optional,
"exclusive": [],
"conditional": conditional,
+ "value_constraints": value_constraints,
"examples": [],
"source": "runtime_api",
}
@@ -242,6 +248,43 @@ def _api_opt_to_dict(opt: dict, rule_type: str = "") ->
dict:
return result
+def _value_constraint_to_dict(constraint: dict) -> dict:
+ """Convert one value constraint to a compact prompt-friendly form."""
+ result = {"expression": constraint.get("expression", "")}
+ tree = constraint.get("conditionTree") or {}
+ if not isinstance(tree, dict):
+ return result
+
+ option = tree.get("option") or {}
+ key = tree.get("key") or option.get("key")
+ if key:
+ result["key"] = key
+
+ expect_value = tree.get("expectValue")
+ if expect_value is not None:
+ result["expect"] = _constraint_value_to_text(expect_value)
+
+ operator = tree.get("compareOperator")
+ if operator:
+ result["operator"] = str(operator)
+
+ condition_operator = tree.get("conditionOperator")
+ if condition_operator:
+ result["condition_operator"] = str(condition_operator)
+
+ return result
+
+
+def _constraint_value_to_text(value) -> str:
+ """Return a stable, prompt-friendly string for a constraint expected
value."""
+ if isinstance(value, str):
+ return value
+ try:
+ return json.dumps(value, ensure_ascii=False)
+ except TypeError:
+ return str(value)
+
+
# ─── Runtime JSON metadata (from Java exporter) ───
_RUNTIME_METADATA: dict | None = None # Parsed connector_metadata.json
@@ -371,6 +414,11 @@ def _runtime_metadata_to_detail(entry: dict) -> dict:
"then_options": [_runtime_opt_to_dict(opt)],
})
+ value_constraints = [
+ _value_constraint_to_dict(constraint)
+ for constraint in entry.get("valueConstraints", [])
+ ]
+
detail = {
"name": entry.get("name", ""),
"types": [entry.get("type", "unknown")],
@@ -378,6 +426,7 @@ def _runtime_metadata_to_detail(entry: dict) -> dict:
"optional": optional,
"exclusive": [],
"conditional": conditional,
+ "value_constraints": value_constraints,
"examples": [],
"source": "runtime_json",
}
@@ -585,6 +634,11 @@ def format_metadata_for_prompt(metadata: dict,
plugin_name: str, plugin_type: st
opt_list = ", ".join(f"`{o['key']}`" for o in unique_opts)
lines.append(f" - When {cond_label} → {opt_list}")
+ if metadata.get("value_constraints"):
+ lines.append("**Value Constraints (must satisfy):**")
+ for constraint in metadata["value_constraints"]:
+ lines.append(_format_value_constraint(constraint, prefix=" - "))
+
# 3. Optional: show important ones with type, rest as key list
important_keys = {"username", "password", "query", "table_path",
"table_list",
"user", "bucket", "path", "topic", "database", "table",
@@ -609,6 +663,29 @@ def format_metadata_for_prompt(metadata: dict,
plugin_name: str, plugin_type: st
return "\n".join(lines)
+def _format_value_constraint(constraint: dict, prefix: str = "- ") -> str:
+ """Format one value constraint without attempting to evaluate it
locally."""
+ key = constraint.get("key")
+ operator = constraint.get("operator")
+ expect = constraint.get("expect")
+ expression = constraint.get("expression", "")
+
+ if key and operator and expect and str(operator).lower() == "extension":
+ line = f"{prefix}`{key}` {expect}"
+ elif key and operator and expect:
+ line = f"{prefix}`{key}` {operator} {expect}"
+ elif key and expect:
+ line = f"{prefix}`{key}` {expect}"
+ elif expression:
+ line = f"{prefix}{expression}"
+ else:
+ line = f"{prefix}<unspecified constraint>"
+
+ if expression and expression not in line.replace("`", ""):
+ line += f" ({expression})"
+ return line
+
+
def _format_opt_detail(opt: dict) -> str:
"""Format one option with full detail for LLM prompt."""
line = f" - `{opt['key']}` ({opt.get('type', '?')})"
@@ -904,7 +981,10 @@ def get_connector_catalog() -> str:
count = sum(len(types) for types in connectors.values())
lines = [f"## Available Connectors ({len(connectors)} connectors, {count}
endpoints) [engine: {engine_status}]\n"]
- lines.append("Use `get_connector_info` tool with `connector_type`
('source' or 'sink') to get type-specific details.")
+ lines.append(
+ "Use `get_connector_info` tool with `connector_type` "
+ "('source', 'sink', or 'transform') to get type-specific details."
+ )
if engine_status == "connected":
lines.append("(Details fetched live from running engine — always
up-to-date)\n")
elif runtime_meta:
@@ -940,8 +1020,8 @@ def get_connector_detail(name: str, connector_type: str |
None = None) -> str |
Args:
name: Connector name (e.g., 'Jdbc', 'Kafka', 'S3File').
- connector_type: 'source' or 'sink' for type-specific options.
- None returns both types with separate sections.
+ connector_type: 'source', 'sink', or 'transform' for type-specific
options.
+ None returns all available types with separate
sections.
Resolution order:
1. Runtime API (live /option-rules endpoint — always accurate)
@@ -952,7 +1032,7 @@ def get_connector_detail(name: str, connector_type: str |
None = None) -> str |
Returns a formatted string with all options, defaults, conditions, and
examples.
"""
# ── 1. Try runtime API ──
- query_types = [connector_type] if connector_type else ["source", "sink"]
+ query_types = [connector_type] if connector_type else ["source", "sink",
"transform"]
api_details: dict[str, dict] = {}
for ptype in query_types:
api_resp = _fetch_option_rules(ptype, name)
@@ -1044,6 +1124,11 @@ def _format_connector_detail_typed(
deps = ", ".join(cond["then_require"])
lines.append(f"- When `{cond['when']}` = `{cond['equals']}` →
also require: {deps}")
+ if detail.get("value_constraints"):
+ lines.append(f"\n## {type_label} — Value Constraints")
+ for constraint in detail["value_constraints"]:
+ lines.append(_format_value_constraint(constraint))
+
if len(show_types) > 1:
lines.append("") # separator between types
@@ -1126,6 +1211,7 @@ def list_connector_names() -> dict:
runtime_meta = _load_runtime_metadata()
sources = set()
sinks = set()
+ transforms = set(TRANSFORMS.keys())
if runtime_meta:
for key in runtime_meta:
parts = key.split(":", 1)
@@ -1135,8 +1221,9 @@ def list_connector_names() -> dict:
sources.add(name)
elif ctype == "sink":
sinks.add(name)
- transforms = list(TRANSFORMS.keys())
- return {"sources": sorted(sources), "sinks": sorted(sinks), "transforms":
transforms}
+ elif ctype == "transform":
+ transforms.add(name)
+ return {"sources": sorted(sources), "sinks": sorted(sinks), "transforms":
sorted(transforms)}
def validate_connector_options(
diff --git a/seatunnel-cli/seatunnel_cli/skills.py
b/seatunnel-cli/seatunnel_cli/skills.py
index 60fcdf9d58..21442e7e92 100644
--- a/seatunnel-cli/seatunnel_cli/skills.py
+++ b/seatunnel-cli/seatunnel_cli/skills.py
@@ -273,6 +273,31 @@ def _collect_required_options(
return result
+def _transform_connector_name(transform: object) -> str:
+ """Return the transform plugin name from a structured or string plan
field."""
+ if not transform:
+ return ""
+ if isinstance(transform, dict):
+ for key in ("connector", "name", "type"):
+ value = transform.get(key)
+ if value:
+ return str(value)
+ return ""
+ return str(transform)
+
+
+def _metadata_targets_for_pipeline(pipeline: PipelineSlot) -> list[tuple[str,
str]]:
+ """Return source, sink, and transform metadata targets for a pipeline."""
+ targets = [
+ (pipeline.source_connector, "source"),
+ (pipeline.sink_connector, "sink"),
+ ]
+ transform_name = _transform_connector_name(pipeline.transform)
+ if transform_name:
+ targets.append((transform_name, "transform"))
+ return [(name, ctype) for name, ctype in targets if name]
+
+
def llm_check_missing_info(
client,
user_request: str,
@@ -516,10 +541,7 @@ class SkillExecutor:
all_required: list[dict] = []
seen_keys: set[str] = set()
for p in self.plan.pipelines:
- for connector, ctype in [
- (p.source_connector, "source"),
- (p.sink_connector, "sink"),
- ]:
+ for connector, ctype in _metadata_targets_for_pipeline(p):
for opt in _collect_required_options(connector, ctype):
if opt["key"] not in seen_keys:
seen_keys.add(opt["key"])
@@ -539,7 +561,7 @@ class SkillExecutor:
seen: set[tuple[str, str]] = set()
sections: list[str] = []
for p in self.plan.pipelines:
- for name, ctype in [(p.source_connector, "source"),
(p.sink_connector, "sink")]:
+ for name, ctype in _metadata_targets_for_pipeline(p):
if (name, ctype) in seen:
continue
seen.add((name, ctype))
diff --git a/seatunnel-cli/tests/test_connector_metadata.py
b/seatunnel-cli/tests/test_connector_metadata.py
new file mode 100644
index 0000000000..0be5be8405
--- /dev/null
+++ b/seatunnel-cli/tests/test_connector_metadata.py
@@ -0,0 +1,304 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import unittest
+from unittest.mock import patch
+
+from seatunnel_cli import connectors
+from seatunnel_cli.skills import PipelineSlot, SkillExecutor, StructuredPlan
+
+
+def _runtime_transform_entry():
+ return {
+ "name": "Sql",
+ "type": "transform",
+ "required": [
+ {
+ "key": "query",
+ "type": "string",
+ "defaultValue": None,
+ "description": "SQL query",
+ "category": "absolutely_required",
+ }
+ ],
+ "optional": [],
+ "conditionRules": [],
+ "valueConstraints": [
+ {
+ "expression": "query must not be blank",
+ "conditionTree": {
+ "key": "query",
+ "expectValue": "must not be blank",
+ "compareOperator": "extension",
+ "conditionOperator": "EXTENSION",
+ "conditionOperatorCategory": "EXTENSION",
+ },
+ }
+ ],
+ }
+
+
+def _api_transform_response():
+ return {
+ "pluginName": "Sql",
+ "pluginType": "transform",
+ "optionRule": {
+ "requiredOptions": [
+ {
+ "ruleType": "absolutely_required",
+ "options": [
+ {
+ "key": "query",
+ "type": "java.lang.String",
+ "defaultValue": None,
+ "description": "SQL query",
+ }
+ ],
+ }
+ ],
+ "optionalOptions": [],
+ "conditionRules": [],
+ "valueConstraints": [
+ {
+ "expression": "query must not be blank",
+ "conditionTree": {
+ "option": {"key": "query"},
+ "expectValue": "must not be blank",
+ "compareOperator": "extension",
+ "conditionOperator": "EXTENSION",
+ "conditionOperatorCategory": "EXTENSION",
+ },
+ }
+ ],
+ },
+ }
+
+
+class ConnectorMetadataSyncTest(unittest.TestCase):
+ def test_api_response_preserves_value_constraints(self):
+ detail = connectors._api_response_to_detail(_api_transform_response())
+
+ self.assertEqual(
+ detail["value_constraints"],
+ [
+ {
+ "expression": "query must not be blank",
+ "key": "query",
+ "expect": "must not be blank",
+ "operator": "extension",
+ "condition_operator": "EXTENSION",
+ }
+ ],
+ )
+
+ def test_runtime_metadata_preserves_value_constraints(self):
+ detail =
connectors._runtime_metadata_to_detail(_runtime_transform_entry())
+
+ self.assertEqual(detail["value_constraints"][0]["key"], "query")
+ self.assertEqual(detail["value_constraints"][0]["operator"],
"extension")
+ self.assertEqual(detail["value_constraints"][0]["expect"], "must not
be blank")
+
+ def test_prompt_includes_value_constraints(self):
+ detail =
connectors._runtime_metadata_to_detail(_runtime_transform_entry())
+
+ prompt = connectors.format_metadata_for_prompt(detail, "Sql",
"transform")
+
+ self.assertIn("Value Constraints", prompt)
+ self.assertIn("`query` must not be blank", prompt)
+ self.assertNotIn("`query` extension must not be blank", prompt)
+
+ def
test_value_constraint_tolerates_missing_or_malformed_condition_tree(self):
+ constraints = [
+ {"expression": "query must not be blank"},
+ {"expression": "query must not be blank", "conditionTree": None},
+ {
+ "expression": "query must not be blank",
+ "conditionTree": ["not", "a", "dict"],
+ },
+ ]
+
+ for constraint in constraints:
+ with self.subTest(constraint=constraint):
+ self.assertEqual(
+ connectors._value_constraint_to_dict(constraint),
+ {"expression": "query must not be blank"},
+ )
+
+ def test_value_constraint_formats_complex_expect_value_as_json(self):
+ constraint = connectors._value_constraint_to_dict(
+ {
+ "expression": "mode should be in the allowed list",
+ "conditionTree": {
+ "key": "mode",
+ "expectValue": ["batch", "streaming"],
+ "compareOperator": "in",
+ },
+ }
+ )
+
+ self.assertEqual(
+ connectors._format_value_constraint(constraint),
+ '- `mode` in ["batch", "streaming"] (mode should be in the allowed
list)',
+ )
+
+ def
test_connector_detail_can_fetch_transform_metadata_from_runtime_api(self):
+ calls = []
+
+ def fake_fetch(plugin_type, plugin_name):
+ calls.append((plugin_type, plugin_name))
+ if plugin_type == "transform" and plugin_name == "Sql":
+ return _api_transform_response()
+ return None
+
+ with patch.object(connectors, "_fetch_option_rules",
side_effect=fake_fetch):
+ with patch.object(connectors, "_load_runtime_metadata",
return_value={}):
+ detail = connectors.get_connector_detail(
+ "Sql", connector_type="transform"
+ )
+
+ self.assertEqual(calls, [("transform", "Sql")])
+ self.assertIn("[source: runtime API]", detail)
+ self.assertIn("Types: transform", detail)
+ self.assertIn("Value Constraints", detail)
+
+ def
test_connector_detail_queries_transform_when_type_is_not_specified(self):
+ calls = []
+
+ def fake_fetch(plugin_type, plugin_name):
+ calls.append((plugin_type, plugin_name))
+ if plugin_type == "transform" and plugin_name == "Sql":
+ return _api_transform_response()
+ return None
+
+ with patch.object(connectors, "_fetch_option_rules",
side_effect=fake_fetch):
+ detail = connectors.get_connector_detail("Sql")
+
+ self.assertEqual(
+ calls,
+ [("source", "Sql"), ("sink", "Sql"), ("transform", "Sql")],
+ )
+ self.assertIn("Types: transform", detail)
+
+ def test_list_connector_names_includes_runtime_transforms(self):
+ runtime_meta = {
+ "source:FakeSource": {"name": "FakeSource", "type": "source"},
+ "sink:Console": {"name": "Console", "type": "sink"},
+ "transform:DataValidator": {"name": "DataValidator", "type":
"transform"},
+ }
+
+ with patch.object(
+ connectors, "_load_runtime_metadata", return_value=runtime_meta
+ ):
+ names = connectors.list_connector_names()
+
+ self.assertIn("DataValidator", names["transforms"])
+ self.assertIn("Sql", names["transforms"])
+
+ def test_skill_executor_fetches_transform_metadata_from_plan(self):
+ plan = StructuredPlan(
+ pipelines=[
+ PipelineSlot(
+ "pipeline_1",
+ source_connector="FakeSource",
+ sink_connector="Console",
+ transform="Sql",
+ )
+ ]
+ )
+ executor = SkillExecutor(plan, skills=[])
+ calls = []
+
+ def fake_fetch(name, connector_type):
+ calls.append((name, connector_type))
+ return {
+ "name": name,
+ "required": [],
+ "optional": [],
+ "conditional": [],
+ "source": "test",
+ }
+
+ with patch(
+ "seatunnel_cli.connectors.fetch_connector_metadata",
side_effect=fake_fetch
+ ):
+ metadata = executor.fetch_all_metadata(lambda *_args: None)
+
+ self.assertEqual(
+ calls,
+ [
+ ("FakeSource", "source"),
+ ("Console", "sink"),
+ ("Sql", "transform"),
+ ],
+ )
+ self.assertIn("### Sql (TRANSFORM)", metadata)
+
+ def
test_skill_executor_fill_and_check_includes_transform_required_options(self):
+ plan = StructuredPlan(
+ pipelines=[
+ PipelineSlot(
+ "pipeline_1",
+ source_connector="FakeSource",
+ sink_connector="Console",
+ transform={"connector": "Sql"},
+ )
+ ]
+ )
+ executor = SkillExecutor(plan, skills=[])
+ collect_calls = []
+
+ def fake_collect(connector, connector_type):
+ collect_calls.append((connector, connector_type))
+ if connector_type == "transform":
+ return [
+ {
+ "key": "query",
+ "type": "string",
+ "description": "SQL query",
+ "connector": connector,
+ "connector_type": connector_type,
+ }
+ ]
+ return []
+
+ with patch(
+ "seatunnel_cli.skills._collect_required_options",
side_effect=fake_collect
+ ):
+ with patch(
+ "seatunnel_cli.skills.llm_check_missing_info",
+ return_value=["Please provide query"],
+ ) as check_missing:
+ missing = executor.fill_and_check(
+ "Filter rows with a SQL transform", client=object()
+ )
+
+ self.assertEqual(
+ collect_calls,
+ [
+ ("FakeSource", "source"),
+ ("Console", "sink"),
+ ("Sql", "transform"),
+ ],
+ )
+ self.assertEqual(missing, ["Please provide query"])
+ required_options = check_missing.call_args[0][2]
+ self.assertEqual(required_options[0]["key"], "query")
+ self.assertEqual(required_options[0]["connector_type"], "transform")
+
+
+if __name__ == "__main__":
+ unittest.main()
