382 lines
13 KiB
Python
382 lines
13 KiB
Python
# openapi_import_dialog.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 <https://www.gnu.org/licenses/>.
|
|
#
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
import gi
|
|
gi.require_version('Soup', '3.0')
|
|
from gi.repository import Adw, Gtk, GObject, GLib, Soup
|
|
from typing import List
|
|
|
|
from .openapi_importer import parse_openapi, build_http_request, OpenApiParseResult, OpenApiOperation
|
|
|
|
# Method → Adwaita accent CSS class
|
|
_METHOD_CLASS = {
|
|
'GET': 'accent',
|
|
'POST': 'success',
|
|
'PUT': 'warning',
|
|
'PATCH': 'warning',
|
|
'DELETE': 'error',
|
|
}
|
|
|
|
|
|
class OpenApiImportDialog(Adw.Dialog):
|
|
"""Two-step dialog: fetch OpenAPI URL → select operations → import."""
|
|
|
|
__gtype_name__ = 'OpenApiImportDialog'
|
|
|
|
__gsignals__ = {
|
|
'import-completed': (GObject.SIGNAL_RUN_FIRST, None, (str,)),
|
|
}
|
|
|
|
def __init__(self, project_manager, existing_projects):
|
|
super().__init__()
|
|
self.set_title("Import from OpenAPI")
|
|
self.set_content_width(580)
|
|
self.set_content_height(560)
|
|
|
|
self._pm = project_manager
|
|
self._existing_projects = list(existing_projects)
|
|
self._parse_result: OpenApiParseResult | None = None
|
|
self._op_checkboxes: List[tuple[OpenApiOperation, Gtk.CheckButton]] = []
|
|
self._soup = Soup.Session.new()
|
|
|
|
self._build_ui()
|
|
|
|
# ------------------------------------------------------------------ build
|
|
|
|
def _build_ui(self):
|
|
tv = Adw.ToolbarView()
|
|
tv.add_top_bar(Adw.HeaderBar())
|
|
|
|
self._stack = Gtk.Stack()
|
|
self._stack.set_transition_type(Gtk.StackTransitionType.SLIDE_LEFT_RIGHT)
|
|
self._stack.set_transition_duration(200)
|
|
self._stack.add_named(self._make_url_page(), "url")
|
|
self._stack.add_named(self._make_ops_page(), "ops")
|
|
tv.set_content(self._stack)
|
|
|
|
ab = Gtk.ActionBar()
|
|
|
|
self._back_btn = Gtk.Button(label="Back")
|
|
self._back_btn.connect("clicked", lambda _: self._show_url_page())
|
|
ab.pack_start(self._back_btn)
|
|
|
|
self._fetch_btn = Gtk.Button(label="Fetch spec")
|
|
self._fetch_btn.add_css_class("suggested-action")
|
|
self._fetch_btn.connect("clicked", self._on_fetch)
|
|
ab.pack_end(self._fetch_btn)
|
|
|
|
self._import_btn = Gtk.Button(label="Import")
|
|
self._import_btn.add_css_class("suggested-action")
|
|
self._import_btn.connect("clicked", self._on_import)
|
|
ab.pack_end(self._import_btn)
|
|
|
|
tv.add_bottom_bar(ab)
|
|
self.set_child(tv)
|
|
self._show_url_page()
|
|
|
|
def _make_url_page(self) -> Gtk.Widget:
|
|
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=16)
|
|
box.set_valign(Gtk.Align.CENTER)
|
|
box.set_vexpand(True)
|
|
box.set_margin_top(32)
|
|
box.set_margin_bottom(24)
|
|
box.set_margin_start(24)
|
|
box.set_margin_end(24)
|
|
|
|
icon = Gtk.Image.new_from_icon_name("openapi-import-symbolic")
|
|
icon.set_pixel_size(64)
|
|
icon.add_css_class("dim-label")
|
|
box.append(icon)
|
|
|
|
title = Gtk.Label(label="Import from OpenAPI")
|
|
title.add_css_class("title-2")
|
|
box.append(title)
|
|
|
|
subtitle = Gtk.Label(
|
|
label="Enter the URL of an OpenAPI 2.0 (Swagger) or 3.x specification (JSON or YAML)"
|
|
)
|
|
subtitle.add_css_class("dim-label")
|
|
subtitle.set_wrap(True)
|
|
subtitle.set_justify(Gtk.Justification.CENTER)
|
|
box.append(subtitle)
|
|
|
|
grp = Adw.PreferencesGroup()
|
|
self._url_row = Adw.EntryRow()
|
|
self._url_row.set_title("Spec URL")
|
|
self._url_row.set_input_purpose(Gtk.InputPurpose.URL)
|
|
self._url_row.connect("entry-activated", lambda _: self._on_fetch(None))
|
|
grp.add(self._url_row)
|
|
box.append(grp)
|
|
|
|
self._err_label = Gtk.Label()
|
|
self._err_label.add_css_class("error")
|
|
self._err_label.set_wrap(True)
|
|
self._err_label.set_visible(False)
|
|
box.append(self._err_label)
|
|
|
|
self._spinner = Gtk.Spinner()
|
|
self._spinner.set_size_request(32, 32)
|
|
self._spinner.set_halign(Gtk.Align.CENTER)
|
|
self._spinner.set_visible(False)
|
|
box.append(self._spinner)
|
|
|
|
return box
|
|
|
|
def _make_ops_page(self) -> Gtk.Widget:
|
|
outer = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)
|
|
outer.set_margin_top(14)
|
|
outer.set_margin_bottom(14)
|
|
outer.set_margin_start(16)
|
|
outer.set_margin_end(16)
|
|
|
|
# API info
|
|
api_grp = Adw.PreferencesGroup()
|
|
api_grp.set_title("API")
|
|
self._api_url_row = Adw.ActionRow()
|
|
self._api_url_row.set_title("Base URL")
|
|
api_grp.add(self._api_url_row)
|
|
outer.append(api_grp)
|
|
|
|
# Operations header
|
|
ops_hdr = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
|
|
ops_hdr.set_margin_top(2)
|
|
ops_lbl = Gtk.Label(label="Operations")
|
|
ops_lbl.add_css_class("title-4")
|
|
ops_lbl.set_hexpand(True)
|
|
ops_lbl.set_xalign(0)
|
|
ops_hdr.append(ops_lbl)
|
|
self._sel_all_btn = Gtk.Button(label="Deselect All")
|
|
self._sel_all_btn.add_css_class("flat")
|
|
self._sel_all_btn.connect("clicked", self._on_select_all)
|
|
ops_hdr.append(self._sel_all_btn)
|
|
outer.append(ops_hdr)
|
|
|
|
# Scrollable operations list
|
|
scroll = Gtk.ScrolledWindow()
|
|
scroll.set_vexpand(True)
|
|
scroll.set_min_content_height(80)
|
|
self._ops_list = Gtk.ListBox()
|
|
self._ops_list.add_css_class("boxed-list")
|
|
self._ops_list.set_selection_mode(Gtk.SelectionMode.NONE)
|
|
scroll.set_child(self._ops_list)
|
|
outer.append(scroll)
|
|
|
|
# Import target
|
|
tgt_grp = Adw.PreferencesGroup()
|
|
tgt_grp.set_title("Import to")
|
|
tgt_grp.set_margin_top(2)
|
|
|
|
new_row = Adw.ActionRow()
|
|
new_row.set_title("Create new project")
|
|
self._new_radio = Gtk.CheckButton()
|
|
self._new_radio.set_active(True)
|
|
new_row.add_prefix(self._new_radio)
|
|
new_row.set_activatable_widget(self._new_radio)
|
|
self._new_name_entry = Gtk.Entry()
|
|
self._new_name_entry.set_placeholder_text("Project name")
|
|
self._new_name_entry.set_hexpand(True)
|
|
self._new_name_entry.set_valign(Gtk.Align.CENTER)
|
|
new_row.add_suffix(self._new_name_entry)
|
|
tgt_grp.add(new_row)
|
|
|
|
if self._existing_projects:
|
|
exist_row = Adw.ActionRow()
|
|
exist_row.set_title("Add to existing project")
|
|
self._exist_radio = Gtk.CheckButton()
|
|
self._exist_radio.set_group(self._new_radio)
|
|
exist_row.add_prefix(self._exist_radio)
|
|
exist_row.set_activatable_widget(self._exist_radio)
|
|
proj_list = Gtk.StringList.new([p.name for p in self._existing_projects])
|
|
self._proj_dd = Gtk.DropDown(model=proj_list)
|
|
self._proj_dd.set_valign(Gtk.Align.CENTER)
|
|
self._proj_dd.set_sensitive(False)
|
|
exist_row.add_suffix(self._proj_dd)
|
|
tgt_grp.add(exist_row)
|
|
self._new_radio.connect("toggled", self._on_target_toggled)
|
|
else:
|
|
self._exist_radio = None
|
|
self._proj_dd = None
|
|
|
|
outer.append(tgt_grp)
|
|
return outer
|
|
|
|
# ----------------------------------------------------------- page control
|
|
|
|
def _show_url_page(self):
|
|
self._stack.set_visible_child_name("url")
|
|
self._back_btn.set_visible(False)
|
|
self._fetch_btn.set_visible(True)
|
|
self._import_btn.set_visible(False)
|
|
|
|
def _show_ops_page(self):
|
|
self._stack.set_visible_child_name("ops")
|
|
self._back_btn.set_visible(True)
|
|
self._fetch_btn.set_visible(False)
|
|
self._import_btn.set_visible(True)
|
|
|
|
# --------------------------------------------------------- event handlers
|
|
|
|
def _on_target_toggled(self, _radio):
|
|
is_new = self._new_radio.get_active()
|
|
self._new_name_entry.set_sensitive(is_new)
|
|
if self._proj_dd:
|
|
self._proj_dd.set_sensitive(not is_new)
|
|
|
|
def _on_select_all(self, _btn):
|
|
all_on = all(cb.get_active() for _, cb in self._op_checkboxes)
|
|
new_state = not all_on
|
|
for _, cb in self._op_checkboxes:
|
|
cb.set_active(new_state)
|
|
self._sel_all_btn.set_label("Deselect All" if new_state else "Select All")
|
|
|
|
def _on_fetch(self, _btn):
|
|
url = self._url_row.get_text().strip()
|
|
if not url:
|
|
self._set_err("Please enter a spec URL")
|
|
return
|
|
if not url.startswith(('http://', 'https://')):
|
|
self._set_err("URL must start with http:// or https://")
|
|
return
|
|
|
|
self._err_label.set_visible(False)
|
|
self._spinner.set_visible(True)
|
|
self._spinner.start()
|
|
self._fetch_btn.set_sensitive(False)
|
|
|
|
try:
|
|
msg = Soup.Message.new("GET", url)
|
|
except Exception as e:
|
|
self._fetch_failed(f"Invalid URL: {e}")
|
|
return
|
|
|
|
msg.get_request_headers().append(
|
|
"Accept", "application/json, application/yaml, text/yaml, */*"
|
|
)
|
|
self._soup.send_and_read_async(msg, GLib.PRIORITY_DEFAULT, None,
|
|
self._on_downloaded, msg)
|
|
|
|
def _on_downloaded(self, session, async_result, msg):
|
|
self._spinner.stop()
|
|
self._spinner.set_visible(False)
|
|
self._fetch_btn.set_sensitive(True)
|
|
|
|
try:
|
|
bytes_data = session.send_and_read_finish(async_result)
|
|
except GLib.Error as e:
|
|
self._set_err(f"Download failed: {e.message}")
|
|
return
|
|
|
|
status = msg.get_status()
|
|
if not (200 <= status < 300):
|
|
self._set_err(f"Server returned HTTP {status}")
|
|
return
|
|
|
|
content = bytes_data.get_data().decode('utf-8', errors='replace')
|
|
parsed = parse_openapi(content)
|
|
|
|
if parsed.error:
|
|
self._set_err(parsed.error)
|
|
return
|
|
|
|
self._parse_result = parsed
|
|
self._populate_ops_page(parsed)
|
|
self._show_ops_page()
|
|
|
|
def _on_import(self, _btn):
|
|
if not self._parse_result:
|
|
return
|
|
|
|
selected = [op for op, cb in self._op_checkboxes if cb.get_active()]
|
|
if not selected:
|
|
return
|
|
|
|
if self._new_radio.get_active():
|
|
name = self._new_name_entry.get_text().strip()
|
|
if not name:
|
|
return
|
|
project = self._pm.add_project(name)
|
|
project_id = project.id
|
|
elif self._exist_radio and self._exist_radio.get_active() and self._proj_dd:
|
|
idx = self._proj_dd.get_selected()
|
|
if idx >= len(self._existing_projects):
|
|
return
|
|
project_id = self._existing_projects[idx].id
|
|
else:
|
|
return
|
|
|
|
for op in selected:
|
|
http_req = build_http_request(op)
|
|
self._pm.add_request(project_id, op.name, http_req)
|
|
|
|
self.emit('import-completed', project_id)
|
|
self.close()
|
|
|
|
# ----------------------------------------------------------------- helpers
|
|
|
|
def _set_err(self, msg: str):
|
|
self._err_label.set_text(msg)
|
|
self._err_label.set_visible(True)
|
|
|
|
def _fetch_failed(self, msg: str):
|
|
self._spinner.stop()
|
|
self._spinner.set_visible(False)
|
|
self._fetch_btn.set_sensitive(True)
|
|
self._set_err(msg)
|
|
|
|
def _populate_ops_page(self, result: OpenApiParseResult):
|
|
self._api_url_row.set_subtitle(result.base_url or "Not specified")
|
|
|
|
while child := self._ops_list.get_first_child():
|
|
self._ops_list.remove(child)
|
|
self._op_checkboxes.clear()
|
|
|
|
for op in sorted(result.operations, key=lambda o: o.name.lower()):
|
|
row = Adw.ActionRow()
|
|
row.set_title(op.name)
|
|
|
|
# Subtitle: "METHOD /path" or description
|
|
if op.name == f'{op.method} {op.path}':
|
|
if op.description:
|
|
row.set_subtitle(op.description)
|
|
else:
|
|
sub = f'{op.method} {op.path}'
|
|
if op.description:
|
|
sub += f' — {op.description}'
|
|
row.set_subtitle(sub)
|
|
|
|
# Method badge suffix
|
|
badge = Gtk.Label(label=op.method)
|
|
badge.add_css_class("caption")
|
|
badge.add_css_class("monospace")
|
|
css_cls = _METHOD_CLASS.get(op.method, 'dim-label')
|
|
badge.add_css_class(css_cls)
|
|
badge.set_valign(Gtk.Align.CENTER)
|
|
row.add_suffix(badge)
|
|
|
|
cb = Gtk.CheckButton()
|
|
cb.set_active(True)
|
|
row.add_prefix(cb)
|
|
row.set_activatable_widget(cb)
|
|
|
|
self._ops_list.append(row)
|
|
self._op_checkboxes.append((op, cb))
|
|
|
|
self._new_name_entry.set_text(result.api_title)
|
|
self._sel_all_btn.set_label("Deselect All")
|