Add OpenAPI/Swagger import (2.0 + 3.x, JSON body templates)
This commit is contained in:
parent
7f921fb08d
commit
fef8dec193
15
data/icons/hicolor/symbolic/apps/openapi-import-symbolic.svg
Normal file
15
data/icons/hicolor/symbolic/apps/openapi-import-symbolic.svg
Normal file
@ -0,0 +1,15 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<svg xmlns="http://www.w3.org/2000/svg" height="16px" viewBox="0 0 16 16" width="16px">
|
||||
<g fill="none" stroke="#222222" stroke-linecap="round" stroke-linejoin="round">
|
||||
<!-- Rounded rectangle frame -->
|
||||
<rect x="1.5" y="1.5" width="13" height="13" rx="2.25" ry="2.25" stroke-width="1.5"/>
|
||||
<!-- Left curly brace { -->
|
||||
<path stroke-width="1.6"
|
||||
d="M 7 4.5 C 6.2 4.5 5.75 4.9 5.75 5.6 L 5.75 7 C 5.75 7.7 5.3 8 4.75 8
|
||||
C 5.3 8 5.75 8.3 5.75 9 L 5.75 10.4 C 5.75 11.1 6.2 11.5 7 11.5"/>
|
||||
<!-- Right curly brace } -->
|
||||
<path stroke-width="1.6"
|
||||
d="M 9 4.5 C 9.8 4.5 10.25 4.9 10.25 5.6 L 10.25 7 C 10.25 7.7 10.7 8 11.25 8
|
||||
C 10.7 8 10.25 8.3 10.25 9 L 10.25 10.4 C 10.25 11.1 9.8 11.5 9 11.5"/>
|
||||
</g>
|
||||
</svg>
|
||||
|
After Width: | Height: | Size: 808 B |
@ -58,6 +58,16 @@
|
||||
</style>
|
||||
</object>
|
||||
</child>
|
||||
<child type="end">
|
||||
<object class="GtkButton" id="import_openapi_button">
|
||||
<property name="icon-name">openapi-import-symbolic</property>
|
||||
<property name="tooltip-text">Import from OpenAPI / Swagger</property>
|
||||
<signal name="clicked" handler="on_import_openapi_clicked"/>
|
||||
<style>
|
||||
<class name="flat"/>
|
||||
</style>
|
||||
</object>
|
||||
</child>
|
||||
</object>
|
||||
</child>
|
||||
|
||||
|
||||
@ -46,6 +46,8 @@ roster_sources = [
|
||||
'script_executor.py',
|
||||
'wsdl_importer.py',
|
||||
'wsdl_import_dialog.py',
|
||||
'openapi_importer.py',
|
||||
'openapi_import_dialog.py',
|
||||
]
|
||||
|
||||
install_data(roster_sources, install_dir: moduledir)
|
||||
|
||||
381
src/openapi_import_dialog.py
Normal file
381
src/openapi_import_dialog.py
Normal file
@ -0,0 +1,381 @@
|
||||
# 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")
|
||||
395
src/openapi_importer.py
Normal file
395
src/openapi_importer.py
Normal file
@ -0,0 +1,395 @@
|
||||
# openapi_importer.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 json
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Dict, List, Optional, Any
|
||||
|
||||
try:
|
||||
import yaml as _yaml
|
||||
_YAML_AVAILABLE = True
|
||||
except ImportError:
|
||||
_YAML_AVAILABLE = False
|
||||
|
||||
# JSON Schema type → representative default value
|
||||
_TYPE_DEFAULTS: Dict[str, Any] = {
|
||||
'string': '',
|
||||
'integer': 0,
|
||||
'number': 0.0,
|
||||
'boolean': False,
|
||||
'null': None,
|
||||
'array': [],
|
||||
'object': {},
|
||||
}
|
||||
|
||||
_FORMAT_DEFAULTS: Dict[str, Any] = {
|
||||
'date-time': '2024-01-01T00:00:00Z',
|
||||
'date': '2024-01-01',
|
||||
'time': '00:00:00',
|
||||
'email': 'user@example.com',
|
||||
'uuid': '00000000-0000-0000-0000-000000000000',
|
||||
'uri': 'https://example.com',
|
||||
'hostname': 'example.com',
|
||||
'ipv4': '0.0.0.0',
|
||||
'int32': 0,
|
||||
'int64': 0,
|
||||
'float': 0.0,
|
||||
'double': 0.0,
|
||||
'byte': '',
|
||||
'binary': '',
|
||||
'password': '',
|
||||
}
|
||||
|
||||
HTTP_METHODS = frozenset({'get', 'post', 'put', 'delete', 'patch', 'head', 'options'})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Public data classes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@dataclass
|
||||
class OpenApiOperation:
|
||||
name: str # operationId or "METHOD /path"
|
||||
method: str # uppercase: GET, POST, …
|
||||
path: str
|
||||
url: str
|
||||
headers: Dict[str, str]
|
||||
body: str # JSON template or ''
|
||||
description: str # summary / description
|
||||
|
||||
|
||||
@dataclass
|
||||
class OpenApiParseResult:
|
||||
api_title: str
|
||||
base_url: str
|
||||
operations: List[OpenApiOperation] = field(default_factory=list)
|
||||
error: Optional[str] = None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Public entry point
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def parse_openapi(content: str) -> OpenApiParseResult:
|
||||
"""Parse OpenAPI 2.0 (Swagger) or 3.x from JSON or YAML."""
|
||||
data = _load(content)
|
||||
if isinstance(data, str): # error message
|
||||
return OpenApiParseResult(api_title='', base_url='', error=data)
|
||||
|
||||
if not isinstance(data, dict):
|
||||
return OpenApiParseResult(api_title='', base_url='',
|
||||
error='Not a valid OpenAPI document')
|
||||
|
||||
if 'swagger' in data:
|
||||
return _parse_v2(data)
|
||||
elif 'openapi' in data:
|
||||
return _parse_v3(data)
|
||||
else:
|
||||
return OpenApiParseResult(
|
||||
api_title='', base_url='',
|
||||
error='Not a recognized OpenAPI document (missing "swagger" or "openapi" key)'
|
||||
)
|
||||
|
||||
|
||||
def build_http_request(operation: OpenApiOperation):
|
||||
"""Convert an OpenApiOperation into an HttpRequest."""
|
||||
from .models import HttpRequest
|
||||
|
||||
syntax = 'JSON' if operation.body.lstrip().startswith('{') else 'RAW'
|
||||
return HttpRequest(
|
||||
method=operation.method,
|
||||
url=operation.url,
|
||||
headers=operation.headers,
|
||||
body=operation.body,
|
||||
syntax=syntax,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Loader (JSON + optional YAML)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _load(content: str):
|
||||
"""Return parsed dict or an error string."""
|
||||
content = content.strip()
|
||||
# Try JSON first
|
||||
try:
|
||||
return json.loads(content)
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
# Fall back to YAML
|
||||
if _YAML_AVAILABLE:
|
||||
try:
|
||||
return _yaml.safe_load(content)
|
||||
except Exception as e:
|
||||
return f'Invalid YAML: {e}'
|
||||
|
||||
# Looks like YAML but library not available
|
||||
if not content.startswith('{'):
|
||||
return ('YAML format detected but PyYAML is not available. '
|
||||
'Please export the spec as JSON.')
|
||||
return 'Invalid JSON'
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# OpenAPI 2.0 (Swagger)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _parse_v2(data: dict) -> OpenApiParseResult:
|
||||
info = data.get('info', {})
|
||||
title = info.get('title', 'API')
|
||||
|
||||
schemes = data.get('schemes', ['https'])
|
||||
scheme = 'https' if 'https' in schemes else (schemes[0] if schemes else 'https')
|
||||
host = data.get('host', 'localhost')
|
||||
base_path = data.get('basePath', '/').rstrip('/')
|
||||
base_url = f'{scheme}://{host}{base_path}'
|
||||
|
||||
definitions = data.get('definitions', {})
|
||||
paths = data.get('paths', {})
|
||||
|
||||
operations = []
|
||||
for path, path_item in paths.items():
|
||||
if not isinstance(path_item, dict):
|
||||
continue
|
||||
path_params = path_item.get('parameters', [])
|
||||
for method, op_data in path_item.items():
|
||||
if method not in HTTP_METHODS or not isinstance(op_data, dict):
|
||||
continue
|
||||
params = _resolve_params(path_params + op_data.get('parameters', []),
|
||||
data.get('parameters', {}))
|
||||
operations.append(_make_op_v2(
|
||||
method.upper(), path, base_url, op_data, params, definitions
|
||||
))
|
||||
|
||||
if not operations:
|
||||
return OpenApiParseResult(api_title=title, base_url=base_url,
|
||||
error='No operations found in this document')
|
||||
|
||||
return OpenApiParseResult(api_title=title, base_url=base_url, operations=operations)
|
||||
|
||||
|
||||
def _make_op_v2(method, path, base_url, op_data, params, definitions) -> OpenApiOperation:
|
||||
op_id = op_data.get('operationId', '')
|
||||
description = op_data.get('summary') or op_data.get('description', '')
|
||||
name = op_id or f'{method} {path}'
|
||||
|
||||
url = base_url + path
|
||||
url = _append_required_query(url, params)
|
||||
|
||||
headers: Dict[str, str] = {}
|
||||
body = ''
|
||||
|
||||
if method in ('POST', 'PUT', 'PATCH', 'DELETE'):
|
||||
body_params = [p for p in params if p.get('in') == 'body']
|
||||
if body_params:
|
||||
consumes = op_data.get('consumes', ['application/json'])
|
||||
if any('json' in ct for ct in consumes):
|
||||
headers['Content-Type'] = 'application/json'
|
||||
schema = body_params[0].get('schema', {})
|
||||
body = _schema_to_json(schema, definitions)
|
||||
else:
|
||||
headers['Content-Type'] = consumes[0] if consumes else 'application/octet-stream'
|
||||
|
||||
form_params = [p for p in params if p.get('in') == 'formData']
|
||||
if form_params and not body:
|
||||
headers['Content-Type'] = 'application/x-www-form-urlencoded'
|
||||
body = '&'.join(f"{p['name']}=" for p in form_params)
|
||||
|
||||
produces = op_data.get('produces', ['application/json'])
|
||||
if any('json' in ct for ct in produces):
|
||||
headers.setdefault('Accept', 'application/json')
|
||||
|
||||
return OpenApiOperation(
|
||||
name=name, method=method, path=path,
|
||||
url=url, headers=headers, body=body, description=description,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# OpenAPI 3.x
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _parse_v3(data: dict) -> OpenApiParseResult:
|
||||
info = data.get('info', {})
|
||||
title = info.get('title', 'API')
|
||||
|
||||
servers = data.get('servers') or [{}]
|
||||
base_url = servers[0].get('url', '').rstrip('/')
|
||||
|
||||
components = data.get('components', {})
|
||||
schemas = components.get('schemas', {})
|
||||
param_defs = components.get('parameters', {})
|
||||
paths = data.get('paths', {})
|
||||
|
||||
operations = []
|
||||
for path, path_item in paths.items():
|
||||
if not isinstance(path_item, dict):
|
||||
continue
|
||||
path_params = _resolve_params(path_item.get('parameters', []), param_defs)
|
||||
for method, op_data in path_item.items():
|
||||
if method not in HTTP_METHODS or not isinstance(op_data, dict):
|
||||
continue
|
||||
params = _resolve_params(path_params + op_data.get('parameters', []),
|
||||
param_defs)
|
||||
operations.append(_make_op_v3(
|
||||
method.upper(), path, base_url, op_data, params, schemas
|
||||
))
|
||||
|
||||
if not operations:
|
||||
return OpenApiParseResult(api_title=title, base_url=base_url,
|
||||
error='No operations found in this document')
|
||||
|
||||
return OpenApiParseResult(api_title=title, base_url=base_url, operations=operations)
|
||||
|
||||
|
||||
def _make_op_v3(method, path, base_url, op_data, params, schemas) -> OpenApiOperation:
|
||||
op_id = op_data.get('operationId', '')
|
||||
description = op_data.get('summary') or op_data.get('description', '')
|
||||
name = op_id or f'{method} {path}'
|
||||
|
||||
url = base_url + path
|
||||
url = _append_required_query(url, params)
|
||||
|
||||
headers: Dict[str, str] = {}
|
||||
body = ''
|
||||
|
||||
req_body = op_data.get('requestBody', {})
|
||||
if req_body and method in ('POST', 'PUT', 'PATCH', 'DELETE'):
|
||||
content = req_body.get('content', {})
|
||||
# Prefer application/json
|
||||
json_ct = next(
|
||||
(ct for ct in content if 'json' in ct),
|
||||
None
|
||||
)
|
||||
if json_ct:
|
||||
headers['Content-Type'] = json_ct
|
||||
schema = content[json_ct].get('schema', {})
|
||||
body = _schema_to_json(schema, schemas)
|
||||
elif content:
|
||||
ct = next(iter(content))
|
||||
headers['Content-Type'] = ct
|
||||
if 'form' in ct:
|
||||
schema = content[ct].get('schema', {})
|
||||
props = schema.get('properties', {})
|
||||
body = '&'.join(f'{k}=' for k in props)
|
||||
|
||||
headers.setdefault('Accept', 'application/json')
|
||||
|
||||
return OpenApiOperation(
|
||||
name=name, method=method, path=path,
|
||||
url=url, headers=headers, body=body, description=description,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _resolve_params(params: list, param_defs: dict) -> list:
|
||||
"""Resolve $ref entries in a parameter list."""
|
||||
resolved = []
|
||||
for p in params:
|
||||
if isinstance(p, dict) and '$ref' in p:
|
||||
ref_name = p['$ref'].split('/')[-1]
|
||||
p = param_defs.get(ref_name, p)
|
||||
resolved.append(p)
|
||||
return resolved
|
||||
|
||||
|
||||
def _append_required_query(url: str, params: list) -> str:
|
||||
required = [p['name'] for p in params
|
||||
if p.get('in') == 'query' and p.get('required', False)]
|
||||
if required:
|
||||
url += '?' + '&'.join(f'{n}=' for n in required)
|
||||
return url
|
||||
|
||||
|
||||
def _schema_to_json(schema: dict, defs: dict, depth: int = 0) -> str:
|
||||
"""Return a pretty-printed JSON template string for the given schema."""
|
||||
try:
|
||||
value = _schema_value(schema, defs, depth)
|
||||
return json.dumps(value, indent=2, ensure_ascii=False)
|
||||
except Exception:
|
||||
return ''
|
||||
|
||||
|
||||
def _schema_value(schema: dict, defs: dict, depth: int = 0) -> Any:
|
||||
"""Recursively build a representative value from a JSON Schema."""
|
||||
if depth > 4:
|
||||
return {}
|
||||
|
||||
# Resolve $ref
|
||||
if '$ref' in schema:
|
||||
ref_name = schema['$ref'].split('/')[-1]
|
||||
schema = defs.get(ref_name, {})
|
||||
if not schema:
|
||||
return {}
|
||||
|
||||
# Combiners: allOf / anyOf / oneOf
|
||||
for combiner in ('allOf', 'anyOf', 'oneOf'):
|
||||
if combiner in schema:
|
||||
merged: Dict[str, Any] = {}
|
||||
for sub in schema[combiner]:
|
||||
if '$ref' in sub:
|
||||
sub = defs.get(sub['$ref'].split('/')[-1], {})
|
||||
result = _schema_value(sub, defs, depth)
|
||||
if isinstance(result, dict):
|
||||
merged.update(result)
|
||||
if merged:
|
||||
return merged
|
||||
# Non-object combiners: return first sub-value
|
||||
first = schema[combiner][0] if schema[combiner] else {}
|
||||
return _schema_value(first, defs, depth)
|
||||
|
||||
type_ = schema.get('type', 'object' if 'properties' in schema else None)
|
||||
|
||||
if type_ == 'object' or 'properties' in schema:
|
||||
result = {}
|
||||
required = schema.get('required', [])
|
||||
for prop, prop_schema in schema.get('properties', {}).items():
|
||||
result[prop] = _schema_value(prop_schema, defs, depth + 1)
|
||||
if not result and 'additionalProperties' in schema:
|
||||
add = schema['additionalProperties']
|
||||
if isinstance(add, dict):
|
||||
result['key'] = _schema_value(add, defs, depth + 1)
|
||||
return result
|
||||
|
||||
if type_ == 'array':
|
||||
items = schema.get('items', {})
|
||||
return [_schema_value(items, defs, depth + 1)] if items else []
|
||||
|
||||
if type_ == 'string':
|
||||
enum = schema.get('enum')
|
||||
if enum:
|
||||
return enum[0]
|
||||
fmt = schema.get('format', '')
|
||||
return _FORMAT_DEFAULTS.get(fmt, '')
|
||||
|
||||
if type_ in ('integer', 'number'):
|
||||
return schema.get('minimum', 0)
|
||||
|
||||
if type_ == 'boolean':
|
||||
return False
|
||||
|
||||
if type_ == 'null':
|
||||
return None
|
||||
|
||||
return _TYPE_DEFAULTS.get(type_, {})
|
||||
@ -10,6 +10,7 @@
|
||||
<file alias="icons/export-symbolic.svg">../data/icons/hicolor/symbolic/apps/export-symbolic.svg</file>
|
||||
<file alias="icons/wsdl-import-symbolic.svg">../data/icons/hicolor/symbolic/apps/wsdl-import-symbolic.svg</file>
|
||||
<file alias="icons/papyrus-vertical-symbolic.svg">../data/icons/hicolor/symbolic/apps/papyrus-vertical-symbolic.svg</file>
|
||||
<file alias="icons/openapi-import-symbolic.svg">../data/icons/hicolor/symbolic/apps/openapi-import-symbolic.svg</file>
|
||||
<file preprocess="xml-stripblanks">widgets/header-row.ui</file>
|
||||
<file preprocess="xml-stripblanks">widgets/history-item.ui</file>
|
||||
<file preprocess="xml-stripblanks">widgets/project-item.ui</file>
|
||||
|
||||
@ -37,6 +37,7 @@ from .tab_manager import TabManager
|
||||
from .icon_picker_dialog import IconPickerDialog
|
||||
from .environments_dialog import EnvironmentsDialog
|
||||
from .wsdl_import_dialog import WsdlImportDialog
|
||||
from .openapi_import_dialog import OpenApiImportDialog
|
||||
from .request_tab_widget import RequestTabWidget
|
||||
from .widgets.history_item import HistoryItem
|
||||
from .widgets.project_item import ProjectItem
|
||||
@ -69,6 +70,7 @@ class RosterWindow(Adw.ApplicationWindow):
|
||||
add_project_button = Gtk.Template.Child()
|
||||
|
||||
import_wsdl_button = Gtk.Template.Child()
|
||||
import_openapi_button = Gtk.Template.Child()
|
||||
|
||||
# History (hidden but kept for compatibility)
|
||||
history_listbox = Gtk.Template.Child()
|
||||
@ -1421,6 +1423,23 @@ class RosterWindow(Adw.ApplicationWindow):
|
||||
dialog.connect('import-completed', self._on_wsdl_import_completed)
|
||||
dialog.present(self)
|
||||
|
||||
@Gtk.Template.Callback()
|
||||
def on_import_openapi_clicked(self, button):
|
||||
"""Open the OpenAPI import dialog."""
|
||||
projects = self.project_manager.load_projects()
|
||||
dialog = OpenApiImportDialog(self.project_manager, projects)
|
||||
dialog.connect('import-completed', self._on_openapi_import_completed)
|
||||
dialog.present(self)
|
||||
|
||||
def _on_openapi_import_completed(self, dialog, project_id: str):
|
||||
"""Refresh sidebar after a successful OpenAPI import."""
|
||||
self._load_projects()
|
||||
projects = self.project_manager.load_projects()
|
||||
for p in projects:
|
||||
if p.id == project_id:
|
||||
self._show_toast(f"Imported {len(p.requests)} operation(s) into '{p.name}'")
|
||||
break
|
||||
|
||||
def _on_wsdl_import_completed(self, dialog, project_id: str):
|
||||
"""Refresh sidebar after a successful WSDL import."""
|
||||
self._load_projects()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user