revert: Remove _configs_map changes in tools_v2 (handled by template renderer now)

This commit is contained in:
YueGuobin 2026-06-14 12:15:30 +08:00
parent c42b3a59bf
commit c3c78f99a1
No known key found for this signature in database
4 changed files with 26 additions and 39 deletions

View File

@ -543,18 +543,12 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
def _configs_map( def _configs_map(
self, device_config_list: list[dict[str, Any]] self, device_config_list: list[dict[str, Any]]
) -> dict[str, list[str]]: ) -> dict[str, list[str]]:
"""Create a mapping of device names to their configuration commands. """Create a mapping of device names to their configuration commands."""
device_configs_map = {}
Merges commands when the same device appears multiple times in the list.
"""
device_configs_map: dict[str, list[str]] = {}
for device_config in device_config_list: for device_config in device_config_list:
device_name = device_config["device_name"] device_name = device_config["device_name"]
commands = device_config.get("config_commands", []) config_commands = device_config["config_commands"]
if device_name in device_configs_map: device_configs_map[device_name] = config_commands
device_configs_map[device_name].extend(commands)
else:
device_configs_map[device_name] = list(commands)
return device_configs_map return device_configs_map

View File

@ -486,18 +486,12 @@ class ExecuteMultipleDeviceCommands(BaseTool):
def _configs_map( def _configs_map(
self, device_config_list: list[dict[str, Any]] self, device_config_list: list[dict[str, Any]]
) -> dict[str, list[str]]: ) -> dict[str, list[str]]:
"""Create a mapping of device names to their diagnostic commands. """Create a mapping of device names to their diagnostic commands."""
device_diagnostic_map = {}
Merges commands when the same device appears multiple times in the list.
"""
device_diagnostic_map: dict[str, list[str]] = {}
for device_config in device_config_list: for device_config in device_config_list:
device_name = device_config["device_name"] device_name = device_config["device_name"]
commands = device_config.get("commands", []) diagnostic_commands = device_config["commands"]
if device_name in device_diagnostic_map: device_diagnostic_map[device_name] = diagnostic_commands
device_diagnostic_map[device_name].extend(commands)
else:
device_diagnostic_map[device_name] = list(commands)
return device_diagnostic_map return device_diagnostic_map

View File

@ -405,23 +405,16 @@ class VPCSCommands(BaseTool):
""" """
Create a mapping of device names to their command lists. Create a mapping of device names to their command lists.
Merges commands when the same device appears multiple times.
Args: Args:
device_config_list: List of device configurations device_config_list: List of device configurations
Returns: Returns:
Dictionary mapping device names to command lists Dictionary mapping device names to command lists
""" """
cmd_map: dict[str, list[str]] = {} return {
for config in device_config_list: config["device_name"]: config["commands"]
name = config["device_name"] for config in device_config_list
cmds = config.get("commands", []) }
if name in cmd_map:
cmd_map[name].extend(cmds)
else:
cmd_map[name] = list(cmds)
return cmd_map
def _prepare_device_hosts_data( def _prepare_device_hosts_data(
self, self,

View File

@ -41,6 +41,9 @@ log = logging.getLogger(__name__)
def _render_template(template: str, device_configs: list[dict], commands_field: str = "config_commands") -> list[dict]: def _render_template(template: str, device_configs: list[dict], commands_field: str = "config_commands") -> list[dict]:
"""Render a Jinja2 template for each device's vars into the specified commands field. """Render a Jinja2 template for each device's vars into the specified commands field.
Entries with the same device_name are merged into a single entry
so they share one Nornir session and avoid output fragmentation.
Each device in device_configs can have: Each device in device_configs can have:
- "vars": dict of template variables (rendered into commands_field) - "vars": dict of template variables (rendered into commands_field)
- commands_field: existing commands merged after rendering if present - commands_field: existing commands merged after rendering if present
@ -49,22 +52,25 @@ def _render_template(template: str, device_configs: list[dict], commands_field:
commands_field: field name for the rendered commands, e.g. "config_commands", "commands" commands_field: field name for the rendered commands, e.g. "config_commands", "commands"
""" """
jinja = JinjaTemplate(template) jinja = JinjaTemplate(template)
results = [] merged: dict[str, dict] = {}
for dev in device_configs: for dev in device_configs:
rendered = dev.copy() name = dev.get("device_name")
vars_data = rendered.pop("vars", {}) if not name:
continue
vars_data = dev.get("vars", {})
if name not in merged:
merged[name] = {"device_name": name, commands_field: list(dev.get(commands_field, []))}
entry = merged[name]
if vars_data: if vars_data:
try: try:
output = jinja.render(**vars_data) output = jinja.render(**vars_data)
lines = [l for l in output.splitlines() if l.strip()] lines = [l for l in output.splitlines() if l.strip()]
existing = rendered.get(commands_field, []) entry[commands_field].extend(lines)
rendered[commands_field] = existing + lines
except JinjaError as e: except JinjaError as e:
error_msg = f"Template rendering failed for '{dev.get('device_name')}': {e}" error_msg = f"Template rendering failed for '{name}': {e}"
log.error(error_msg) log.error(error_msg)
return [{"error": error_msg}] return [{"error": error_msg}]
results.append(rendered) return list(merged.values())
return results
# ── Tool handlers ────────────────────────────────────────────────────────── # ── Tool handlers ──────────────────────────────────────────────────────────