From 56292725d9afa0dbb0eb855931df195a3f0ba789 Mon Sep 17 00:00:00 2001 From: Ryan Reed Date: Tue, 5 Sep 2023 21:35:27 -0400 Subject: [PATCH] Initial rewrite to utilize central file --- src/transpose/__init__.py | 4 +- src/transpose/console.py | 155 ++++++++++++++------- src/transpose/transpose.py | 267 ++++++++++++++++++++++++++----------- src/transpose/utils.py | 60 --------- 4 files changed, 301 insertions(+), 185 deletions(-) diff --git a/src/transpose/__init__.py b/src/transpose/__init__.py index c8d96b7..6526dc8 100644 --- a/src/transpose/__init__.py +++ b/src/transpose/__init__.py @@ -5,8 +5,8 @@ from importlib.metadata import version from .logger import create_logger DEFAULT_XDG_PATH = os.environ.get("XDG_DATA_HOME", f"{os.environ['HOME']}/.local/share") -DEFAULT_CACHE_FILENAME = ".transpose.json" -DEFAULT_STORE_PATH = f"{DEFAULT_XDG_PATH}/transpose" +STORE_PATH = f"{DEFAULT_XDG_PATH}/transpose" +DEFAULT_STORE_PATH = os.environ.get("TRANSPOSE_STORE_PATH", STORE_PATH) version = version("transpose") diff --git a/src/transpose/console.py b/src/transpose/console.py index 9976cd9..20259c9 100644 --- a/src/transpose/console.py +++ b/src/transpose/console.py @@ -1,47 +1,63 @@ import argparse -import os -from transpose import Transpose, version, DEFAULT_STORE_PATH, DEFAULT_CACHE_FILENAME +from transpose import Transpose, version, DEFAULT_STORE_PATH +from .exceptions import TransposeError def entry_point() -> None: args = parse_arguments() + config_path = f"{args.store_path}/transpose.json" + + try: + run(args, config_path) + except TransposeError as e: + print(f"Tranpose Error: {e}") - t = Transpose( - target_path=args.target_path, - cache_filename=args.cache_filename, - ) + +def run(args, config_path) -> None: + t = Transpose(config_path) if args.action == "apply": - t.apply() - elif args.action == "create": - t.create(stored_path=args.stored_path) + t.apply(args.name, force=args.force) elif args.action == "restore": - t.restore() + t.restore(args.name, force=args.force) elif args.action == "store": - t.store(store_path=args.store_path, name=args.name) + t.store(args.name, args.target_path) + elif args.action == "config": + if args.config_action == "add": + t.config.add(args.name, args.path) + t.config.save(config_path) + elif args.config_action == "get": + print(t.config.get(args.name)) + elif args.config_action == "list": + for name in t.config.entries: + print(f"\t{name:<30} -> {t.config.entries[name].path}") + elif args.config_action == "remove": + t.config.remove(args.name) + t.config.save(config_path) + elif args.config_action == "update": + t.config.update(args.name, args.path) + t.config.save(config_path) def parse_arguments(args=None): - cache_filename = os.environ.get("TRANSPOSE_CACHE_FILENAME", DEFAULT_CACHE_FILENAME) - store_path = os.environ.get("TRANSPOSE_STORE_PATH", DEFAULT_STORE_PATH) - base_parser = argparse.ArgumentParser(add_help=False) - base_parser.add_argument( - "--cache-filename", - dest="cache_filename", - nargs="?", - default=cache_filename, - help="The name of the cache file added to the target directory (default: %(default)s)", - ) parser = argparse.ArgumentParser( parents=[base_parser], description=""" - Move and symlink a path for easier management + Move and symlink a path for easy, central management """, ) parser.add_argument("--version", action="version", version=f"Transpose {version}") + parser.add_argument( + "-s", + "--store-path", + dest="store_path", + nargs="?", + default=DEFAULT_STORE_PATH, + help="The location to store the moved entities (default: %(default)s)", + ) subparsers = parser.add_subparsers( help="Transpose Action", dest="action", required=True @@ -49,27 +65,14 @@ def parse_arguments(args=None): apply_parser = subparsers.add_parser( "apply", - help="Recreate the symlink from the cache file (useful after moving store loction)", + help="Recreate the symlink for an entity (useful after moving store locations)", parents=[base_parser], ) apply_parser.add_argument( - "target_path", - help="The path to the directory to locate the cache file", - ) - - create_parser = subparsers.add_parser( - "create", - help="Create the cache file from an already stored path. Only creates the cache file.", - parents=[base_parser], - ) - create_parser.add_argument( - "target_path", - help="The path to the directory that should by a symlink", - ) - create_parser.add_argument( - "stored_path", - help="The path that is currently stored (the target of the symlink)", + "name", + help="The name of the stored entity to apply", ) + apply_parser.add_argument("--force", dest="force", action="store_true") restore_parser = subparsers.add_parser( "restore", @@ -77,9 +80,10 @@ def parse_arguments(args=None): parents=[base_parser], ) restore_parser.add_argument( - "target_path", - help="The path to the directory to restore", + "name", + help="The name of the stored entity to restore", ) + restore_parser.add_argument("--force", dest="force", action="store_true") store_parser = subparsers.add_parser( "store", @@ -96,13 +100,68 @@ def parse_arguments(args=None): default=None, help="The name of the directory that will be created in the store path (default: target_path)", ) - store_parser.add_argument( - "-s", - "--store-path", - dest="store_path", - nargs="?", - default=store_path, - help="The path to where the targets should be stored (default: %(default)s)", + + config_parser = subparsers.add_parser( + "config", + help="Modify the transpose config file without any filesystem changes", + parents=[base_parser], + ) + config_subparsers = config_parser.add_subparsers( + help="Transpose Config Action", dest="config_action", required=True + ) + + config_add_parser = config_subparsers.add_parser( + "add", + help="Add an entry manually to the tranpose config", + parents=[base_parser], + ) + config_add_parser.add_argument( + "name", + help="The name of the entry in the store path", + ) + config_add_parser.add_argument( + "path", + help="The path of the directory that should be symlinked to the store", + ) + + config_get_parser = config_subparsers.add_parser( + "get", + help="Retrieve the settings of a specific entity, such as the path", + parents=[base_parser], + ) + config_get_parser.add_argument( + "name", + help="The name of the entry in the store path", + ) + + config_subparsers.add_parser( + "list", + help="List the names of all entities in the transpose config", + parents=[base_parser], + ) + + config_remove_parser = config_subparsers.add_parser( + "remove", + help="Remove an entry from the config", + parents=[base_parser], + ) + config_remove_parser.add_argument( + "name", + help="The name of the entry in the store path", + ) + + config_update_parser = config_subparsers.add_parser( + "update", + help="Update an entry of the transpose config", + parents=[base_parser], + ) + config_update_parser.add_argument( + "name", + help="The name of the entry in the store path", + ) + config_update_parser.add_argument( + "path", + help="The path of the directory that should be symlinked to the store", ) return parser.parse_args(args) diff --git a/src/transpose/transpose.py b/src/transpose/transpose.py index c359c66..15c06dc 100644 --- a/src/transpose/transpose.py +++ b/src/transpose/transpose.py @@ -1,111 +1,228 @@ -import pathlib +from dataclasses import asdict, dataclass, field +from pathlib import Path +# from typing import Self + +import json + +from . import version as transpose_version from .exceptions import TransposeError -from .utils import check_path, create_cache, get_cache, move, remove, symlink +from .utils import move, remove, symlink -class Transpose: - def __init__( - self, - target_path: str, - cache_filename: str = None, - ) -> None: - self.target_path = pathlib.Path(target_path) +@dataclass +class TransposeEntry: + name: str + path: str - if not cache_filename: - cache_filename = ".transpose.json" - self.cache_filename = cache_filename - self.cache_path = pathlib.Path(self.target_path).joinpath(cache_filename) - def apply(self) -> None: +@dataclass +class TransposeConfig: + entries: dict = field(default_factory=dict) + version: str = field(default=transpose_version) + + def add(self, name: str, path: str) -> None: """ - Recreate the symlink from an existing cache file + Add a new entry to the entries + + Args: + name: The name of the entry (must not exist) + path: The path where the entry originally exists + + Returns: + None """ - if not self.cache_path.exists(): - raise TransposeError( - f"Cache file does not exist indicating target is not managed by Transpose: {self.cache_path}" - ) + if self.entries.get(name): + raise TransposeError(f"'{name}' already exists") + + self.entries[name] = TransposeEntry(name=name, path=path) - cache = get_cache(self.cache_path) - original_path = pathlib.Path(cache["original_path"]).expanduser() + def get(self, name: str) -> TransposeEntry: + """ + Get an entry by the name - if original_path.is_symlink(): - remove(original_path) + Args: + name: The name of the entry (must exist) - symlink(target_path=self.cache_path.parent, symlink_path=original_path) + Returns: + TransposeEntry + """ + try: + return self.entries[name] + except KeyError: + raise TransposeError(f"'{name}' does not exist in Transpose config entries") - def create(self, stored_path: str) -> None: + def remove(self, name: str) -> None: """ - Create the cache file from the target directory and stored directory + Remove an entry by name - This is useful if a path is already stored somewhere else but the cache file is missing + Args: + name: The name of the entry (must exist) - Ideally, the target should be a symlink or not exist so a restore or apply can function + Returns: + None """ - stored_path = pathlib.Path(stored_path) - if not stored_path.exists(): - raise TransposeError(f"Stored path does not exist: {stored_path}") + try: + del self.entries[name] + except KeyError: + raise TransposeError(f"'{name}' does not exist in Transpose config entries") - self.cache_path = stored_path.joinpath(self.cache_filename) + def update(self, name: str, path: str) -> None: + """ + Update an entry by name - create_cache( - cache_path=self.cache_path, - original_path=self.target_path, - ) + Args: + name: The name of the entry (must exist) + path: The path where the entry originally exists - def restore(self) -> None: + Returns: + None """ - Restores a previously Transpose managed directory to it's previous location. + try: + self.entries[name].path = path + except KeyError: + raise TransposeError(f"'{name}' does not exist in Transpose config entries") + + @staticmethod + def load(config_path: str): # -> Self: + in_config = json.load(open(config_path, "r")) + config = TransposeConfig() + try: + for name in in_config["entries"]: + config.add(name, in_config["entries"][name]["path"]) + except (KeyError, TypeError) as e: + raise TransposeError(f"Unrecognized Transpose config file format: {e}") + + return config + + def save(self, config_path: str) -> None: """ - if not self.cache_path.exists(): - raise TransposeError( - f"Cache file does not exist indicating target is not managed by Transpose: {self.cache_path}" - ) - if not self.target_path.exists(): - raise TransposeError(f"Target path does not exist: {self.target_path}") + Save the Config to a location in JSON format - cache = get_cache(self.cache_path) - original_path = pathlib.Path(cache["original_path"]).expanduser() + Args: + path: The path to save the json file - if original_path.is_symlink(): - remove(original_path) - elif original_path.exists(): - raise TransposeError( - f"Original path in cache file already exists: {original_path}" - ) + Returns: + None + """ + config_path = Path(config_path) + config_path.parent.mkdir(parents=True, exist_ok=True) + + with open(str(config_path), "w") as f: + json.dump(self.to_dict(), f) + + def to_dict(self) -> dict: + return asdict(self) - try: - move(source=self.target_path, destination=original_path) - except FileNotFoundError: - raise TransposeError( - f"Original path, {original_path}, does not exist. Use '-f' to create the path" - ) - new_cache_path = pathlib.Path(original_path).joinpath(self.cache_filename) - remove(new_cache_path) +class Transpose: + config: TransposeConfig + config_path: Path + store_path: Path + + def __init__(self, config_path: str) -> None: + self.config = TransposeConfig.load(config_path) + self.config_path = Path(config_path) + self.store_path = self.config_path.parent + + if not self.store_path.exists(): + self.store_path.mkdir(parents=True) - def store(self, store_path: str, name: str = None) -> None: + def apply(self, name: str, force: bool = False) -> None: """ - Moves a directory to a central location and creates a symlink to the old path. + Create/recreate the symlink to an existing entry + + Args: + name: The name of the entry (must exist) + force: If enabled and path already exists, move the path to '{path}-bak' + + Returns: + None """ - if name is None: - name = self.target_path.name + if not self.config.entries.get(name): + raise TransposeError(f"Entry does not exist: '{name}'") + + if self.config.entries[name].path.exists(): + if self.config.entries[name].path.is_symlink(): + remove(self.config.entries[name].path) + elif force: # Backup the existing path, just in case + move( + self.config.entries[name].path, + self.config.entries[name].path.joinpath("-bak"), + ) + else: + raise TransposeError( + f"Entry path already exists, cannot restore (force required): '{self.config.entries[name].path}'" + ) + + symlink( + target_path=self.store_path.joinpath(name), + symlink_path=self.config.entries[name].path, + ) - new_location = pathlib.Path(store_path).joinpath(name) + def restore(self, name: str, force: bool = False) -> None: + """ + Remove the symlink and move the stored entry back to it's original path + + Args: + name: The name of the entry (must exist) + force: If enabled and path already exists, move the path to '{path}-bak' + + Returns: + None + """ + if not self.config.entries.get(name): + raise TransposeError(f"Could not locate entry by name: '{name}'") + + if self.config.entries[name].path.exists(): + if self.config.entries[name].path.is_symlink(): + remove(self.config.entries[name].path) + elif force: # Backup the existing path, just in case + move( + self.config.entries[name].path, + self.config.entries[name].path.joinpath("-bak"), + ) + else: + raise TransposeError( + f"Entry path already exists, cannot restore (force required): '{self.config.entries[name].path}'" + ) + + move(self.store_path.joinpath(name), self.config.entries[name].path) + + self.config.remove(name) + self.config.save(self.config_path) + + def store(self, name: str, source_path: str) -> None: + """ + Move the source path to the store path, create a symlink, and update the config + + Args: + name: The name of the entry (must exist) + source_path: The directory or file to be stored - if not check_path(path=self.target_path): + Returns: + None + """ + if self.config.entries.get(name): raise TransposeError( - f"Target path, {self.target_path}, does not exist. Cannot continue." + f"Entry already exists: '{name}' ({self.config.entries[name].path})" ) - if check_path(path=new_location): + + storage_path = self.store_path.joinpath(name) + if storage_path.exists(): + raise TransposeError(f"Store path already exists: '{storage_path}'") + + source_path = self.config.entries[name].path + if not source_path.exists(): + raise TransposeError(f"Source path does not exist: '{source_path}'") + + if not source_path.is_dir() and not source_path.is_file(): raise TransposeError( - f"Store path, {new_location}, already exists. Cannot continue." + f"Source path must be a directory or file: '{source_path}'" ) - create_cache( - cache_path=self.cache_path, - original_path=self.target_path, - ) + move(source=source_path, destination=storage_path) + symlink(target_path=storage_path, symlink_path=source_path) - move(source=self.target_path, destination=new_location) - symlink(target_path=new_location, symlink_path=self.target_path) + self.config.add(name, source_path) + self.config.save(self.config_path) diff --git a/src/transpose/utils.py b/src/transpose/utils.py index ef7735a..be7a115 100644 --- a/src/transpose/utils.py +++ b/src/transpose/utils.py @@ -1,66 +1,6 @@ -import json import shutil from pathlib import Path -from typing import Dict - -from . import version - - -def check_path(path: Path, is_symlink: bool = False) -> bool: - """ - Checks whether a path exists and is a directory (doesn't support single files) - - Args: - path: The location to the path being verified - is_symlink: Should this path be a symlink? - - Returns: - bool - """ - if is_symlink and not path.is_symlink(): - return False - if not is_symlink and path.is_symlink(): - return False - if not path.exists(): - return False - if not path.is_dir(): - return False - - return True - - -def create_cache(cache_path: Path, original_path: Path) -> None: - """ - Create a cache file for transpose settings in the stored directory - - Args: - cache_path: Path to store the cache file - original_path: Path where the stored directory originated - - Returns: - None - """ - template = { - "version": version, - "original_path": str(original_path.absolute()).replace(str(Path.home()), "~"), - } - - with open(str(cache_path), "w") as f: - json.dump(template, f) - - -def get_cache(cache_path: Path) -> Dict: - """ - Read a JSON cache file - - Args: - cache_path: Path to the Transpose cache file - - Returns: - dict: Cache file contents - """ - return json.load(open(cache_path, "r")) def move(source: Path, destination: Path) -> None: