Moved history to own subcommand

This commit is contained in:
Qubasa
2023-12-11 19:09:34 +01:00
parent 87a8f5518b
commit 771c0e43ca
10 changed files with 112 additions and 95 deletions

View File

@@ -6,7 +6,7 @@ from pathlib import Path
from types import ModuleType from types import ModuleType
from typing import Any from typing import Any
from . import backups, config, flakes, machines, secrets, vms, webui from . import backups, config, flakes, history, machines, secrets, vms, webui
from .custom_logger import setup_logging from .custom_logger import setup_logging
from .dirs import get_clan_flake_toplevel, is_clan_flake from .dirs import get_clan_flake_toplevel, is_clan_flake
from .ssh import cli as ssh_cli from .ssh import cli as ssh_cli
@@ -111,6 +111,9 @@ def create_parser(prog: str | None = None) -> argparse.ArgumentParser:
parser_vms = subparsers.add_parser("vms", help="manage virtual machines") parser_vms = subparsers.add_parser("vms", help="manage virtual machines")
vms.register_parser(parser_vms) vms.register_parser(parser_vms)
parser_history = subparsers.add_parser("history", help="manage history")
history.register_parser(parser_history)
if argcomplete: if argcomplete:
argcomplete.autocomplete(parser) argcomplete.autocomplete(parser)

View File

@@ -1,8 +1,6 @@
# !/usr/bin/env python3 # !/usr/bin/env python3
import argparse import argparse
from clan_cli.flakes.add import register_add_parser
from clan_cli.flakes.history import register_list_parser
from clan_cli.flakes.inspect import register_inspect_parser from clan_cli.flakes.inspect import register_inspect_parser
from .create import register_create_parser from .create import register_create_parser
@@ -18,9 +16,5 @@ def register_parser(parser: argparse.ArgumentParser) -> None:
) )
create_parser = subparser.add_parser("create", help="Create a clan flake") create_parser = subparser.add_parser("create", help="Create a clan flake")
register_create_parser(create_parser) register_create_parser(create_parser)
add_parser = subparser.add_parser("add", help="Add a clan flake")
register_add_parser(add_parser)
list_parser = subparser.add_parser("list", help="List recently used flakes")
register_list_parser(list_parser)
inspect_parser = subparser.add_parser("inspect", help="Inspect a clan flake") inspect_parser = subparser.add_parser("inspect", help="Inspect a clan flake")
register_inspect_parser(inspect_parser) register_inspect_parser(inspect_parser)

View File

@@ -1,22 +0,0 @@
# !/usr/bin/env python3
import argparse
from pathlib import Path
from clan_cli.flakes.history import push_history
from ..async_cmd import CmdOut, runforcli
async def add_flake(path: Path) -> dict[str, CmdOut]:
push_history(path)
return {}
def add_flake_command(args: argparse.Namespace) -> None:
runforcli(add_flake, args.path)
# takes a (sub)parser and configures it
def register_add_parser(parser: argparse.ArgumentParser) -> None:
parser.add_argument("path", type=Path, help="Path to the flake", default=Path("."))
parser.set_defaults(func=add_flake_command)

View File

@@ -0,0 +1,22 @@
# !/usr/bin/env python3
import argparse
from .add import register_add_parser
from .list import register_list_parser
from .update import register_update_parser
# takes a (sub)parser and configures it
def register_parser(parser: argparse.ArgumentParser) -> None:
subparser = parser.add_subparsers(
title="command",
description="the command to run",
help="the command to run",
required=True,
)
add_parser = subparser.add_parser("add", help="Add a clan flake")
register_add_parser(add_parser)
list_parser = subparser.add_parser("list", help="List recently used flakes")
register_list_parser(list_parser)
update_parser = subparser.add_parser("update", help="Update a clan flake")
register_update_parser(update_parser)

View File

@@ -1,14 +1,15 @@
# !/usr/bin/env python3 # !/usr/bin/env python3
import argparse import argparse
import dataclasses import dataclasses
import datetime
import json import json
from dataclasses import dataclass import os
from datetime import datetime
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from clan_cli.dirs import user_history_file from clan_cli.flakes.inspect import FlakeConfig, inspect_flake
from ..dirs import user_history_file
from ..locked_open import locked_open from ..locked_open import locked_open
@@ -19,10 +20,12 @@ class EnhancedJSONEncoder(json.JSONEncoder):
return super().default(o) return super().default(o)
@dataclass @dataclasses.dataclass
class HistoryEntry: class HistoryEntry:
path: str path: str
last_used: str last_used: str
dir_datetime: str
flake: FlakeConfig
def list_history() -> list[HistoryEntry]: def list_history() -> list[HistoryEntry]:
@@ -42,7 +45,14 @@ def list_history() -> list[HistoryEntry]:
return logs return logs
def push_history(path: Path) -> list[HistoryEntry]: def get_dir_time(path: Path) -> str:
# Get the last modified dir time in seconds
dir_mtime = os.path.getmtime(path)
dir_datetime = datetime.datetime.fromtimestamp(dir_mtime).isoformat()
return dir_datetime
def add_history(path: Path) -> list[HistoryEntry]:
user_history_file().parent.mkdir(parents=True, exist_ok=True) user_history_file().parent.mkdir(parents=True, exist_ok=True)
logs = list_history() logs = list_history()
@@ -51,24 +61,34 @@ def push_history(path: Path) -> list[HistoryEntry]:
for entry in logs: for entry in logs:
if entry.path == str(path): if entry.path == str(path):
found = True found = True
entry.last_used = datetime.now().isoformat() entry.last_used = datetime.datetime.now().isoformat()
flake = inspect_flake(path, "defaultVM")
flake.flake_url = str(flake.flake_url)
dir_datetime = get_dir_time(path)
history = HistoryEntry(
flake=flake,
dir_datetime=dir_datetime,
path=str(path),
last_used=datetime.datetime.now().isoformat(),
)
if not found: if not found:
logs.append( logs.append(history)
HistoryEntry(path=str(path), last_used=datetime.now().isoformat())
)
f.write(json.dumps(logs, cls=EnhancedJSONEncoder)) f.write(json.dumps(logs, cls=EnhancedJSONEncoder, indent=4))
f.truncate() f.truncate()
return logs return logs
def list_history_command(args: argparse.Namespace) -> None: def add_history_command(args: argparse.Namespace) -> None:
for history_entry in list_history(): add_history(args.path)
print(history_entry.path)
# takes a (sub)parser and configures it # takes a (sub)parser and configures it
def register_list_parser(parser: argparse.ArgumentParser) -> None: def register_add_parser(parser: argparse.ArgumentParser) -> None:
parser.set_defaults(func=list_history_command) parser.add_argument("path", type=Path, help="Path to the flake", default=Path("."))
parser.set_defaults(func=add_history_command)

