diff --git a/pkgs/clan-app/clan_app/api/api_bridge.py b/pkgs/clan-app/clan_app/api/api_bridge.py index 4cba4eb20..1df429d71 100644 --- a/pkgs/clan-app/clan_app/api/api_bridge.py +++ b/pkgs/clan-app/clan_app/api/api_bridge.py @@ -5,7 +5,7 @@ from contextlib import ExitStack from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any -from clan_lib.api import ApiResponse +from clan_lib.api import ApiError, ApiResponse, ErrorDataClass from clan_lib.api.tasks import WebThread from clan_lib.async_run import set_current_thread_opkey, set_should_cancel @@ -43,8 +43,6 @@ class ApiBridge(ABC): def process_request(self, request: BackendRequest) -> None: """Process an API request through the middleware chain.""" - from .middleware import MiddlewareContext - with ExitStack() as stack: context = MiddlewareContext( request=request, @@ -75,8 +73,6 @@ class ApiBridge(ABC): location: list[str], ) -> None: """Send an error response.""" - from clan_lib.api import ApiError, ErrorDataClass - error_data = ErrorDataClass( op_key=op_key, status="error", diff --git a/pkgs/clan-app/clan_app/app.py b/pkgs/clan-app/clan_app/app.py index 1a7151167..1275a6684 100644 --- a/pkgs/clan-app/clan_app/app.py +++ b/pkgs/clan-app/clan_app/app.py @@ -1,5 +1,6 @@ import logging import os +import time from dataclasses import dataclass from pathlib import Path @@ -95,8 +96,6 @@ def app_run(app_opts: ClanAppOptions) -> int: log.info("Press Ctrl+C to stop the server") try: # Keep the main thread alive - import time - while True: time.sleep(1) except KeyboardInterrupt: diff --git a/pkgs/clan-cli/api.py b/pkgs/clan-cli/api.py index 226894fbb..44525038a 100755 --- a/pkgs/clan-cli/api.py +++ b/pkgs/clan-cli/api.py @@ -2,14 +2,12 @@ import json -from clan_lib.api import load_in_all_api_functions +from clan_lib.api import API, load_in_all_api_functions def main() -> None: load_in_all_api_functions() - from clan_lib.api import API - schema = API.to_json_schema() print(f"""{json.dumps(schema, indent=2)}""") diff --git a/pkgs/clan-cli/clan_cli/completions.py b/pkgs/clan-cli/clan_cli/completions.py index f14ff7fea..941a027c4 100644 --- a/pkgs/clan-cli/clan_cli/completions.py +++ b/pkgs/clan-cli/clan_cli/completions.py @@ -4,13 +4,16 @@ import json import subprocess import threading from collections.abc import Callable, Iterable +from pathlib import Path from types import ModuleType from typing import Any from clan_lib.cmd import run +from clan_lib.dirs import get_clan_flake_toplevel_or_env from clan_lib.flake.flake import Flake from clan_lib.nix import nix_eval from clan_lib.persist.inventory_store import InventoryStore +from clan_lib.templates import list_templates """ This module provides dynamic completions. @@ -31,7 +34,6 @@ COMPLETION_TIMEOUT: int = 3 def clan_dir(flake: str | None) -> str | None: if flake is not None: return flake - from clan_lib.dirs import get_clan_flake_toplevel_or_env path_result = get_clan_flake_toplevel_or_env() return str(path_result) if path_result is not None else None @@ -210,10 +212,6 @@ def complete_secrets( **_kwargs: Any, ) -> Iterable[str]: """Provides completion functionality for clan secrets""" - from clan_lib.flake.flake import Flake - - from .secrets.secrets import list_secrets - flake = ( clan_dir_result if (clan_dir_result := clan_dir(getattr(parsed_args, "flake", None))) @@ -233,10 +231,6 @@ def complete_users( **_kwargs: Any, ) -> Iterable[str]: """Provides completion functionality for clan users""" - from pathlib import Path - - from .secrets.users import list_users - flake = ( clan_dir_result if (clan_dir_result := clan_dir(getattr(parsed_args, "flake", None))) @@ -256,10 +250,6 @@ def complete_groups( **_kwargs: Any, ) -> Iterable[str]: """Provides completion functionality for clan groups""" - from pathlib import Path - - from .secrets.groups import list_groups - flake = ( clan_dir_result if (clan_dir_result := clan_dir(getattr(parsed_args, "flake", None))) @@ -280,8 +270,6 @@ def complete_templates_disko( **_kwargs: Any, ) -> Iterable[str]: """Provides completion functionality for disko templates""" - from clan_lib.templates import list_templates - flake = ( clan_dir_result if (clan_dir_result := clan_dir(getattr(parsed_args, "flake", None))) @@ -304,8 +292,6 @@ def complete_templates_clan( **_kwargs: Any, ) -> Iterable[str]: """Provides completion functionality for clan templates""" - from clan_lib.templates import list_templates - flake = ( clan_dir_result if (clan_dir_result := clan_dir(getattr(parsed_args, "flake", None))) @@ -331,8 +317,6 @@ def complete_vars_for_machine( Only completes vars that already exist in the vars directory on disk. This is fast as it only scans the filesystem without any evaluation. """ - from pathlib import Path - machine_name = getattr(parsed_args, "machine", None) if not machine_name: return [] diff --git a/pkgs/clan-cli/clan_cli/facts/generate.py b/pkgs/clan-cli/clan_cli/facts/generate.py index 359ec3257..c25297fc3 100644 --- a/pkgs/clan-cli/clan_cli/facts/generate.py +++ b/pkgs/clan-cli/clan_cli/facts/generate.py @@ -7,6 +7,7 @@ from collections.abc import Callable from pathlib import Path from tempfile import TemporaryDirectory +from clan_lib import bwrap from clan_lib.cmd import RunOpts, run from clan_lib.errors import ClanError from clan_lib.flake import require_flake @@ -104,7 +105,6 @@ def generate_service_facts( machine.facts_data[service]["generator"]["prompt"], ) env["prompt_value"] = prompt_value - from clan_lib import bwrap if sys.platform == "linux" and bwrap.bubblewrap_works(): cmd = bubblewrap_cmd(generator, facts_dir, secrets_dir) diff --git a/pkgs/clan-cli/clan_cli/hyperlink.py b/pkgs/clan-cli/clan_cli/hyperlink.py index 7b7e3ed25..272255409 100644 --- a/pkgs/clan-cli/clan_cli/hyperlink.py +++ b/pkgs/clan-cli/clan_cli/hyperlink.py @@ -1,3 +1,6 @@ +import sys + + # Implementation of OSC8 def hyperlink(text: str, url: str) -> str: """Generate OSC8 escape sequence for hyperlinks. @@ -20,10 +23,7 @@ def hyperlink_same_text_and_url(url: str) -> str: def help_hyperlink(description: str, url: str) -> str: - import sys - - """ - Keep the description and the link the same to support legacy terminals. + """Keep the description and the link the same to support legacy terminals. """ if sys.argv[0].__contains__("docs.py"): return docs_hyperlink(description, url) diff --git a/pkgs/clan-cli/clan_cli/tests/sshd.py b/pkgs/clan-cli/clan_cli/tests/sshd.py index 9696a7f30..10f80f8ad 100644 --- a/pkgs/clan-cli/clan_cli/tests/sshd.py +++ b/pkgs/clan-cli/clan_cli/tests/sshd.py @@ -123,8 +123,6 @@ def sshd( unused_tcp_port: "PortFunction", monkeypatch: pytest.MonkeyPatch, ) -> Iterator[Sshd]: - import subprocess - port = unused_tcp_port() sshd = shutil.which("sshd") assert sshd is not None, "no sshd binary found" diff --git a/pkgs/clan-cli/clan_cli/tests/test_machines_cli.py b/pkgs/clan-cli/clan_cli/tests/test_machines_cli.py index 893f146f5..7d476e2bf 100644 --- a/pkgs/clan-cli/clan_cli/tests/test_machines_cli.py +++ b/pkgs/clan-cli/clan_cli/tests/test_machines_cli.py @@ -1,5 +1,8 @@ # ruff: noqa: SLF001 +import argparse + import pytest +from clan_cli.machines.update import register_update_parser from clan_cli.secrets.folders import sops_machines_folder from clan_cli.tests import fixtures_flakes from clan_cli.tests.age_keys import SopsSetup, assert_secrets_file_recipients @@ -69,10 +72,6 @@ def test_machine_subcommands( def test_machines_update_with_tags( test_flake_with_core: fixtures_flakes.FlakeForTest, # noqa: ARG001 ) -> None: - import argparse - - from clan_cli.machines.update import register_update_parser - parser = argparse.ArgumentParser() register_update_parser(parser) diff --git a/pkgs/clan-vm-manager/clan_vm_manager/components/trayicon.py b/pkgs/clan-vm-manager/clan_vm_manager/components/trayicon.py index d08d85841..3674583f5 100644 --- a/pkgs/clan-vm-manager/clan_vm_manager/components/trayicon.py +++ b/pkgs/clan-vm-manager/clan_vm_manager/components/trayicon.py @@ -28,8 +28,11 @@ from typing import Any, ClassVar import gi gi.require_version("Gtk", "4.0") + from gi.repository import GdkPixbuf, Gio, GLib, Gtk +from clan_vm_manager.assets import loc + # DUMMY IMPLEMENTATION ################################################ @@ -596,8 +599,6 @@ class StatusNotifierImplementation(BaseImplementation): ) self.tray_icon.register() - from clan_vm_manager.assets import loc - icon_path = str(loc / "clan_white_notext.png") self.set_icon(icon_path)