BLE001: fix

pkgs/clan-app/clan_app/api/api_bridge.py

@@ -59,7 +59,7 @@ class ApiBridge(ABC):
                         f"{middleware.__class__.__name__} => {request.method_name}",
                     )
                     middleware.process(context)
-                except Exception as e:
+                except Exception as e:  # noqa: BLE001
                     # If middleware fails, handle error
                     self.send_api_error_response(
                         request.op_key or "unknown",

pkgs/clan-app/clan_app/deps/http/http_bridge.py

@@ -148,7 +148,7 @@ class HttpBridge(ApiBridge, BaseHTTPRequestHandler):
             self.send_header("Content-Type", content_type)
             self.end_headers()
             self.wfile.write(file_data)
-        except Exception as e:
+        except (OSError, json.JSONDecodeError, UnicodeDecodeError) as e:
             log.error(f"Error reading Swagger file: {e!s}")
             self.send_error(500, "Internal Server Error")
 
@@ -252,7 +252,7 @@ class HttpBridge(ApiBridge, BaseHTTPRequestHandler):
         gen_op_key = str(uuid.uuid4())
         try:
             self._handle_api_request(method_name, request_data, gen_op_key)
-        except Exception as e:
+        except RuntimeError as e:
             log.exception(f"Error processing API request {method_name}")
             self.send_api_error_response(
                 gen_op_key,
@@ -275,7 +275,7 @@ class HttpBridge(ApiBridge, BaseHTTPRequestHandler):
                 ["http_bridge", "POST", method_name],
             )
             return None
-        except Exception as e:
+        except (OSError, ValueError, UnicodeDecodeError) as e:
             self.send_api_error_response(
                 "post",
                 f"Error reading request: {e!s}",
@@ -305,7 +305,7 @@ class HttpBridge(ApiBridge, BaseHTTPRequestHandler):
                 op_key=op_key,
             )
 
-        except Exception as e:
+        except (KeyError, TypeError, ValueError) as e:
             self.send_api_error_response(
                 gen_op_key,
                 str(e),

pkgs/clan-app/clan_app/deps/webview/webview.py

@@ -81,7 +81,7 @@ class Webview:
                     msg = message_queue.get()  # Blocks until available
                     js_code = f"window.notifyBus({json.dumps(msg)});"
                     self.eval(js_code)
-                except Exception as e:
+                except (json.JSONDecodeError, RuntimeError, AttributeError) as e:
                     print("Bridge notify error:", e)
                 sleep(0.01)  # avoid busy loop
 
@@ -211,7 +211,7 @@ class Webview:
             try:
                 result = callback(*args)
                 success = True
-            except Exception as e:
+            except Exception as e:  # noqa: BLE001
                 result = str(e)
                 success = False
             self.return_(seq.decode(), 0 if success else 1, json.dumps(result))

pkgs/clan-cli/clan_cli/network/list.py

@@ -1,6 +1,7 @@
 import argparse
 import logging
 
+from clan_lib.errors import ClanError
 from clan_lib.flake import require_flake
 from clan_lib.network.network import networks_from_flake
 
@@ -54,7 +55,7 @@ def list_command(args: argparse.Namespace) -> None:
         try:
             is_running = network.is_running()
             running_status = "Yes" if is_running else "No"
-        except Exception:
+        except ClanError:
             running_status = "Error"
 
         print(

pkgs/clan-cli/clan_cli/secrets/sops.py

@@ -83,7 +83,7 @@ class KeyType(enum.Enum):
 
                 except FileNotFoundError:
                     return
-                except Exception as ex:
+                except OSError as ex:
                     log.warning(f"Could not read age keys from {key_path}", exc_info=ex)
 
             if keys := os.environ.get("SOPS_AGE_KEY"):

pkgs/clan-cli/clan_lib/async_run/__init__.py

@@ -155,7 +155,7 @@ class AsyncThread[**P, R](threading.Thread):
             set_should_cancel(lambda: self.stop_event.is_set())
             # Arguments for ParamSpec "P@AsyncThread" are missing
             self.result = AsyncResult(_result=self.function(*self.args, **self.kwargs))
-        except Exception as ex:
+        except Exception as ex:  # noqa: BLE001
             self.result = AsyncResult(_result=ex)
         finally:
             self.finished = True

pkgs/clan-cli/clan_lib/flake/flake.py

@@ -846,7 +846,7 @@ class Flake:
             return
         try:
             self._cache.load_from_file(path)
-        except Exception as e:
+        except (OSError, json.JSONDecodeError, KeyError, ValueError) as e:
             log.warning(f"Failed load eval cache: {e}. Continue without cache")
 
     def prefetch(self) -> None:

pkgs/clan-cli/clan_lib/machines/update.py

@@ -234,7 +234,7 @@ def run_machine_update(
                 is_mobile = machine.select(
                     "config.system.clan.deployment.nixosMobileWorkaround",
                 )
-            except Exception:
+            except ClanError:
                 is_mobile = False
             # if the machine is mobile, we retry to deploy with the mobile workaround method
             if is_mobile:

pkgs/clan-cli/clan_lib/network/network.py

@@ -161,12 +161,9 @@ def get_best_remote(machine: "Machine") -> Iterator["Remote"]:
     if target_host:
         log.debug(f"Using targetHost from inventory for {machine.name}: {target_host}")
         # Create a direct network with just this machine
-        try:
+        remote = Remote.from_ssh_uri(machine_name=machine.name, address=target_host)
-            remote = Remote.from_ssh_uri(machine_name=machine.name, address=target_host)
+        yield remote
-            yield remote
+        return
-            return
-        except Exception as e:
-            log.debug(f"Inventory targetHost not reachable for {machine.name}: {e}")
 
     # Step 2: Try existing networks by priority
     try:
@@ -189,7 +186,7 @@ def get_best_remote(machine: "Machine") -> Iterator["Remote"]:
                         )
                         yield network.remote(machine.name)
                         return
-                except Exception as e:
+                except ClanError as e:
                     log.debug(f"Failed to reach {machine.name} via {network_name}: {e}")
             else:
                 try:
@@ -202,34 +199,26 @@ def get_best_remote(machine: "Machine") -> Iterator["Remote"]:
                             )
                             yield connected_network.remote(machine.name)
                             return
-                except Exception as e:
+                except ClanError as e:
                     log.debug(
                         f"Failed to establish connection to {machine.name} via {network_name}: {e}",
                     )
-    except Exception as e:
+    except (ImportError, AttributeError, KeyError) as e:
         log.debug(f"Failed to use networking modules to determine machines remote: {e}")
 
     # Step 3: Try targetHost from machine nixos config
-    try:
+    target_host = machine.select('config.clan.core.networking."targetHost"')
-        target_host = machine.select('config.clan.core.networking."targetHost"')
+    if target_host:
-        if target_host:
+        log.debug(
-            log.debug(
+            f"Using targetHost from machine config for {machine.name}: {target_host}",
-                f"Using targetHost from machine config for {machine.name}: {target_host}",
+        )
-            )
+        # Check if reachable
-            # Check if reachable
+        remote = Remote.from_ssh_uri(
-            try:
+            machine_name=machine.name,
-                remote = Remote.from_ssh_uri(
+            address=target_host,
-                    machine_name=machine.name,
+        )
-                    address=target_host,
+        yield remote
-                )
+        return
-                yield remote
-                return
-            except Exception as e:
-                log.debug(
-                    f"Machine config targetHost not reachable for {machine.name}: {e}",
-                )
-    except Exception as e:
-        log.debug(f"Could not get targetHost from machine config: {e}")
 
     # No connection method found
     msg = f"Could not find any way to connect to machine '{machine.name}'. No targetHost configured and machine not reachable via any network."
@@ -249,12 +238,7 @@ def get_network_overview(networks: dict[str, Network]) -> dict:
         else:
             with module.connection(network) as conn:
                 for peer_name in conn.peers:
-                    try:
+                    result[network_name]["peers"][peer_name] = conn.ping(
-                        result[network_name]["peers"][peer_name] = conn.ping(
+                        peer_name,
-                            peer_name,
+                    )
-                        )
-                    except ClanError:
-                        log.warning(
-                            f"getting host for machine: {peer_name} in network: {network_name} failed",
-                        )
     return result

pkgs/clan-cli/clan_lib/network/qr_code.py

@@ -29,6 +29,7 @@ class QRCodeData:
 
     @contextmanager
     def get_best_remote(self) -> Iterator[Remote]:
+        errors = []
         for address in self.addresses:
             try:
                 log.debug(f"Establishing connection via {address}")
@@ -39,8 +40,13 @@ class QRCodeData:
                     if ping_time is not None:
                         log.info(f"reachable via {address} after connection")
                         yield address.remote
-            except Exception as e:
+            except ClanError as e:
-                log.debug(f"Failed to establish connection via {address}: {e}")
+                errors.append((address, e))
+        if not errors:
+            msg = "No reachable remote found in QR code data: " + ", ".join(
+                f"{addr.remote} ({err})" for addr, err in errors
+            )
+            raise ClanError(msg)
 
 
 def read_qr_json(qr_data: dict[str, Any], flake: Flake) -> QRCodeData:

pkgs/clan-cli/clan_lib/ssh/remote_test.py

@@ -239,7 +239,7 @@ def test_run_exception(hosts: list[Remote], runtime: AsyncRuntime) -> None:
             runtime.async_run(None, host.run_local, ["exit 1"], RunOpts(shell=True))  # noqa: S604
         runtime.join_all()
         runtime.check_all()
-    except Exception:  # noqa: S110
+    except ClanError:
         pass
     else:
         msg = "should have raised Exception"
@@ -255,7 +255,7 @@ def test_run_function_exception(hosts: list[Remote], runtime: AsyncRuntime) -> N
             runtime.async_run(None, some_func, host)
         runtime.join_all()
         runtime.check_all()
-    except Exception:  # noqa: S110
+    except ClanError:
         pass
     else:
         msg = "should have raised Exception"

pkgs/clan-cli/clan_lib/ssh/sudo_askpass_proxy.py

@@ -90,7 +90,7 @@ class SudoAskpassProxy:
                     ssh_process.stdin.flush()
                 else:
                     print(stripped_line)
-        except Exception as e:
+        except (OSError, ClanError) as e:
             logger.error(f"Error processing passwords requests output: {e}")
 
     def run(self) -> str:

pkgs/clan-vm-manager/clan_vm_manager/components/executor.py

@@ -84,7 +84,7 @@ def _init_proc(
     print(linebreak + f" {func.__name__}:{pid} " + linebreak, file=sys.stderr)
     try:
         func(**kwargs)
-    except Exception as ex:
+    except Exception as ex:  # noqa: BLE001
         traceback.print_exc()
         if on_except is not None:
             on_except(ex, mp.current_process())

pkgs/clan-vm-manager/clan_vm_manager/components/vmobj.py

@@ -345,7 +345,7 @@ class VMObject(GObject.Object):
                     raise ClanError(msg)
                 with self.qmp_wrap.qmp_ctx() as qmp:
                     qmp.command("system_powerdown")
-            except Exception as ex:
+            except (ClanError, OSError, ConnectionError) as ex:
                 log.debug(f"QMP command 'system_powerdown' ignored. Error: {ex}")
 
             # Try 20 times to stop the VM

pkgs/clan-vm-manager/clan_vm_manager/singletons/use_vms.py

@@ -139,7 +139,7 @@ class ClanStore:
 
                 # Convert the byte array to a string and print it
                 logs_view.set_message(contents.decode("utf-8"))
-            except Exception as e:
+            except (GLib.Error, UnicodeDecodeError) as e:
                 print(f"Error reading file: {e}")
 
         # only one vm can output logs at a time

pkgs/classgen/main.py

@@ -1,7 +1,6 @@
 import argparse
 import json
 import logging
-import traceback
 from collections.abc import Callable, Iterable
 from functools import partial
 from pathlib import Path
@@ -454,8 +453,4 @@ def main() -> None:
 
 
 if __name__ == "__main__":
-    try:
+    main()
-        main()
-    except Exception:
-        print("An error occurred:")
-        traceback.print_exc()