treewide: reformat

This commit is contained in:
Michael Hoang
2025-10-13 17:36:50 +02:00
parent 9b71f106f6
commit cbcfcd507d
26 changed files with 66 additions and 68 deletions

View File

@@ -41,7 +41,7 @@ class ApiBridge(Protocol):
def process_request(self, request: BackendRequest) -> None:
"""Process an API request through the middleware chain."""
from clan_app.middleware.base import MiddlewareContext # noqa: PLC0415
from clan_app.middleware.base import MiddlewareContext
with ExitStack() as stack:
# Capture the current call stack up to this point
@@ -62,7 +62,7 @@ class ApiBridge(Protocol):
)
middleware.process(context)
except Exception as e:
from clan_app.middleware.base import ( # noqa: PLC0415
from clan_app.middleware.base import (
MiddlewareError,
)

View File

@@ -191,13 +191,13 @@ class HttpBridge(ApiBridge, BaseHTTPRequestHandler):
return file_data
def do_OPTIONS(self) -> None:
def do_OPTIONS(self) -> None: # noqa: N802
"""Handle CORS preflight requests."""
self.send_response_only(200)
self._send_cors_headers()
self.end_headers()
def do_GET(self) -> None:
def do_GET(self) -> None: # noqa: N802
"""Handle GET requests."""
parsed_url = urlparse(self.path)
path = parsed_url.path
@@ -211,7 +211,7 @@ class HttpBridge(ApiBridge, BaseHTTPRequestHandler):
else:
self.send_api_error_response("info", "Not Found", ["http_bridge", "GET"])
def do_POST(self) -> None:
def do_POST(self) -> None: # noqa: N802
"""Handle POST requests."""
parsed_url = urlparse(self.path)
path = parsed_url.path

View File

@@ -34,7 +34,7 @@ class WebviewBridge(ApiBridge):
log.debug(f"Sending response: {serialized}")
# Import FuncStatus locally to avoid circular import
from .webview import FuncStatus # noqa: PLC0415
from .webview import FuncStatus
self.webview.return_(response._op_key, FuncStatus.SUCCESS, serialized) # noqa: SLF001

View File

@@ -9,7 +9,7 @@ def main() -> None:
load_in_all_api_functions()
# import lazily since we otherwise we do not have all api functions loaded according to Qubasa
from clan_lib.api import API # noqa: PLC0415
from clan_lib.api import API
schema = API.to_json_schema()
print(f"""{json.dumps(schema, indent=2)}""")

View File

@@ -102,7 +102,7 @@ class TestFlake(Flake):
opts: "ListOptions | None" = None, # noqa: ARG002
) -> "dict[str, MachineResponse]":
"""List machines of a clan"""
from clan_lib.machines.actions import ( # noqa: PLC0415
from clan_lib.machines.actions import (
InventoryMachine,
MachineResponse,
)

View File

@@ -231,7 +231,7 @@ def remove_machine_command(args: argparse.Namespace) -> None:
def add_group_argument(parser: argparse.ArgumentParser) -> None:
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_groups,
)
@@ -334,7 +334,7 @@ def register_groups_parser(parser: argparse.ArgumentParser) -> None:
help="the name of the machines to add",
type=machine_name_type,
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_machines,
)
@@ -353,7 +353,7 @@ def register_groups_parser(parser: argparse.ArgumentParser) -> None:
help="the name of the machines to remove",
type=machine_name_type,
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_machines,
)
@@ -369,7 +369,7 @@ def register_groups_parser(parser: argparse.ArgumentParser) -> None:
help="the name of the user to add",
type=user_name_type,
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_users,
)
@@ -388,7 +388,7 @@ def register_groups_parser(parser: argparse.ArgumentParser) -> None:
help="the name of the user to remove",
type=user_name_type,
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_users,
)
@@ -407,7 +407,7 @@ def register_groups_parser(parser: argparse.ArgumentParser) -> None:
help="the name of the secret",
type=secret_name_type,
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_secrets,
)
@@ -426,7 +426,7 @@ def register_groups_parser(parser: argparse.ArgumentParser) -> None:
help="the name of the secret",
type=secret_name_type,
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_secrets,
)

View File

@@ -69,7 +69,7 @@ def register_import_sops_parser(parser: argparse.ArgumentParser) -> None:
default=[],
help="the group to import the secrets to",
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_groups,
)
@@ -82,7 +82,7 @@ def register_import_sops_parser(parser: argparse.ArgumentParser) -> None:
default=[],
help="the machine to import the secrets to",
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_machines,
)
@@ -95,7 +95,7 @@ def register_import_sops_parser(parser: argparse.ArgumentParser) -> None:
default=[],
help="the user to import the secrets to",
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_users,
)

