Add roster API for scripts to set environment variables

This commit is contained in:
vesp 2026-01-01 02:27:22 +01:00
parent c2778f419a
commit 37c29ac092
6 changed files with 320 additions and 13 deletions

View File

@ -37,10 +37,11 @@ class EnvironmentsDialog(Adw.Dialog):
'environments-updated': (GObject.SIGNAL_RUN_FIRST, None, ()),
}
def __init__(self, project, project_manager):
def __init__(self, project, project_manager, recently_updated_variables=None):
super().__init__()
self.project = project
self.project_manager = project_manager
self.recently_updated_variables = recently_updated_variables or {}
self.size_group = Gtk.SizeGroup(mode=Gtk.SizeGroupMode.HORIZONTAL)
self.header_row = None
self.data_rows = []
@ -71,10 +72,14 @@ class EnvironmentsDialog(Adw.Dialog):
# Create data rows for each variable
for var_name in self.project.variable_names:
# Check if variable was recently updated
update_timestamp = self.recently_updated_variables.get(var_name)
row = VariableDataRow(
var_name,
self.project.environments,
self.size_group
self.size_group,
update_timestamp=update_timestamp
)
row.connect('variable-changed',
self._on_variable_changed, var_name, row)

View File

@ -269,3 +269,24 @@ class ProjectManager:
break
break
self.save_projects(projects)
def batch_update_environment_variables(self, project_id: str, env_id: str, variables: dict):
"""
Update multiple variables in an environment in one operation.
Args:
project_id: Project ID
env_id: Environment ID
variables: Dict of variable_name -> value pairs to update
"""
projects = self.load_projects()
for p in projects:
if p.id == project_id:
for env in p.environments:
if env.id == env_id:
# Update all variables
for var_name, var_value in variables.items():
env.variables[var_name] = var_value
break
break
self.save_projects(projects)

View File

@ -863,6 +863,24 @@ class RequestTabWidget(Gtk.Box):
self.script_status_icon.set_from_icon_name("dialog-error-symbolic")
output_text = script_result.error if script_result.error else "Unknown error"
# Append variable updates to output
if script_result.variable_updates:
if output_text:
output_text += "\n"
output_text += "\n--- Variables Set ---"
for var_name, var_value in script_result.variable_updates.items():
# Truncate long values for display
display_value = var_value if len(var_value) <= 50 else var_value[:47] + "..."
output_text += f"\n>>> {var_name} = '{display_value}'"
# Append warnings to output
if script_result.warnings:
if output_text:
output_text += "\n"
output_text += "\n--- Warnings ---"
for warning in script_result.warnings:
output_text += f"\n{warning}"
# Set output text
buffer = self.script_output_textview.get_buffer()
buffer.set_text(output_text)

View File