View File

@@ -0,0 +1,14 @@
# !/usr/bin/env python3
import argparse
from .add import list_history
def list_history_command(args: argparse.Namespace) -> None:
for history_entry in list_history():
print(history_entry.path)
# takes a (sub)parser and configures it
def register_list_parser(parser: argparse.ArgumentParser) -> None:
parser.set_defaults(func=list_history_command)

View File

@@ -0,0 +1,35 @@
# !/usr/bin/env python3
import argparse
import copy
import datetime
import json
from pathlib import Path
from ..dirs import user_history_file
from ..locked_open import locked_open
from .add import EnhancedJSONEncoder, HistoryEntry, get_dir_time, list_history
def update_history() -> list[HistoryEntry]:
with locked_open(user_history_file(), "w+") as f:
logs = list_history()
new_logs = []
for entry in logs:
new_entry = copy.deepcopy(entry)
new_time = get_dir_time(Path(entry.path))
if new_time != entry.dir_datetime:
new_entry.dir_datetime = new_time
new_entry.last_used = datetime.datetime.now().isoformat()
new_logs.append(new_entry)
f.write(json.dumps(new_logs, cls=EnhancedJSONEncoder, indent=4))
f.truncate()
return new_logs
def add_update_command(args: argparse.Namespace) -> None:
update_history()
# takes a (sub)parser and configures it
def register_update_parser(parser: argparse.ArgumentParser) -> None:
parser.set_defaults(func=add_update_command)

View File

@@ -17,7 +17,7 @@ from clan_cli.webui.api_outputs import (
) )
from ...async_cmd import run from ...async_cmd import run
from ...flakes import add, create from ...flakes import create
from ...nix import nix_command, nix_flake_show from ...nix import nix_command, nix_flake_show
from ..tags import Tags from ..tags import Tags
@@ -45,16 +45,6 @@ async def get_attrs(url: AnyUrl | Path) -> list[str]:
return flake_attrs return flake_attrs
@router.post("/api/flake/history", tags=[Tags.flake])
async def flake_history_append(flake_dir: Path) -> None:
await add.add_flake(flake_dir)
@router.get("/api/flake/history", tags=[Tags.flake])
async def flake_history_list() -> list[Path]:
return []
# TODO: Check for directory traversal # TODO: Check for directory traversal
@router.get("/api/flake/attrs", tags=[Tags.flake]) @router.get("/api/flake/attrs", tags=[Tags.flake])
async def inspect_flake_attrs(url: AnyUrl | Path) -> FlakeAttrResponse: async def inspect_flake_attrs(url: AnyUrl | Path) -> FlakeAttrResponse:

View File

@@ -7,7 +7,7 @@ from fixtures_flakes import FlakeForTest
from pytest import CaptureFixture from pytest import CaptureFixture
from clan_cli.dirs import user_history_file from clan_cli.dirs import user_history_file
from clan_cli.flakes.history import HistoryEntry from clan_cli.history.add import HistoryEntry
if TYPE_CHECKING: if TYPE_CHECKING:
pass pass

View File

@@ -73,45 +73,6 @@ class VM:
# start/end indexes can be used optionally for pagination # start/end indexes can be used optionally for pagination
def get_initial_vms(start: int = 0, end: int | None = None) -> list[VM]: def get_initial_vms(start: int = 0, end: int | None = None) -> list[VM]:
# vms = [
# VM(
# base=VMBase(
# icon=assets.loc / "cybernet.jpeg",
# name="Cybernet Clan",
# url="clan://cybernet.lol",
# _path=Path(__file__).parent.parent / "test_democlan",
# status=False,
# ),
# ),
# VM(
# base=VMBase(
# icon=assets.loc / "zenith.jpeg",
# name="Zenith Clan",
# url="clan://zenith.lol",
# _path=Path(__file__).parent.parent / "test_democlan",
# status=False,
# )
# ),
# VM(
# base=VMBase(
# icon=assets.loc / "firestorm.jpeg",
# name="Firestorm Clan",
# url="clan://firestorm.lol",
# _path=Path(__file__).parent.parent / "test_democlan",
# status=False,
# ),
# ),
# VM(
# base=VMBase(
# icon=assets.loc / "placeholder.jpeg",
# name="Placeholder Clan",
# url="clan://demo.lol",
# _path=Path(__file__).parent.parent / "test_democlan",
# status=True,
# ),
# ),
# ]
vm_list = [] vm_list = []
# TODO: list_history() should return a list of dicts, not a list of paths # TODO: list_history() should return a list of dicts, not a list of paths