# window.py
#
# Copyright 2025 Pavel Baksy
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
#
# SPDX-License-Identifier: GPL-3.0-or-later
import gi
gi.require_version('GtkSource', '5')
from gi.repository import Adw, Gtk, GLib, Gio, GtkSource
from typing import Dict, Optional
import logging
from .models import HttpRequest, HttpResponse, HistoryEntry, RequestTab
from .constants import (
UI_TOAST_TIMEOUT_SECONDS,
DELAY_TAB_CREATION_MS,
DISPLAY_URL_NAME_MAX_LENGTH,
PROJECT_NAME_MAX_LENGTH,
REQUEST_NAME_MAX_LENGTH,
)
from .http_client import HttpClient
from .history_manager import HistoryManager
from .project_manager import ProjectManager
from .tab_manager import TabManager
from .icon_picker_dialog import IconPickerDialog
from .environments_dialog import EnvironmentsDialog
from .request_tab_widget import RequestTabWidget
from .widgets.history_item import HistoryItem
from .widgets.project_item import ProjectItem
from datetime import datetime
import json
import uuid
logger = logging.getLogger(__name__)
@Gtk.Template(resource_path='/cz/bugsy/roster/main-window.ui')
class RosterWindow(Adw.ApplicationWindow):
__gtype_name__ = 'RosterWindow'
# Toast overlay
toast_overlay = Gtk.Template.Child()
# Top bar widgets
save_request_button = Gtk.Template.Child()
export_request_button = Gtk.Template.Child()
new_request_button = Gtk.Template.Child()
tab_view = Gtk.Template.Child()
tab_bar = Gtk.Template.Child()
# Panes
main_pane = Gtk.Template.Child()
# Sidebar widgets
projects_listbox = Gtk.Template.Child()
add_project_button = Gtk.Template.Child()
# History (hidden but kept for compatibility)
history_listbox = Gtk.Template.Child()
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.http_client = HttpClient()
self.history_manager = HistoryManager()
self.project_manager = ProjectManager()
self.tab_manager = TabManager()
# Map AdwTabPage to tab data
self.page_to_widget: Dict[Adw.TabPage, RequestTabWidget] = {}
self.page_to_tab: Dict[Adw.TabPage, RequestTab] = {}
self.current_tab_id: Optional[str] = None
# Track recently updated variables for visual marking
# Format: {project_id: {var_name: timestamp}}
self.recently_updated_variables: Dict[str, Dict[str, str]] = {}
# 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)
# Create window actions
self._create_actions()
# Setup custom CSS
self._setup_custom_css()
# Setup UI
self._setup_tab_system()
self._load_projects()
self._load_history()
# Connect to close-request to warn about unsaved changes
self.connect("close-request", self._on_close_request)
# Create first tab
self._create_new_tab()
def _on_close_request(self, window) -> bool:
"""Handle window close request - warn if there are unsaved changes."""
# Check if any tabs have unsaved changes
modified_tabs = []
for page, widget in self.page_to_widget.items():
if widget.modified:
tab = self.page_to_tab.get(page)
if tab:
modified_tabs.append(tab)
if modified_tabs:
# Show warning dialog
dialog = Adw.AlertDialog()
dialog.set_heading("Unsaved Changes")
if len(modified_tabs) == 1:
dialog.set_body(f"'{modified_tabs[0].name}' has unsaved changes. Close anyway?")
else:
dialog.set_body(f"{len(modified_tabs)} requests have unsaved changes. Close anyway?")
dialog.add_response("cancel", "Cancel")
dialog.add_response("close", "Close")
dialog.set_response_appearance("close", Adw.ResponseAppearance.DESTRUCTIVE)
dialog.set_default_response("cancel")
dialog.set_close_response("cancel")
dialog.connect("response", self._on_close_app_dialog_response)
dialog.present(self)
# Prevent closing for now - will close if user confirms
return True
# No unsaved changes, allow closing
return False
def _on_close_app_dialog_response(self, dialog, response) -> None:
"""Handle close application dialog response."""
if response == "close":
# User confirmed - close the window
self.destroy()
def _setup_custom_css(self) -> None:
"""Setup custom CSS for UI styling."""
css_provider = Gtk.CssProvider()
css_provider.load_from_data(b"""
/* AdwToolbarView handles header bar heights automatically */
/* Just add minimal custom styling for other elements */
/* Stack switchers styling (Headers/Body tabs) */
stackswitcher button {
padding: 6px 16px;
min-height: 32px;
border-radius: 6px;
margin: 0 2px;
}
stackswitcher button:checked {
font-weight: 900;
}
/* Warning styling for undefined variables */
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));
}
/* Sensitive variable styling */
entry.sensitive-variable {
font-family: monospace;
}
entry.sensitive-variable-name {
color: @accent_color;
font-weight: 600;
}
""")
Gtk.StyleContext.add_provider_for_display(
self.get_display(),
css_provider,
Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION
)
def _setup_tab_system(self) -> None:
"""Set up the tab system."""
# Connect new request button
self.new_request_button.connect("clicked", self._on_new_request_clicked)
def _on_tab_selected(self, tab_view, param) -> None:
"""Handle tab selection change."""
page = tab_view.get_selected_page()
if not page:
return
# Get the RequestTab for this page
tab = self.page_to_tab.get(page)
if tab:
self.current_tab_id = tab.id
def _on_tab_close_page(self, tab_view, page) -> bool:
"""Handle tab close request."""
# Get the RequestTab and widget for this page
tab = self.page_to_tab.get(page)
widget = self.page_to_widget.get(page)
# If already removed, allow close (prevents double-processing)
if not tab or not widget:
return False # Allow close
# Check if tab has unsaved changes
if widget.modified:
# Show warning dialog
dialog = Adw.AlertDialog()
dialog.set_heading("Close Unsaved Request?")
dialog.set_body(f"'{tab.name}' has unsaved changes. Close anyway?")
dialog.add_response("cancel", "Cancel")
dialog.add_response("close", "Close")
dialog.set_response_appearance("close", Adw.ResponseAppearance.DESTRUCTIVE)
dialog.set_default_response("cancel")
dialog.set_close_response("cancel")
def on_response(dlg, response):
if response == "close":
# Remove from our tracking FIRST to prevent re-triggering
self.tab_manager.close_tab(tab.id)
del self.page_to_tab[page]
del self.page_to_widget[page]
# Check if we need to create a new tab
remaining_tabs = len(self.page_to_tab)
if remaining_tabs == 0:
# Schedule creating exactly one new tab
GLib.timeout_add(DELAY_TAB_CREATION_MS, self._create_new_tab_once)
# Close the page
tab_view.close_page_finish(page, True)
else:
# Cancel the close
tab_view.close_page_finish(page, False)
dialog.connect("response", on_response)
dialog.present(self)
return True # Defer close decision to dialog
else:
# No unsaved changes, allow close
self.tab_manager.close_tab(tab.id)
del self.page_to_tab[page]
del self.page_to_widget[page]
# If this will be the last tab, create a new one
# Check after we've removed our tracking to get accurate count
remaining_tabs = len(self.page_to_tab)
if remaining_tabs == 0:
# Use a single-shot timer to create exactly one new tab
GLib.timeout_add(50, self._create_new_tab_once)
return False # Allow close
def _create_new_tab_once(self) -> bool:
"""Create a new tab (one-time callback)."""
# Only create if there really are no tabs
if self.tab_view.get_n_pages() == 0:
self._create_new_tab()
return False # Don't repeat
def _is_empty_new_request_tab(self) -> bool:
"""Check if current tab is an empty 'New Request' tab."""
if not self.current_tab_id:
return False
# Find the current page
page = self.tab_view.get_selected_page()
if not page:
return False
tab = self.page_to_tab.get(page)
widget = self.page_to_widget.get(page)
if not tab or not widget:
return False
# Check if it's a "New Request" (not from saved request)
if tab.saved_request_id:
return False
# Check if it's empty
request = widget.get_request()
# Consider headers empty if they're either empty or only contain default headers
from .models import HttpRequest
has_only_default_headers = request.headers == HttpRequest.default_headers()
is_empty = (
not request.url.strip() and
not request.body.strip() and
(not request.headers or has_only_default_headers)
)
return is_empty
def _find_tab_by_saved_request_id(self, saved_request_id) -> Optional[RequestTab]:
"""Find a tab by its saved_request_id. Returns None if not found."""
for tab in self.tab_manager.tabs:
if tab.saved_request_id == saved_request_id:
return tab
return None
def _generate_copy_name(self, base_name: str) -> str:
"""Generate a unique copy name like 'ReqA (copy)', 'ReqA (copy 2)', etc."""
# Check if "Name (copy)" exists
existing_names = {tab.name for tab in self.tab_manager.tabs}
copy_name = f"{base_name} (copy)"
if copy_name not in existing_names:
return copy_name
# Find the highest number used
counter = 2
while f"{base_name} (copy {counter})" in existing_names:
counter += 1
return f"{base_name} (copy {counter})"
def _create_new_tab(self, name="New Request", request=None, saved_request_id=None, response=None,
project_id=None, selected_environment_id=None, scripts=None) -> str:
"""Create a new tab with RequestTabWidget."""
# Create tab in tab manager
if not request:
request = HttpRequest(method="GET", url="", headers=HttpRequest.default_headers(), body="", syntax="RAW")
tab = self.tab_manager.create_tab(name, request, saved_request_id, response,
project_id, selected_environment_id, scripts)
self.current_tab_id = tab.id
# Create RequestTabWidget for this tab
widget = RequestTabWidget(tab.id, request, response, project_id, selected_environment_id, scripts)
widget.original_request = tab.original_request
widget.project_manager = self.project_manager # Inject project manager
# Set original scripts for change tracking (if this is a saved request)
if saved_request_id and scripts:
from .models import Scripts
widget.original_scripts = Scripts(
preprocessing=scripts.preprocessing,
postprocessing=scripts.postprocessing
)
# Populate environment dropdown if project_id is set
if project_id and hasattr(widget, '_populate_environment_dropdown'):
widget._populate_environment_dropdown()
# Connect to send button
widget.send_button.connect("clicked", lambda btn: self._on_send_clicked(widget))
# Create AdwTabPage
page = self.tab_view.append(widget)
page.set_title(name)
page.set_indicator_activatable(False)
# Store mappings
self.page_to_tab[page] = tab
self.page_to_widget[page] = widget
# Connect to modified state changes
widget.connect('modified-changed', lambda w, modified: self._on_tab_modified_changed(page, modified))
# Select this new page
self.tab_view.set_selected_page(page)
return tab.id
def _get_project_for_saved_request(self, saved_request_id):
"""Find which project contains a saved request. Returns (project_id, default_env_id) or (None, None)."""
if not saved_request_id:
return None, None
projects = self.project_manager.load_projects()
for project in projects:
for request in project.requests:
if request.id == saved_request_id:
# Found the project - get default environment (first one)
default_env_id = project.environments[0].id if project.environments else None
return project.id, default_env_id
return None, None
def _on_tab_modified_changed(self, page, modified):
"""Update tab indicator when modified state changes."""
# Update the tab title with/without modified indicator
tab = self.page_to_tab.get(page)
if tab:
if modified:
# Add star to title
page.set_title(f"*{tab.name}")
else:
# Remove star from title
page.set_title(tab.name)
def _switch_to_tab(self, tab_id: str) -> None:
"""Switch to a tab by its ID."""
# Find the page for this tab
for page, tab in self.page_to_tab.items():
if tab.id == tab_id:
self.tab_view.set_selected_page(page)
return
def _close_current_tab(self) -> None:
"""Close the currently active tab."""
page = self.tab_view.get_selected_page()
if page:
self.tab_view.close_page(page)
def _focus_url_field(self) -> None:
"""Focus the URL field in the current tab."""
page = self.tab_view.get_selected_page()
if page:
widget = self.page_to_widget.get(page)
if widget and hasattr(widget, 'url_entry'):
widget.url_entry.grab_focus()
def _send_current_request(self) -> None:
"""Send the request from the current tab."""
page = self.tab_view.get_selected_page()
if page:
widget = self.page_to_widget.get(page)
if widget and hasattr(widget, 'send_button'):
# Trigger the send button click
self._on_send_clicked(widget)
def _on_new_request_clicked(self, button):
"""Handle New Request button click."""
self._create_new_tab()
def _create_actions(self) -> None:
"""Create window-level actions."""
# New tab shortcut (Ctrl+T)
action = Gio.SimpleAction.new("new-tab", None)
action.connect("activate", lambda a, p: self._on_new_request_clicked(None))
self.add_action(action)
self.get_application().set_accels_for_action("win.new-tab", ["t"])
# Close tab shortcut (Ctrl+W)
action = Gio.SimpleAction.new("close-tab", None)
action.connect("activate", lambda a, p: self._close_current_tab())
self.add_action(action)
self.get_application().set_accels_for_action("win.close-tab", ["w"])
# Save request shortcut (Ctrl+S)
action = Gio.SimpleAction.new("save-request", None)
action.connect("activate", lambda a, p: self.on_save_request_clicked(None))
self.add_action(action)
self.get_application().set_accels_for_action("win.save-request", ["s"])
# Focus URL field shortcut (Ctrl+L)
action = Gio.SimpleAction.new("focus-url", None)
action.connect("activate", lambda a, p: self._focus_url_field())
self.add_action(action)
self.get_application().set_accels_for_action("win.focus-url", ["l"])
# Send request shortcut (Ctrl+Return)
action = Gio.SimpleAction.new("send-request", None)
action.connect("activate", lambda a, p: self._send_current_request())
self.add_action(action)
self.get_application().set_accels_for_action("win.send-request", ["Return"])
def _on_send_clicked(self, widget):
"""Handle Send button click from a tab widget."""
# Clear previous preprocessing results
if hasattr(widget, '_clear_preprocessing_results'):
widget._clear_preprocessing_results()
# Validate URL
request = widget.get_request()
if not request.url.strip():
self._show_toast("Please enter a URL")
return
# Execute preprocessing script (if exists)
scripts = widget.get_scripts()
modified_request = request
if scripts and scripts.preprocessing.strip():
from .script_executor import ScriptContext
preprocessing_result = self._execute_preprocessing(
widget, scripts.preprocessing, request
)
# Display preprocessing results
if hasattr(widget, 'display_preprocessing_results'):
widget.display_preprocessing_results(preprocessing_result)
# Handle errors - DON'T send request if preprocessing failed
if not preprocessing_result.success:
self._show_toast(f"Preprocessing error: {preprocessing_result.error}")
return # Early return, don't send request
# 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
)
# Process variable updates from preprocessing
self._process_variable_updates(preprocessing_result, widget, context)
# Use modified request (if returned)
if preprocessing_result.modified_request:
modified_request = preprocessing_result.modified_request
# Disable send button during request
widget.send_button.set_sensitive(False)
def proceed_with_request(env):
"""Continue with request after environment is loaded."""
# Apply variable substitution to (possibly modified) request
substituted_request = modified_request
if env:
from .variable_substitution import VariableSubstitution
substituted_request, undefined = VariableSubstitution.substitute_request(modified_request, env)
# Log undefined variables for debugging
if undefined:
logger.warning(f"Undefined variables in request: {', '.join(undefined)}")
# Execute async
def callback(response, error, user_data):
"""Callback runs on main thread."""
# Re-enable send button
widget.send_button.set_sensitive(True)
# Update tab with response
if response:
widget.display_response(response)
# Execute postprocessing script if exists
scripts = widget.get_scripts()
if scripts and scripts.postprocessing.strip():
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,
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:
widget.display_error(error or "Unknown error")
widget._clear_script_results()
# Save response to tab
page = self.tab_view.get_selected_page()
if page:
tab = self.page_to_tab.get(page)
if tab:
tab.response = response
# Create history entry with redacted sensitive variables
# Determine which request to save to history
request_for_history = modified_request
has_redacted = False
if env and widget.project_id:
# Get the project's sensitive variables list
projects = self.project_manager.load_projects()
sensitive_vars = []
for p in projects:
if p.id == widget.project_id:
sensitive_vars = p.sensitive_variables
break
# Create redacted request (only non-sensitive variables substituted)
if sensitive_vars:
from .variable_substitution import VariableSubstitution
request_for_history, _, has_redacted = VariableSubstitution.substitute_request_excluding_sensitive(
modified_request, env, sensitive_vars
)
else:
# No sensitive variables defined, use fully substituted request
request_for_history = substituted_request
elif env:
# Environment loaded but no project - use fully substituted request
request_for_history = substituted_request
entry = HistoryEntry(
timestamp=datetime.now().isoformat(),
request=request_for_history,
response=response,
error=error,
has_redacted_variables=has_redacted
)
self.history_manager.add_entry(entry)
# Refresh history panel to show new entry
self._load_history()
self.http_client.execute_request_async(substituted_request, callback, None)
# Fetch environment asynchronously then proceed
if widget.selected_environment_id:
widget.get_selected_environment(proceed_with_request)
else:
proceed_with_request(None)
# History and Project Management
def _load_history(self) -> None:
"""Load history from file and populate list."""
# Clear existing history items
while True:
child = self.history_listbox.get_first_child()
if child is None:
break
self.history_listbox.remove(child)
# Load and display history
entries = self.history_manager.load_history()
for entry in entries:
item = HistoryItem(entry)
item.connect('load-requested', self._on_history_load_requested, entry)
item.connect('delete-requested', self._on_history_delete_requested, entry)
self.history_listbox.append(item)
def _on_history_load_requested(self, widget, entry):
"""Handle load request from history item."""
# Smart loading: replaces empty tabs or creates new tab if modified
self._load_request_from_entry(entry)
def _on_history_delete_requested(self, widget, entry):
"""Handle delete request from history item."""
# Delete the entry from history
self.history_manager.delete_entry(entry)
# Refresh the history panel
self._load_history()
def _load_request_from_entry(self, entry):
"""Load request from history entry - smart loading based on current tab state."""
request = entry.request
# Generate name from method and URL
url_parts = request.url.split('/')
url_name = url_parts[-1] if url_parts else request.url
name = f"{request.method} {url_name[:DISPLAY_URL_NAME_MAX_LENGTH]}" if url_name else f"{request.method} Request"
# Check if current tab is an empty "New Request"
if self._is_empty_new_request_tab():
# Replace the empty tab with this request
if self.current_tab_id:
current_tab = self.tab_manager.get_tab_by_id(self.current_tab_id)
if current_tab:
# Update the tab
current_tab.name = name
current_tab.request = request
current_tab.saved_request_id = None # Not from saved requests
current_tab.original_request = HttpRequest(
method=request.method,
url=request.url,
headers=request.headers.copy(),
body=request.body,
syntax=getattr(request, 'syntax', 'RAW')
)
current_tab.modified = False
# Load into UI - get current widget and update it
page = self.tab_view.get_selected_page()
if page:
widget = self.page_to_widget.get(page)
if widget:
widget._load_request(request)
widget.original_request = current_tab.original_request
widget.modified = False
# Update tab page title
page.set_title(name)
else:
# Current tab has changes or is not a "New Request"
# Create a new tab
self._create_new_tab(
name=name,
request=request,
saved_request_id=None
)
self._show_toast("Request loaded from history")
def _execute_preprocessing(self, widget, script_code, request):
"""
Execute preprocessing script and return result.
Args:
widget: RequestTabWidget instance
script_code: JavaScript preprocessing script
request: HttpRequest to be preprocessed
Returns:
ScriptResult with modified request and variable updates
"""
from .script_executor import ScriptExecutor, ScriptContext
# Create context for variable access
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 preprocessing
return ScriptExecutor.execute_preprocessing_script(
script_code,
request,
context
)
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: str) -> None:
"""Show a toast notification."""
toast = Adw.Toast()
toast.set_title(message)
toast.set_timeout(UI_TOAST_TIMEOUT_SECONDS)
self.toast_overlay.add_toast(toast)
# Project Management Methods
def _validate_project_name(self, name: str) -> tuple[bool, str]:
"""
Validate project name.
Args:
name: The project name to validate
Returns:
Tuple of (is_valid, error_message)
"""
# Check if empty
if not name or not name.strip():
return False, "Project name cannot be empty"
# Check length
if len(name) > PROJECT_NAME_MAX_LENGTH:
return False, f"Project name is too long (max {PROJECT_NAME_MAX_LENGTH} characters)"
# Check for invalid characters (file system unsafe characters)
invalid_chars = ['/', '\\', ':', '*', '?', '"', '<', '>', '|', '\0']
for char in invalid_chars:
if char in name:
return False, f"Project name cannot contain '{char}'"
# Check if it's only whitespace
if name.strip() == '':
return False, "Project name cannot be only whitespace"
return True, ""
def _validate_request_name(self, name: str) -> tuple[bool, str]:
"""
Validate request name.
Args:
name: The request name to validate
Returns:
Tuple of (is_valid, error_message)
"""
# Check if empty
if not name or not name.strip():
return False, "Request name cannot be empty"
# Check length
if len(name) > REQUEST_NAME_MAX_LENGTH:
return False, f"Request name is too long (max {REQUEST_NAME_MAX_LENGTH} characters)"
# Check for invalid characters (file system unsafe characters)
invalid_chars = ['/', '\\', ':', '*', '?', '"', '<', '>', '|', '\0']
for char in invalid_chars:
if char in name:
return False, f"Request name cannot contain '{char}'"
return True, ""
def _load_projects(self) -> None:
"""Load and display projects."""
# Clear existing
while child := self.projects_listbox.get_first_child():
self.projects_listbox.remove(child)
# Load and populate
projects = self.project_manager.load_projects()
for project in projects:
item = ProjectItem(project)
item.connect('edit-requested', self._on_project_edit, project)
item.connect('delete-requested', self._on_project_delete, project)
item.connect('add-request-requested', self._on_add_to_project, project)
item.connect('request-load-requested', self._on_load_request)
item.connect('request-delete-requested', self._on_delete_request, project)
item.connect('manage-environments-requested', self._on_manage_environments, project)
self.projects_listbox.append(item)
@Gtk.Template.Callback()
def on_add_project_clicked(self, button):
"""Show dialog to create project."""
dialog = Adw.AlertDialog()
dialog.set_heading("New Project")
dialog.set_body("Enter a name for the new project:")
entry = Gtk.Entry()
entry.set_placeholder_text("Project name")
dialog.set_extra_child(entry)
dialog.add_response("cancel", "Cancel")
dialog.add_response("create", "Create")
dialog.set_response_appearance("create", Adw.ResponseAppearance.SUGGESTED)
dialog.set_default_response("create")
dialog.set_close_response("cancel")
def on_response(dlg, response):
if response == "create":
name = entry.get_text().strip()
is_valid, error_msg = self._validate_project_name(name)
if is_valid:
self.project_manager.add_project(name)
self._load_projects()
self._show_toast(f"Project '{name}' created")
else:
self._show_toast(error_msg)
dialog.connect("response", on_response)
dialog.present(self)
def _on_project_edit(self, widget, project):
"""Show edit project dialog with icon picker."""
dialog = Adw.AlertDialog()
dialog.set_heading("Edit Project")
dialog.set_body(f"Edit '{project.name}':")
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=12)
# Project name entry
name_label = Gtk.Label(label="Name:", xalign=0)
entry = Gtk.Entry()
entry.set_text(project.name)
box.append(name_label)
box.append(entry)
# Icon selector button
icon_label = Gtk.Label(label="Icon:", xalign=0)
icon_button = Gtk.Button()
icon_button.set_child(Gtk.Image.new_from_icon_name(project.icon))
icon_button.selected_icon = project.icon # Store current selection
def on_icon_button_clicked(btn):
icon_dialog = IconPickerDialog(current_icon=btn.selected_icon)
def on_icon_selected(dlg, icon_name):
btn.selected_icon = icon_name
btn.set_child(Gtk.Image.new_from_icon_name(icon_name))
icon_dialog.connect('icon-selected', on_icon_selected)
icon_dialog.present(self)
icon_button.connect('clicked', on_icon_button_clicked)
box.append(icon_label)
box.append(icon_button)
dialog.set_extra_child(box)
dialog.add_response("cancel", "Cancel")
dialog.add_response("save", "Save")
dialog.set_response_appearance("save", Adw.ResponseAppearance.SUGGESTED)
def on_response(dlg, response):
if response == "save":
new_name = entry.get_text().strip()
new_icon = icon_button.selected_icon
is_valid, error_msg = self._validate_project_name(new_name)
if is_valid:
self.project_manager.update_project(project.id, new_name, new_icon)
self._load_projects()
self._show_toast(f"Project updated")
else:
self._show_toast(error_msg)
dialog.connect("response", on_response)
dialog.present(self)
def _on_project_delete(self, widget, project):
"""Show delete confirmation."""
dialog = Adw.AlertDialog()
dialog.set_heading("Delete Project?")
dialog.set_body(f"Delete '{project.name}' and all its saved requests? This cannot be undone.")
dialog.add_response("cancel", "Cancel")
dialog.add_response("delete", "Delete")
dialog.set_response_appearance("delete", Adw.ResponseAppearance.DESTRUCTIVE)
dialog.set_close_response("cancel")
def on_response(dlg, response):
if response == "delete":
self.project_manager.delete_project(project.id)
self._load_projects()
dialog.connect("response", on_response)
dialog.present(self)
def _on_manage_environments(self, widget, project):
"""Show environments management dialog."""
# Get latest project data (includes variables created by scripts)
# Note: load_projects() uses cached data that's kept up-to-date by save operations
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
self._load_projects()
dialog.connect('environments-updated', on_environments_updated)
dialog.present(self)
@Gtk.Template.Callback()
def on_save_request_clicked(self, button):
"""Save current request to a project."""
# Get request from current tab widget
page = self.tab_view.get_selected_page()
if not page:
self._show_toast("No active request")
return
widget = self.page_to_widget.get(page)
if not widget:
self._show_toast("No active request")
return
request = widget.get_request()
scripts = widget.get_scripts()
if not request.url.strip():
self._show_toast("Cannot save: URL is empty")
return
projects = self.project_manager.load_projects()
if not projects:
# Show error - no projects
dialog = Adw.AlertDialog()
dialog.set_heading("No Projects")
dialog.set_body("Create a project first before saving requests.")
dialog.add_response("ok", "OK")
dialog.present(self)
return
# Create save dialog
dialog = Adw.AlertDialog()
dialog.set_heading("Save Request")
dialog.set_body("Choose a project and enter a name:")
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=12)
# Check if current tab has a saved request or project (for pre-filling)
current_tab = None
preselect_project_index = 0
prefill_name = ""
if self.current_tab_id:
current_tab = self.tab_manager.get_tab_by_id(self.current_tab_id)
if current_tab:
if current_tab.saved_request_id:
# Find which project this request belongs to
for i, p in enumerate(projects):
for req in p.requests:
if req.id == current_tab.saved_request_id:
preselect_project_index = i
prefill_name = current_tab.name
break
elif current_tab.project_id:
# Tab associated with a project (e.g., from "Add Request")
for i, p in enumerate(projects):
if p.id == current_tab.project_id:
preselect_project_index = i
break
if current_tab.name and current_tab.name != "New Request":
# Copy tab or named unsaved tab - prefill with tab name
prefill_name = current_tab.name
# Project dropdown
project_list = Gtk.StringList()
for p in projects:
project_list.append(p.name)
dropdown = Gtk.DropDown(model=project_list)
dropdown.set_selected(preselect_project_index) # Pre-select project
box.append(dropdown)
# Name entry
entry = Gtk.Entry()
entry.set_placeholder_text("Request name")
if prefill_name:
entry.set_text(prefill_name) # Pre-fill name
# Select all text for easy editing
entry.select_region(0, -1)
box.append(entry)
dialog.set_extra_child(box)
dialog.add_response("cancel", "Cancel")
dialog.add_response("save", "Save")
dialog.set_response_appearance("save", Adw.ResponseAppearance.SUGGESTED)
def on_response(dlg, response):
if response == "save":
name = entry.get_text().strip()
is_valid, error_msg = self._validate_request_name(name)
if is_valid:
selected = dropdown.get_selected()
project = projects[selected]
# Check for duplicate name
existing = self.project_manager.find_request_by_name(project.id, name)
if existing:
# Show overwrite confirmation
self._show_overwrite_dialog(project, name, existing.id, request, scripts)
else:
# No duplicate, save normally
saved_request = self.project_manager.add_request(project.id, name, request, scripts)
self._load_projects()
self._show_toast(f"Saved as '{name}'")
# Clear modified flag on current tab
self._mark_tab_as_saved(saved_request.id, name, request, scripts)
else:
self._show_toast(error_msg)
dialog.connect("response", on_response)
dialog.present(self)
@Gtk.Template.Callback()
def on_export_request_clicked(self, button):
"""Handle Export button click - export request as cURL command."""
# Get current tab
page = self.tab_view.get_selected_page()
if not page:
self._show_toast("No active request")
return
widget = self.page_to_widget.get(page)
if not widget:
self._show_toast("No active request")
return
# Get request from widget (raw with {{variables}})
request = widget.get_request()
# Validate URL
if not request.url.strip():
self._show_toast("Cannot export: URL is empty")
return
# Apply variable substitution if environment is selected
substituted_request = request
if widget.selected_environment_id:
env = widget.get_selected_environment()
if env:
from .variable_substitution import VariableSubstitution
substituted_request, undefined = VariableSubstitution.substitute_request(request, env)
# Warn about undefined variables
if undefined:
logger.warning(f"Undefined variables in export: {', '.join(undefined)}")
# Show warning toast but continue with export
self._show_toast(f"Warning: {len(undefined)} undefined variable(s)")
# Show export dialog with substituted request
from .export_dialog import ExportDialog
dialog = ExportDialog(substituted_request)
dialog.present(self)
def _mark_tab_as_saved(self, saved_request_id, name, request, scripts=None):
"""Mark the current tab as saved (clear modified flag)."""
page = self.tab_view.get_selected_page()
if not page:
return
tab = self.page_to_tab.get(page)
widget = self.page_to_widget.get(page)
if tab and widget:
# Update tab metadata
tab.saved_request_id = saved_request_id
tab.name = name
tab.modified = False
tab.scripts = scripts
# Update original_request to match saved state
original = HttpRequest(
method=request.method,
url=request.url,
headers=request.headers.copy(),
body=request.body,
syntax=request.syntax
)
tab.original_request = original
# Update original_scripts to match saved state
if scripts:
from .models import Scripts
original_scripts = Scripts(
preprocessing=scripts.preprocessing,
postprocessing=scripts.postprocessing
)
widget.original_scripts = original_scripts
else:
widget.original_scripts = None
# Update widget state
widget.original_request = original
widget.modified = False
# Update tab page title (widget.modified is now False, so no star)
page.set_title(name)
def _show_overwrite_dialog(self, project, name, existing_request_id, request, scripts=None):
"""Show dialog asking if user wants to overwrite existing request."""
dialog = Adw.AlertDialog()
dialog.set_heading("Request Already Exists")
dialog.set_body(f"A request named '{name}' already exists in '{project.name}'.\n\nDo you want to overwrite it?")
dialog.add_response("cancel", "Cancel")
dialog.add_response("overwrite", "Overwrite")
dialog.set_response_appearance("overwrite", Adw.ResponseAppearance.DESTRUCTIVE)
dialog.set_default_response("cancel")
dialog.set_close_response("cancel")
def on_overwrite_response(dlg, response):
if response == "overwrite":
# Update the existing request
self.project_manager.update_request(project.id, existing_request_id, name, request, scripts)
self._load_projects()
self._show_toast(f"Updated '{name}'")
# Clear modified flag on current tab
self._mark_tab_as_saved(existing_request_id, name, request, scripts)
dialog.connect("response", on_overwrite_response)
dialog.present(self)
def _on_add_to_project(self, widget, project):
"""Create a new empty request tab associated with the project."""
# Get default environment for the project
default_env_id = project.environments[0].id if project.environments else None
# Create a new tab with project_id set so Save will pre-select this project
self._create_new_tab(
name="New Request",
project_id=project.id,
selected_environment_id=default_env_id
)
self._show_toast(f"New request for '{project.name}'")
def _on_load_request(self, widget, saved_request):
"""Load saved request - smart loading based on current tab state."""
req = saved_request.request
scripts = saved_request.scripts
# First, check if this request is already open in an unmodified tab
existing_tab = self._find_tab_by_saved_request_id(saved_request.id)
if existing_tab and not existing_tab.is_modified():
# Switch to the existing unmodified tab
self._switch_to_tab(existing_tab.id)
self._show_toast(f"Switched to '{saved_request.name}'")
return
# Check if there's a modified tab with this saved_request_id
# If so, this is a copy and should not be linked
is_copy = existing_tab and existing_tab.is_modified()
if is_copy:
# Generate numbered copy name
tab_name = self._generate_copy_name(saved_request.name)
link_to_saved = None
else:
tab_name = saved_request.name
link_to_saved = saved_request.id
# Check if current tab is an empty "New Request"
if self._is_empty_new_request_tab():
# Replace the empty tab with this request
page = self.tab_view.get_selected_page()
if page:
widget = self.page_to_widget.get(page)
current_tab = self.page_to_tab.get(page)
if widget and current_tab:
# Get project context
project_id, default_env_id = self._get_project_for_saved_request(link_to_saved)
# Update the tab metadata
current_tab.name = tab_name
current_tab.request = req
current_tab.saved_request_id = link_to_saved
current_tab.project_id = project_id
current_tab.selected_environment_id = default_env_id
current_tab.scripts = scripts
# Update widget with project context
widget.project_id = project_id
widget.selected_environment_id = default_env_id
if project_id and hasattr(widget, '_show_environment_selector'):
widget._show_environment_selector()
# Update the tab page title
page.set_title(tab_name)
# Load request into widget
widget._load_request(req)
# Load scripts into widget
if scripts:
widget.load_scripts(scripts)
# Set original scripts for change tracking
from .models import Scripts
widget.original_scripts = Scripts(
preprocessing=scripts.preprocessing,
postprocessing=scripts.postprocessing
)
else:
widget.original_scripts = None
if is_copy:
# This is a copy - mark as unsaved
current_tab.original_request = None
widget.original_request = None
widget.modified = True
else:
# This is linked to saved request
original = HttpRequest(
method=req.method,
url=req.url,
headers=req.headers.copy(),
body=req.body,
syntax=req.syntax
)
current_tab.original_request = original
widget.original_request = original
widget.modified = False
else:
# Current tab has changes or is not a "New Request"
# Create a new tab
project_id, default_env_id = self._get_project_for_saved_request(link_to_saved)
tab_id = self._create_new_tab(
name=tab_name,
request=req,
saved_request_id=link_to_saved,
project_id=project_id,
selected_environment_id=default_env_id,
scripts=scripts
)
# If it's a copy, clear the original_request to mark as unsaved
if is_copy:
new_tab = self.tab_manager.get_tab_by_id(tab_id)
if new_tab:
new_tab.original_request = None
new_tab.modified = True
# Also update the widget
page = self.tab_view.get_selected_page()
if page:
widget = self.page_to_widget.get(page)
if widget:
widget.original_request = None
widget.modified = True
def _on_delete_request(self, widget, saved_request, project):
"""Delete saved request with confirmation."""
dialog = Adw.AlertDialog()
dialog.set_heading("Delete Request?")
dialog.set_body(f"Delete '{saved_request.name}'? This cannot be undone.")
dialog.add_response("cancel", "Cancel")
dialog.add_response("delete", "Delete")
dialog.set_response_appearance("delete", Adw.ResponseAppearance.DESTRUCTIVE)
def on_response(dlg, response):
if response == "delete":
self.project_manager.delete_request(project.id, saved_request.id)
self._load_projects()
dialog.connect("response", on_response)
dialog.present(self)