@samitouri / QOSami_Agents_Zero-1 / commits / e5bd449d

refactor(api): Extract plugin action handlers into separate methods

Refactor the Plugins API handler by extracting inline action handlers into dedicated private methods. This improves code organization and readability by separating each action (get_config, get_toggle_status, list_configs, delete_config, delete_plugin, get_default_config, save_config, toggle_plugin, get_doc, run_init_script, get_init_exec) into its own method while maintaining the same functionality.

frdel committed Mar 12, 2026 at 07:55 UTC e5bd449d660ce2d68b2dcf74482d144a246907cd
1 file changed +284 -241
api/plugins.py
+284 -241
@@ -19,267 +19,310 @@ class Plugins(ApiHandler):
19
20 # Accept legacy aliases during migration.
21 if action == "get_config":
22 - plugin_name = input.get("plugin_name", "")
23 - project_name = input.get("project_name", "")
24 - agent_profile = input.get("agent_profile", "")
25 - if not plugin_name:
26 - return Response(status=400, response="Missing plugin_name")
27 -
28 - result = plugins.find_plugin_assets(
29 - plugins.CONFIG_FILE_NAME,
30 - plugin_name=plugin_name,
31 - project_name=project_name,
32 - agent_profile=agent_profile,
33 - only_first=True,
34 - )
35 - if result:
36 - entry = result[0]
37 - path = entry.get("path", "")
38 - settings = files.read_file_json(path) if path else {}
39 - loaded_project_name = entry.get("project_name", "")
40 - loaded_agent_profile = entry.get("agent_profile", "")
41 - else:
42 - settings = plugins.get_plugin_config(plugin_name, agent=None) or {}
43 - default_path = files.get_abs_path(
44 - plugins.find_plugin_dir(plugin_name), plugins.CONFIG_DEFAULT_FILE_NAME
45 - )
46 - path = default_path if files.exists(default_path) else ""
47 - loaded_project_name = ""
48 - loaded_agent_profile = ""
49 -
50 - return {
51 - "ok": True,
52 - "loaded_path": path,
53 - "loaded_project_name": loaded_project_name,
54 - "loaded_agent_profile": loaded_agent_profile,
55 - "data": settings,
56 - }
22 + return self._get_config(input)
23
24 if action == "get_toggle_status":
59 - plugin_name = input.get("plugin_name", "")
60 - project_name = input.get("project_name", "")
61 - agent_profile = input.get("agent_profile", "")
62 - if not plugin_name:
63 - return Response(status=400, response="Missing plugin_name")
64 -
65 - meta = plugins.get_plugin_meta(plugin_name)
66 - if not meta:
67 - return Response(status=404, response="Plugin not found")
68 -
69 - if meta.always_enabled:
70 - return {
71 - "ok": True,
72 - "status": "enabled",
73 - "loaded_project_name": project_name,
74 - "loaded_agent_profile": agent_profile,
75 - "loaded_path": "",
76 - }
77 -
78 - result = plugins.find_plugin_assets(
79 - plugins.TOGGLE_FILE_PATTERN,
80 - plugin_name=plugin_name,
81 - project_name=project_name,
82 - agent_profile=agent_profile,
83 - only_first=True,
84 - )
85 -
86 - if result:
87 - entry = result[0]
88 - path = entry.get("path", "")
89 - status = "enabled" if path.endswith(plugins.ENABLED_FILE_NAME) else "disabled"
90 - return {
91 - "ok": True,
92 - "status": status,
93 - "loaded_project_name": entry.get("project_name", ""),
94 - "loaded_agent_profile": entry.get("agent_profile", ""),
95 - "loaded_path": path,
96 - }
97 -
98 - return {
99 - "ok": True,
100 - "status": "enabled",
101 - "loaded_project_name": "",
102 - "loaded_agent_profile": "",
103 - "loaded_path": "",
104 - }
25 + return self._get_toggle_status(input)
26
27 if action == "list_configs":
107 - plugin_name = input.get("plugin_name", "")
108 - asset_type = input.get("asset_type", "config")
109 - if not plugin_name:
110 - return Response(status=400, response="Missing plugin_name")
111 -
112 - configs = plugins.find_plugin_assets(
113 - plugins.CONFIG_FILE_NAME if asset_type == "config" else plugins.TOGGLE_FILE_PATTERN,
114 - plugin_name=plugin_name,
115 - project_name="*",
116 - agent_profile="*",
117 - only_first=False,
118 - )
119 -
120 - return {"ok": True, "data": configs}
28 + return self._list_configs(input)
29
30 if action == "delete_config":
123 - plugin_name = input.get("plugin_name", "")
124 - path = input.get("path", "")
125 - if not plugin_name:
126 - return Response(status=400, response="Missing plugin_name")
127 - if not path:
128 - return Response(status=400, response="Missing path")
129 -
130 - configs = plugins.find_plugin_assets(
131 - plugins.CONFIG_FILE_NAME,
132 - plugin_name=plugin_name,
133 - project_name="*",
134 - agent_profile="*",
135 - only_first=False,
136 - )
137 - toggles = plugins.find_plugin_assets(
138 - plugins.TOGGLE_FILE_PATTERN,
139 - plugin_name=plugin_name,
140 - project_name="*",
141 - agent_profile="*",
142 - only_first=False,
143 - )
144 - allowed_paths = {c.get("path", "") for c in configs + toggles}
145 - if path not in allowed_paths:
146 - return Response(status=400, response="Invalid path")
147 -
148 - if not files.exists(path):
149 - return {"ok": True}
150 -
151 - try:
152 - os.remove(path)
153 - except Exception as e:
154 - return Response(status=500, response=f"Failed to delete config: {str(e)}")
155 -
156 - return {"ok": True}
31 + return self._delete_config(input)
32
33 if action == "delete_plugin":
159 - plugin_name = input.get("plugin_name", "")
160 - if not plugin_name:
161 - return Response(status=400, response="Missing plugin_name")
162 - try:
163 - plugins.delete_plugin(plugin_name)
164 - except FileNotFoundError as e:
165 - return Response(status=404, response=str(e))
166 - except ValueError as e:
167 - return Response(status=400, response=str(e))
168 - except Exception as e:
169 - return Response(status=500, response=f"Failed to delete plugin: {str(e)}")
170 - return {"ok": True}
34 + return self._delete_plugin(input)
35
36 if action == "get_default_config":
173 - plugin_name = input.get("plugin_name", "")
174 - if not plugin_name:
175 - return Response(status=400, response="Missing plugin_name")
176 - settings = plugins.get_default_plugin_config(plugin_name)
177 - return {"ok": True, "data": settings or {}}
37 + return self._get_default_config(input)
38
39 if action == "save_config":
180 - plugin_name = input.get("plugin_name", "")
181 - project_name = input.get("project_name", "")
182 - agent_profile = input.get("agent_profile", "")
183 - settings = input.get("settings", {})
184 - if not plugin_name:
185 - return Response(status=400, response="Missing plugin_name")
186 - if not isinstance(settings, dict):
187 - return Response(status=400, response="settings must be an object")
188 - plugins.save_plugin_config(plugin_name, project_name, agent_profile, settings)
189 - return {"ok": True}
40 + return self._save_config(input)
41
42 if action == "toggle_plugin":
192 - plugin_name = input.get("plugin_name", "")
193 - enabled = input.get("enabled")
194 - project_name = input.get("project_name", "")
195 - agent_profile = input.get("agent_profile", "")
196 - clear_overrides = bool(input.get("clear_overrides", False))
197 -
198 - if not plugin_name:
199 - return Response(status=400, response="Missing plugin_name")
200 - if enabled is None:
201 - return Response(status=400, response="Missing enabled state")
202 -
203 - plugins.toggle_plugin(
204 - plugin_name, bool(enabled), project_name, agent_profile, clear_overrides
205 - )
206 - return {"ok": True}
43 + return self._toggle_plugin(input)
44
45 if action == "get_doc":
209 - plugin_name = input.get("plugin_name", "")
210 - doc = input.get("doc", "") # "readme" or "license"
211 - if not plugin_name:
212 - return Response(status=400, response="Missing plugin_name")
213 - if doc not in ("readme", "license"):
214 - return Response(status=400, response="doc must be 'readme' or 'license'")
215 -
216 - plugin_dir = plugins.find_plugin_dir(plugin_name)
217 - if not plugin_dir:
218 - return Response(status=404, response="Plugin not found")
219 -
220 - filename = "README.md" if doc == "readme" else "LICENSE"
221 - file_path = files.get_abs_path(plugin_dir, filename)
222 - if not files.exists(file_path):
223 - return Response(status=404, response=f"{filename} not found")
224 -
225 - return {"ok": True, "content": files.read_file(file_path), "filename": filename}
46 + return self._get_doc(input)
47
48 if action == "run_init_script":
228 - plugin_name = input.get("plugin_name", "")
229 - if not plugin_name:
230 - return Response(status=400, response="Missing plugin_name")
49 + return self._run_init_script(input)
50
232 - plugin_dir = plugins.find_plugin_dir(plugin_name)
233 - if not plugin_dir:
234 - return Response(status=404, response="Plugin not found")
235 -
236 - init_script = files.get_abs_path(plugin_dir, "initialize.py")
237 - if not files.exists(init_script):
238 - return Response(status=404, response="initialize.py not found")
239 -
240 - executed_at = datetime.now(timezone.utc).isoformat()
241 - try:
242 - result = subprocess.run(
243 - [sys.executable, init_script],
244 - stdout=subprocess.PIPE,
245 - stderr=subprocess.STDOUT,
246 - text=True,
247 - cwd=plugin_dir,
248 - timeout=120,
249 - )
250 - exit_code = result.returncode
251 - output = result.stdout or ""
252 - except subprocess.TimeoutExpired:
253 - exit_code = -1
254 - output = "Error: script timed out after 120 seconds"
255 - except Exception as e:
256 - exit_code = -1
257 - output = f"Error: {str(e)}"
258 -
259 - exec_record = {"executed_at": executed_at, "exit_code": exit_code}
260 - exec_path = plugins.determine_plugin_asset_path(plugin_name, "", "", "init_exec.json")
261 - if exec_path:
262 - files.write_file(exec_path, json.dumps(exec_record))
51 + if action == "get_init_exec":
52 + return self._get_init_exec(input)
53 +
54 + return Response(status=400, response=f"Unknown action: {action}")
55 +
56 + def _get_config(self, input: dict) -> dict | Response:
57 + plugin_name = input.get("plugin_name", "")
58 + project_name = input.get("project_name", "")
59 + agent_profile = input.get("agent_profile", "")
60 + if not plugin_name:
61 + return Response(status=400, response="Missing plugin_name")
62 +
63 + result = plugins.find_plugin_assets(
64 + plugins.CONFIG_FILE_NAME,
65 + plugin_name=plugin_name,
66 + project_name=project_name,
67 + agent_profile=agent_profile,
68 + only_first=True,
69 + )
70 + if result:
71 + entry = result[0]
72 + path = entry.get("path", "")
73 + settings = files.read_file_json(path) if path else {}
74 + loaded_project_name = entry.get("project_name", "")
75 + loaded_agent_profile = entry.get("agent_profile", "")
76 + else:
77 + settings = plugins.get_plugin_config(plugin_name, agent=None) or {}
78 + default_path = files.get_abs_path(
79 + plugins.find_plugin_dir(plugin_name), plugins.CONFIG_DEFAULT_FILE_NAME
80 + )
81 + path = default_path if files.exists(default_path) else ""
82 + loaded_project_name = ""
83 + loaded_agent_profile = ""
84 +
85 + return {
86 + "ok": True,
87 + "loaded_path": path,
88 + "loaded_project_name": loaded_project_name,
89 + "loaded_agent_profile": loaded_agent_profile,
90 + "data": settings,
91 + }
92 +
93 + def _get_toggle_status(self, input: dict) -> dict | Response:
94 + plugin_name = input.get("plugin_name", "")
95 + project_name = input.get("project_name", "")
96 + agent_profile = input.get("agent_profile", "")
97 + if not plugin_name:
98 + return Response(status=400, response="Missing plugin_name")
99 +
100 + meta = plugins.get_plugin_meta(plugin_name)
101 + if not meta:
102 + return Response(status=404, response="Plugin not found")
103 +
104 + if meta.always_enabled:
105 + return {
106 + "ok": True,
107 + "status": "enabled",
108 + "loaded_project_name": project_name,
109 + "loaded_agent_profile": agent_profile,
110 + "loaded_path": "",
111 + }
112
113 + result = plugins.find_plugin_assets(
114 + plugins.TOGGLE_FILE_PATTERN,
115 + plugin_name=plugin_name,
116 + project_name=project_name,
117 + agent_profile=agent_profile,
118 + only_first=True,
119 + )
120 +
121 + if result:
122 + entry = result[0]
123 + path = entry.get("path", "")
124 + status = (
125 + "enabled" if path.endswith(plugins.ENABLED_FILE_NAME) else "disabled"
126 + )
127 return {
265 - "ok": exit_code == 0,
266 - "output": output,
267 - "exit_code": exit_code,
268 - "executed_at": executed_at,
128 + "ok": True,
129 + "status": status,
130 + "loaded_project_name": entry.get("project_name", ""),
131 + "loaded_agent_profile": entry.get("agent_profile", ""),
132 + "loaded_path": path,
133 }
134
271 - if action == "get_init_exec":
272 - plugin_name = input.get("plugin_name", "")
273 - if not plugin_name:
274 - return Response(status=400, response="Missing plugin_name")
275 -
276 - exec_path = plugins.determine_plugin_asset_path(plugin_name, "", "", "init_exec.json")
277 - if exec_path and files.exists(exec_path):
278 - try:
279 - data = json.loads(files.read_file(exec_path))
280 - return {"ok": True, "data": data}
281 - except Exception:
282 - pass
283 - return {"ok": True, "data": None}
284 -
285 - return Response(status=400, response=f"Unknown action: {action}")
\ No newline at end of file
135 + return {
136 + "ok": True,
137 + "status": "enabled",
138 + "loaded_project_name": "",
139 + "loaded_agent_profile": "",
140 + "loaded_path": "",
141 + }
142 +
143 + def _list_configs(self, input: dict) -> dict | Response:
144 + plugin_name = input.get("plugin_name", "")
145 + asset_type = input.get("asset_type", "config")
146 + if not plugin_name:
147 + return Response(status=400, response="Missing plugin_name")
148 +
149 + configs = plugins.find_plugin_assets(
150 + (
151 + plugins.CONFIG_FILE_NAME
152 + if asset_type == "config"
153 + else plugins.TOGGLE_FILE_PATTERN
154 + ),
155 + plugin_name=plugin_name,
156 + project_name="*",
157 + agent_profile="*",
158 + only_first=False,
159 + )
160 +
161 + return {"ok": True, "data": configs}
162 +
163 + def _delete_config(self, input: dict) -> dict | Response:
164 + plugin_name = input.get("plugin_name", "")
165 + path = input.get("path", "")
166 + if not plugin_name:
167 + return Response(status=400, response="Missing plugin_name")
168 + if not path:
169 + return Response(status=400, response="Missing path")
170 +
171 + configs = plugins.find_plugin_assets(
172 + plugins.CONFIG_FILE_NAME,
173 + plugin_name=plugin_name,
174 + project_name="*",
175 + agent_profile="*",
176 + only_first=False,
177 + )
178 + toggles = plugins.find_plugin_assets(
179 + plugins.TOGGLE_FILE_PATTERN,
180 + plugin_name=plugin_name,
181 + project_name="*",
182 + agent_profile="*",
183 + only_first=False,
184 + )
185 + allowed_paths = {c.get("path", "") for c in configs + toggles}
186 + if path not in allowed_paths:
187 + return Response(status=400, response="Invalid path")
188 +
189 + if not files.exists(path):
190 + return {"ok": True}
191 +
192 + try:
193 + os.remove(path)
194 + except Exception as e:
195 + return Response(status=500, response=f"Failed to delete config: {str(e)}")
196 +
197 + return {"ok": True}
198 +
199 + def _delete_plugin(self, input: dict) -> dict | Response:
200 + plugin_name = input.get("plugin_name", "")
201 + if not plugin_name:
202 + return Response(status=400, response="Missing plugin_name")
203 + try:
204 + plugins.delete_plugin(plugin_name)
205 + except FileNotFoundError as e:
206 + return Response(status=404, response=str(e))
207 + except ValueError as e:
208 + return Response(status=400, response=str(e))
209 + except Exception as e:
210 + return Response(status=500, response=f"Failed to delete plugin: {str(e)}")
211 + return {"ok": True}
212 +
213 + def _get_default_config(self, input: dict) -> dict | Response:
214 + plugin_name = input.get("plugin_name", "")
215 + if not plugin_name:
216 + return Response(status=400, response="Missing plugin_name")
217 + settings = plugins.get_default_plugin_config(plugin_name)
218 + return {"ok": True, "data": settings or {}}
219 +
220 + def _save_config(self, input: dict) -> dict | Response:
221 + plugin_name = input.get("plugin_name", "")
222 + project_name = input.get("project_name", "")
223 + agent_profile = input.get("agent_profile", "")
224 + settings = input.get("settings", {})
225 + if not plugin_name:
226 + return Response(status=400, response="Missing plugin_name")
227 + if not isinstance(settings, dict):
228 + return Response(status=400, response="settings must be an object")
229 + plugins.save_plugin_config(plugin_name, project_name, agent_profile, settings)
230 + return {"ok": True}
231 +
232 + def _toggle_plugin(self, input: dict) -> dict | Response:
233 + plugin_name = input.get("plugin_name", "")
234 + enabled = input.get("enabled")
235 + project_name = input.get("project_name", "")
236 + agent_profile = input.get("agent_profile", "")
237 + clear_overrides = bool(input.get("clear_overrides", False))
238 +
239 + if not plugin_name:
240 + return Response(status=400, response="Missing plugin_name")
241 + if enabled is None:
242 + return Response(status=400, response="Missing enabled state")
243 +
244 + plugins.toggle_plugin(
245 + plugin_name, bool(enabled), project_name, agent_profile, clear_overrides
246 + )
247 + return {"ok": True}
248 +
249 + def _get_doc(self, input: dict) -> dict | Response:
250 + plugin_name = input.get("plugin_name", "")
251 + doc = input.get("doc", "")
252 + if not plugin_name:
253 + return Response(status=400, response="Missing plugin_name")
254 + if doc not in ("readme", "license"):
255 + return Response(status=400, response="doc must be 'readme' or 'license'")
256 +
257 + plugin_dir = plugins.find_plugin_dir(plugin_name)
258 + if not plugin_dir:
259 + return Response(status=404, response="Plugin not found")
260 +
261 + filename = "README.md" if doc == "readme" else "LICENSE"
262 + file_path = files.get_abs_path(plugin_dir, filename)
263 + if not files.exists(file_path):
264 + return Response(status=404, response=f"{filename} not found")
265 +
266 + return {"ok": True, "content": files.read_file(file_path), "filename": filename}
267 +
268 + def _run_init_script(self, input: dict) -> dict | Response:
269 + plugin_name = input.get("plugin_name", "")
270 + if not plugin_name:
271 + return Response(status=400, response="Missing plugin_name")
272 +
273 + plugin_dir = plugins.find_plugin_dir(plugin_name)
274 + if not plugin_dir:
275 + return Response(status=404, response="Plugin not found")
276 +
277 + init_script = files.get_abs_path(plugin_dir, "initialize.py")
278 + if not files.exists(init_script):
279 + return Response(status=404, response="initialize.py not found")
280 +
281 + executed_at = datetime.now(timezone.utc).isoformat()
282 + try:
283 + result = subprocess.run(
284 + [sys.executable, init_script],
285 + stdout=subprocess.PIPE,
286 + stderr=subprocess.STDOUT,
287 + text=True,
288 + cwd=plugin_dir,
289 + timeout=120,
290 + )
291 + exit_code = result.returncode
292 + output = result.stdout or ""
293 + except subprocess.TimeoutExpired:
294 + exit_code = -1
295 + output = "Error: script timed out after 120 seconds"
296 + except Exception as e:
297 + exit_code = -1
298 + output = f"Error: {str(e)}"
299 +
300 + exec_record = {"executed_at": executed_at, "exit_code": exit_code}
301 + exec_path = plugins.determine_plugin_asset_path(
302 + plugin_name, "", "", "init_exec.json"
303 + )
304 + if exec_path:
305 + files.write_file(exec_path, json.dumps(exec_record))
306 +
307 + return {
308 + "ok": exit_code == 0,
309 + "output": output,
310 + "exit_code": exit_code,
311 + "executed_at": executed_at,
312 + }
313 +
314 + def _get_init_exec(self, input: dict) -> dict | Response:
315 + plugin_name = input.get("plugin_name", "")
316 + if not plugin_name:
317 + return Response(status=400, response="Missing plugin_name")
318 +
319 + exec_path = plugins.determine_plugin_asset_path(
320 + plugin_name, "", "", "init_exec.json"
321 + )
322 + if exec_path and files.exists(exec_path):
323 + try:
324 + data = json.loads(files.read_file(exec_path))
325 + return {"ok": True, "data": data}
326 + except Exception:
327 + pass
328 + return {"ok": True, "data": None}