@ -20,9 +20,13 @@
import subprocess
import json
import tempfile
import re
from pathlib import Path
from typing import Tuple, Optional
from dataclasses import dataclass
from typing import Tuple, Optional, Dict, List, TYPE_CHECKING
from dataclasses import dataclass, field
if TYPE_CHECKING:
from .project_manager import ProjectManager
@dataclass
@ -31,6 +35,16 @@ class ScriptResult:
success: bool
output: str # console.log output or error message
error: Optional[str] = None
variable_updates: Dict[str, str] = field(default_factory=dict) # Variables set by script
warnings: List[str] = field(default_factory=list) # Warnings (e.g., no env selected)
@dataclass
class ScriptContext:
"""Context passed to script executor for variable updates."""
project_id: Optional[str]
environment_id: Optional[str]
project_manager: 'ProjectManager'
class ScriptExecutor:
@ -39,16 +53,21 @@ class ScriptExecutor:
TIMEOUT_SECONDS = 5
@staticmethod
def execute_postprocessing_script(script_code: str, response) -> ScriptResult:
def execute_postprocessing_script(
script_code: str,
response,
context: Optional[ScriptContext] = None
) -> ScriptResult:
"""
Execute postprocessing script with response object.
Args:
script_code: JavaScript code to execute
response: HttpResponse object
context: Optional context for variable updates
Returns:
ScriptResult with output or error
ScriptResult with output, variable updates, or error
"""
if not script_code.strip():
return ScriptResult(success=True, output="")
@ -60,6 +79,21 @@ class ScriptExecutor:
script_with_context = f"""
const response = {json.dumps(response_obj)};
// Roster API for variable management
const roster = {{
__variables: {{}},
setVariable: function(name, value) {{
this.__variables[String(name)] = String(value);
}},
setVariables: function(obj) {{
for (let key in obj) {{
if (obj.hasOwnProperty(key)) {{
this.__variables[String(key)] = String(obj[key]);
}}
}}
}}
}};
// Capture console.log output
let consoleOutput = [];
const console = {{
@ -76,8 +110,10 @@ try {{
throw error;
}}
// Output results
// Output results: console logs + separator + variables JSON
print(consoleOutput.join('\\n'));
print('___ROSTER_VARIABLES___');
print(JSON.stringify(roster.__variables));
"""
# Execute with gjs
@ -86,8 +122,37 @@ print(consoleOutput.join('\\n'));
timeout=ScriptExecutor.TIMEOUT_SECONDS
)
# Parse output and extract variables
console_output = ""
variable_updates = {}
warnings = []
if returncode == 0:
return ScriptResult(success=True, output=output.strip())
# Split output into console logs and variables JSON
parts = output.split('___ROSTER_VARIABLES___')
console_output = parts[0].strip()
if len(parts) > 1:
try:
variables_json = parts[1].strip()
raw_variables = json.loads(variables_json)
# Validate variable names and add to updates
for name, value in raw_variables.items():
if ScriptExecutor._is_valid_variable_name(name):
variable_updates[name] = value
else:
warnings.append(f"Invalid variable name '{name}' (must be alphanumeric + underscore)")
except json.JSONDecodeError:
warnings.append("Failed to parse variable updates from script")
return ScriptResult(
success=True,
output=console_output,
variable_updates=variable_updates,
warnings=warnings
)
else:
error_msg = error.strip() if error else "Script execution failed"
return ScriptResult(success=False, output="", error=error_msg)
@ -112,6 +177,19 @@ print(consoleOutput.join('\\n'));
'responseTime': response.response_time_ms
}
@staticmethod
def _is_valid_variable_name(name: str) -> bool:
"""
Validate variable name (alphanumeric + underscore).
Args:
name: Variable name to validate
Returns:
True if valid, False otherwise
"""
return bool(re.match(r'^\w+$', name))
@staticmethod
def _run_gjs_script(script_code: str, timeout: int = 5) -> Tuple[str, str, int]:
"""

View File

@ -17,7 +17,8 @@
#
# SPDX-License-Identifier: GPL-3.0-or-later
from gi.repository import Gtk, GObject
from gi.repository import Gtk, GObject, GLib
from datetime import datetime, timedelta
@Gtk.Template(resource_path='/cz/vesp/roster/widgets/variable-data-row.ui')
@ -37,12 +38,13 @@ class VariableDataRow(Gtk.Box):
'value-changed': (GObject.SIGNAL_RUN_FIRST, None, (str, str)), # env_id, value
}
def __init__(self, variable_name, environments, size_group):
def __init__(self, variable_name, environments, size_group, update_timestamp=None):
super().__init__()
self.variable_name = variable_name
self.environments = environments
self.size_group = size_group
self.value_entries = {}
self.update_timeout_id = None
# Set variable name
self.name_entry.set_text(variable_name)
@ -53,6 +55,10 @@ class VariableDataRow(Gtk.Box):
self._populate_values()
# Apply visual marking if recently updated
if update_timestamp:
self._apply_update_marking(update_timestamp)
def _populate_values(self):
"""Populate value entries for each environment."""
# Clear existing
@ -104,3 +110,52 @@ class VariableDataRow(Gtk.Box):
"""Refresh with updated environments list."""
self.environments = environments
self._populate_values()
def _apply_update_marking(self, timestamp_str):
"""
Apply visual marking to show variable was recently updated.
Args:
timestamp_str: ISO format timestamp of when variable was updated
"""
try:
# Parse timestamp
update_time = datetime.fromisoformat(timestamp_str)
time_diff = datetime.now() - update_time
# Only apply marking if updated within last 3 minutes
if time_diff < timedelta(minutes=3):
# Add CSS class for visual styling
self.add_css_class('recently-updated')
# Set tooltip with update time
seconds_ago = int(time_diff.total_seconds())
if seconds_ago < 10:
tooltip_text = "✨ Updated just now by script"
elif seconds_ago < 60:
tooltip_text = f"✨ Updated {seconds_ago} seconds ago by script"
else:
minutes_ago = seconds_ago // 60
if minutes_ago == 1:
tooltip_text = "✨ Updated 1 minute ago by script"
else:
tooltip_text = f"✨ Updated {minutes_ago} minutes ago by script"
self.set_tooltip_text(tooltip_text)
# Keep highlighting visible for 60 seconds (plenty of time to see it)
if self.update_timeout_id:
GLib.source_remove(self.update_timeout_id)
self.update_timeout_id = GLib.timeout_add_seconds(60, self._remove_update_marking)
except (ValueError, AttributeError):
# Invalid timestamp format, skip marking
pass
def _remove_update_marking(self):
"""Remove visual marking after timeout."""
self.remove_css_class('recently-updated')
self.set_tooltip_text("")
self.update_timeout_id = None
return False # Don't repeat timeout

View File

@ -68,6 +68,10 @@ class RosterWindow(Adw.ApplicationWindow):
self.page_to_tab = {} # AdwTabPage -> RequestTab
self.current_tab_id = None
# Track recently updated variables for visual marking
# Format: {project_id: {var_name: timestamp}}
self.recently_updated_variables = {}
# Connect to tab view signals
self.tab_view.connect('close-page', self._on_tab_close_page)
self.tab_view.connect('notify::selected-page', self._on_tab_selected)
@ -155,6 +159,22 @@ class RosterWindow(Adw.ApplicationWindow):
entry.warning {
background-color: mix(@warning_bg_color, @view_bg_color, 0.3);
}
/* Visual marking for recently updated variables */
.recently-updated {
background: linear-gradient(90deg,
mix(@success_bg_color, @view_bg_color, 0.25),
mix(@success_bg_color, @view_bg_color, 0.05));
border-left: 4px solid @success_color;
border-radius: 6px;
transition: all 0.3s ease;
}
.recently-updated:hover {
background: linear-gradient(90deg,
mix(@success_bg_color, @view_bg_color, 0.35),
mix(@success_bg_color, @view_bg_color, 0.15));
}
""")
Gtk.StyleContext.add_provider_for_display(
@ -447,12 +467,29 @@ class RosterWindow(Adw.ApplicationWindow):
# Execute postprocessing script if exists
scripts = widget.get_scripts()
if scripts and scripts.postprocessing.strip():
from .script_executor import ScriptExecutor
from .script_executor import ScriptExecutor, ScriptContext
# Create context for variable updates
context = None
if widget.project_id and widget.selected_environment_id:
context = ScriptContext(
project_id=widget.project_id,
environment_id=widget.selected_environment_id,
project_manager=self.project_manager
)
# Execute script with context
script_result = ScriptExecutor.execute_postprocessing_script(
scripts.postprocessing,
response
response,
context
)
# Display script results
widget.display_script_results(script_result)
# Process variable updates
self._process_variable_updates(script_result, widget, context)
else:
widget._clear_script_results()
else:
@ -559,6 +596,85 @@ class RosterWindow(Adw.ApplicationWindow):
self._show_toast("Request loaded from history")
def _process_variable_updates(self, script_result, widget, context):
"""
Process variable updates from postprocessing script.
Args:
script_result: ScriptResult containing variable updates
widget: RequestTabWidget instance
context: ScriptContext or None
"""
# Check if there are any variable updates
if not script_result.variable_updates:
return
# Show warnings if context is missing
if not context or not context.project_id or not context.environment_id:
if widget.project_id:
self._show_toast("Cannot set variables: no environment selected")
else:
self._show_toast("Cannot set variables: tab not associated with project")
return
# Get project and environment
projects = self.project_manager.load_projects()
project = None
environment = None
for p in projects:
if p.id == context.project_id:
project = p
for env in p.environments:
if env.id == context.environment_id:
environment = env
break
break
if not project or not environment:
self._show_toast("Cannot set variables: project or environment not found")
return
# Process each variable update
created_vars = []
updated_vars = []
# First pass: identify and create missing variables
for var_name in script_result.variable_updates.keys():
if var_name not in project.variable_names:
# Auto-create variable
self.project_manager.add_variable(context.project_id, var_name)
created_vars.append(var_name)
else:
updated_vars.append(var_name)
# Second pass: batch update all variable values
self.project_manager.batch_update_environment_variables(
context.project_id,
context.environment_id,
script_result.variable_updates
)
# Track updated variables for visual marking
if context.project_id not in self.recently_updated_variables:
self.recently_updated_variables[context.project_id] = {}
current_time = datetime.now().isoformat()
for var_name in script_result.variable_updates.keys():
self.recently_updated_variables[context.project_id][var_name] = current_time
# Show toast notification
env_name = environment.name
if created_vars and updated_vars:
vars_list = ', '.join(created_vars + updated_vars)
self._show_toast(f"Created {len(created_vars)} and updated {len(updated_vars)} variables in {env_name}")
elif created_vars:
vars_list = ', '.join(created_vars)
self._show_toast(f"Created {len(created_vars)} variable(s) in {env_name}: {vars_list}")
elif updated_vars:
vars_list = ', '.join(updated_vars)
self._show_toast(f"Updated {len(updated_vars)} variable(s) in {env_name}: {vars_list}")
def _show_toast(self, message):
"""Show a toast notification."""
toast = Adw.Toast()
@ -689,7 +805,21 @@ class RosterWindow(Adw.ApplicationWindow):
def _on_manage_environments(self, widget, project):
"""Show environments management dialog."""
dialog = EnvironmentsDialog(project, self.project_manager)
# Reload project from disk to get latest data (including variables created by scripts)
projects = self.project_manager.load_projects()
fresh_project = None
for p in projects:
if p.id == project.id:
fresh_project = p
break
if not fresh_project:
fresh_project = project # Fallback to original if not found
# Get recently updated variables for this project
recently_updated = self.recently_updated_variables.get(project.id, {})
dialog = EnvironmentsDialog(fresh_project, self.project_manager, recently_updated)
def on_environments_updated(dlg):
# Reload projects to reflect changes