View File

@@ -172,7 +172,7 @@ def register_machines_parser(parser: argparse.ArgumentParser) -> None:
help="the name of the machine",
type=machine_name_type,
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_machines,
)
@@ -192,7 +192,7 @@ def register_machines_parser(parser: argparse.ArgumentParser) -> None:
help="the name of the machine",
type=machine_name_type,
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_machines,
)
@@ -207,7 +207,7 @@ def register_machines_parser(parser: argparse.ArgumentParser) -> None:
help="the name of the machine",
type=machine_name_type,
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_machines,
)
@@ -225,7 +225,7 @@ def register_machines_parser(parser: argparse.ArgumentParser) -> None:
help="the name of the machine",
type=machine_name_type,
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_machines,
complete_secrets,
@@ -250,7 +250,7 @@ def register_machines_parser(parser: argparse.ArgumentParser) -> None:
help="the name of the machine",
type=machine_name_type,
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_machines,
complete_secrets,

View File

@@ -255,7 +255,7 @@ def add_secret_argument(parser: argparse.ArgumentParser, autocomplete: bool) ->
type=secret_name_type,
)
if autocomplete:
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_secrets,
)
@@ -467,7 +467,7 @@ def register_secrets_parser(subparser: argparse._SubParsersAction) -> None:
default=[],
help="the group to import the secrets to (can be repeated)",
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_groups,
)
@@ -480,7 +480,7 @@ def register_secrets_parser(subparser: argparse._SubParsersAction) -> None:
default=[],
help="the machine to import the secrets to (can be repeated)",
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_machines,
)
@@ -493,7 +493,7 @@ def register_secrets_parser(subparser: argparse._SubParsersAction) -> None:
default=[],
help="the user to import the secrets to (can be repeated)",
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_users,
)

View File

