PLC0415: fix
This commit is contained in:
@@ -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",
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)}""")
|
||||
|
||||
|
||||
@@ -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 []
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user