@@ -281,7 +281,7 @@ def register_users_parser(parser: argparse.ArgumentParser) -> None:
help="the name of the user",
type=user_name_type,
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_users,
)
@@ -295,7 +295,7 @@ def register_users_parser(parser: argparse.ArgumentParser) -> None:
help="the name of the user",
type=user_name_type,
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_users,
)
@@ -312,7 +312,7 @@ def register_users_parser(parser: argparse.ArgumentParser) -> None:
help="the name of the user",
type=user_name_type,
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_secrets,
complete_users,
@@ -336,7 +336,7 @@ def register_users_parser(parser: argparse.ArgumentParser) -> None:
help="the name of the group",
type=user_name_type,
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_secrets,
complete_users,
@@ -360,7 +360,7 @@ def register_users_parser(parser: argparse.ArgumentParser) -> None:
help="the name of the user",
type=user_name_type,
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_users,
)
@@ -378,7 +378,7 @@ def register_users_parser(parser: argparse.ArgumentParser) -> None:
help="the name of the user",
type=user_name_type,
)
from clan_cli.completions import ( # noqa: PLC0415
from clan_cli.completions import (
add_dynamic_completer,
complete_users,
)

View File

@@ -171,7 +171,7 @@ class StoreBase(ABC):
if generator.share:
log_info = log.info
else:
from clan_lib.machines.machines import Machine # noqa: PLC0415
from clan_lib.machines.machines import Machine
machine_obj = Machine(name=generator.machines[0], flake=self.flake)
log_info = machine_obj.info

View File

@@ -53,7 +53,7 @@ def vars_status(
flake: Flake,
generator_name: None | str = None,
) -> VarStatus:
from clan_cli.vars.generator import Generator # noqa: PLC0415
from clan_cli.vars.generator import Generator
machine = Machine(name=machine_name, flake=flake)
missing_secret_vars = []

View File

@@ -482,7 +482,7 @@ class Generator:
if sys.platform == "linux" and bwrap.bubblewrap_works():
cmd = bubblewrap_cmd(str(final_script), tmpdir)
elif sys.platform == "darwin":
from clan_lib.sandbox_exec import sandbox_exec_cmd # noqa: PLC0415
from clan_lib.sandbox_exec import sandbox_exec_cmd
cmd = stack.enter_context(sandbox_exec_cmd(str(final_script), tmpdir))
else:

View File

@@ -54,7 +54,7 @@ class SecretStore(StoreBase):
def ensure_machine_key(self, machine: str) -> None:
"""Ensure machine has sops keys initialized."""
# no need to generate keys if we don't manage secrets
from clan_cli.vars.generator import Generator # noqa: PLC0415
from clan_cli.vars.generator import Generator
vars_generators = Generator.get_machine_generators([machine], self.flake)
if not vars_generators:
@@ -143,7 +143,7 @@ class SecretStore(StoreBase):
"""
if generators is None:
from clan_cli.vars.generator import Generator # noqa: PLC0415
from clan_cli.vars.generator import Generator
generators = Generator.get_machine_generators([machine], self.flake)
file_found = False
@@ -218,7 +218,7 @@ class SecretStore(StoreBase):
return [store_folder]
def populate_dir(self, machine: str, output_dir: Path, phases: list[str]) -> None:
from clan_cli.vars.generator import Generator # noqa: PLC0415
from clan_cli.vars.generator import Generator
vars_generators = Generator.get_machine_generators([machine], self.flake)
if "users" in phases or "services" in phases:
@@ -292,7 +292,7 @@ class SecretStore(StoreBase):
)
def collect_keys_for_secret(self, machine: str, path: Path) -> set[sops.SopsKey]:
from clan_cli.secrets.secrets import ( # noqa: PLC0415
from clan_cli.secrets.secrets import (
collect_keys_for_path,
collect_keys_for_type,
)
@@ -354,10 +354,10 @@ class SecretStore(StoreBase):
ClanError: If the specified file_name is not found
"""
from clan_cli.secrets.secrets import update_keys # noqa: PLC0415
from clan_cli.secrets.secrets import update_keys
if generators is None:
from clan_cli.vars.generator import Generator # noqa: PLC0415
from clan_cli.vars.generator import Generator
generators = Generator.get_machine_generators([machine], self.flake)
file_found = False

View File

@@ -319,9 +319,9 @@ def load_in_all_api_functions() -> None:
We have to make sure python loads every wrapped function at least once.
This is done by importing all modules from the clan_lib and clan_cli packages.
"""
import clan_cli # noqa: PLC0415 # Avoid circular imports - many modules import from clan_lib.api
import clan_cli # Avoid circular imports - many modules import from clan_lib.api
import clan_lib # noqa: PLC0415 # Avoid circular imports - many modules import from clan_lib.api
import clan_lib # Avoid circular imports - many modules import from clan_lib.api
import_all_modules_from_package(clan_lib)
import_all_modules_from_package(clan_cli)

View File

@@ -88,7 +88,7 @@ def list_system_storage_devices() -> Blockdevices:
A list of detected block devices with metadata like size, path, type, etc.
"""
from clan_lib.nix import nix_shell # noqa: PLC0415
from clan_lib.nix import nix_shell
cmd = nix_shell(
["util-linux"],
@@ -124,7 +124,7 @@ def get_clan_directory_relative(flake: Flake) -> str:
ClanError: If the flake evaluation fails or directories cannot be found
"""
from clan_lib.dirs import get_clan_directories # noqa: PLC0415
from clan_lib.dirs import get_clan_directories
_, relative_dir = get_clan_directories(flake)
return relative_dir

View File

@@ -1162,7 +1162,7 @@ class Flake:
opts: "ListOptions | None" = None,
) -> "dict[str, MachineResponse]":
"""List machines of a clan"""
from clan_lib.machines.actions import list_machines # noqa: PLC0415
from clan_lib.machines.actions import list_machines
return list_machines(self, opts)

View File

@@ -18,14 +18,14 @@ def locked_open(filename: Path, mode: str = "r") -> Generator:
def write_history_file(data: Any) -> None:
from clan_lib.dirs import user_history_file # noqa: PLC0415
from clan_lib.dirs import user_history_file
with locked_open(user_history_file(), "w+") as f:
f.write(json.dumps(data, cls=ClanJSONEncoder, indent=4))
def read_history_file() -> list[dict]:
from clan_lib.dirs import user_history_file # noqa: PLC0415
from clan_lib.dirs import user_history_file
with locked_open(user_history_file(), "r") as f:
content: str = f.read()

View File

@@ -33,7 +33,7 @@ class Machine:
def get_inv_machine(self) -> "InventoryMachine":
# Import on demand to avoid circular imports
from clan_lib.machines.actions import get_machine # noqa: PLC0415
from clan_lib.machines.actions import get_machine
return get_machine(self.flake, self.name)
@@ -95,7 +95,7 @@ class Machine:
@cached_property
def secret_vars_store(self) -> StoreBase:
from clan_cli.vars.secret_modules import password_store # noqa: PLC0415
from clan_cli.vars.secret_modules import password_store
secret_module = self.select("config.clan.core.vars.settings.secretModule")
module = importlib.import_module(secret_module)
@@ -126,7 +126,7 @@ class Machine:
return self.flake.path
def target_host(self) -> Remote:
from clan_lib.network.network import get_best_remote # noqa: PLC0415
from clan_lib.network.network import get_best_remote
with get_best_remote(self) as remote:
return remote

View File

@@ -42,7 +42,7 @@ def _suggest_similar_names(
def get_available_machines(flake: Flake) -> list[str]:
from clan_lib.machines.list import list_machines # noqa: PLC0415
from clan_lib.machines.list import list_machines
machines = list_machines(flake)
return list(machines.keys())

View File

@@ -34,7 +34,7 @@ class Peer:
_var: dict[str, str] = self._host["var"]
machine_name = _var["machine"]
generator = _var["generator"]
from clan_lib.machines.machines import Machine # noqa: PLC0415
from clan_lib.machines.machines import Machine
machine = Machine(name=machine_name, flake=self.flake)
var = get_machine_var(

View File

@@ -88,7 +88,7 @@ def nix_eval(flags: list[str]) -> list[str]:
],
)
if os.environ.get("IN_NIX_SANDBOX"):
from clan_lib.dirs import nixpkgs_source # noqa: PLC0415
from clan_lib.dirs import nixpkgs_source
return [
*default_flags,
@@ -169,7 +169,7 @@ def nix_shell(packages: list[str], cmd: list[str]) -> list[str]:
if not missing_packages:
return cmd
from clan_lib.dirs import nixpkgs_flake # noqa: PLC0415
from clan_lib.dirs import nixpkgs_flake
return [
*nix_command(["shell", "--inputs-from", f"{nixpkgs_flake()!s}"]),

View File

@@ -464,12 +464,12 @@ class Remote:
self,
opts: "ConnectionOptions | None" = None,
) -> None:
from clan_lib.network.check import check_machine_ssh_reachable # noqa: PLC0415
from clan_lib.network.check import check_machine_ssh_reachable
return check_machine_ssh_reachable(self, opts)
def check_machine_ssh_login(self) -> None:
from clan_lib.network.check import check_machine_ssh_login # noqa: PLC0415
from clan_lib.network.check import check_machine_ssh_login
return check_machine_ssh_login(self)

View File

@@ -290,9 +290,7 @@ def collect_commands() -> list[Category]:
# 3. sort by title alphabetically
return (c.title.split(" ")[0], c.title, weight)
result = sorted(result, key=weight_cmd_groups)
return result
return sorted(result, key=weight_cmd_groups)
def build_command_reference() -> None:

View File

@@ -36,7 +36,7 @@ class MPProcess:
def _set_proc_name(name: str) -> None:
if sys.platform != "linux":
return
import ctypes # noqa: PLC0415
import ctypes
# Define the prctl function with the appropriate arguments and return type
libc = ctypes.CDLL("libc.so.6")

View File

@@ -759,12 +759,12 @@ class Win32Implementation(BaseImplementation):
SM_CXSMICON = 49
if sys.platform == "win32":
from ctypes import Structure # noqa: PLC0415
from ctypes import Structure
class WNDCLASSW(Structure):
"""Windows class structure for window registration."""
from ctypes import CFUNCTYPE, wintypes # noqa: PLC0415
from ctypes import CFUNCTYPE, wintypes
LPFN_WND_PROC = CFUNCTYPE(
wintypes.INT,
@@ -789,7 +789,7 @@ class Win32Implementation(BaseImplementation):
class MENUITEMINFOW(Structure):
"""Windows menu item information structure."""
from ctypes import wintypes # noqa: PLC0415
from ctypes import wintypes
_fields_: ClassVar = [
("cb_size", wintypes.UINT),
@@ -809,7 +809,7 @@ class Win32Implementation(BaseImplementation):
class NOTIFYICONDATAW(Structure):
"""Windows notification icon data structure."""
from ctypes import wintypes # noqa: PLC0415
from ctypes import wintypes
_fields_: ClassVar = [
("cb_size", wintypes.DWORD),
@@ -1061,7 +1061,7 @@ class Win32Implementation(BaseImplementation):
if sys.platform != "win32":
return
from ctypes import wintypes # noqa: PLC0415
from ctypes import wintypes
if self._menu is None:
self.update_menu()
@@ -1110,7 +1110,7 @@ class Win32Implementation(BaseImplementation):
if sys.platform != "win32":
return 0
from ctypes import wintypes # noqa: PLC0415
from ctypes import wintypes
if msg == self.WM_TRAYICON:
if l_param == self.WM_RBUTTONUP: