| @ -0,0 +1,5 @@ | |||||
| [flake8] | |||||
| ignore = E203, E266, E501, W503, F403, F401 | |||||
| max-line-length = 120 | |||||
| max-complexity = 18 | |||||
| select = B,C,E,F,W,T4,B9 | |||||
| @ -0,0 +1,120 @@ | |||||
| # Swap files | |||||
| *.sw[a-p] | |||||
| # Byte-compiled / optimized / DLL files | |||||
| __pycache__/ | |||||
| *.py[cod] | |||||
| *$py.class | |||||
| # C extensions | |||||
| *.so | |||||
| # Distribution / packaging | |||||
| .Python | |||||
| build/ | |||||
| develop-eggs/ | |||||
| dist/ | |||||
| downloads/ | |||||
| eggs/ | |||||
| .eggs/ | |||||
| lib/ | |||||
| lib64/ | |||||
| parts/ | |||||
| sdist/ | |||||
| var/ | |||||
| wheels/ | |||||
| *.egg-info/ | |||||
| .installed.cfg | |||||
| *.egg | |||||
| MANIFEST | |||||
| poetry.lock | |||||
| # PyInstaller | |||||
| # Usually these files are written by a python script from a template | |||||
| # before PyInstaller builds the exe, so as to inject date/other infos into it. | |||||
| *.manifest | |||||
| *.spec | |||||
| # Installer logs | |||||
| pip-log.txt | |||||
| pip-delete-this-directory.txt | |||||
| # Unit test / coverage reports | |||||
| htmlcov/ | |||||
| .tox/ | |||||
| .nox/ | |||||
| .coverage | |||||
| .coverage.* | |||||
| .cache | |||||
| nosetests.xml | |||||
| coverage.xml | |||||
| *.cover | |||||
| .hypothesis/ | |||||
| .pytest_cache/ | |||||
| .mypy_cache/ | |||||
| # Translations | |||||
| *.mo | |||||
| *.pot | |||||
| # Django stuff: | |||||
| *.log | |||||
| local_settings.py | |||||
| db.sqlite3 | |||||
| # Flask stuff: | |||||
| instance/ | |||||
| .webassets-cache | |||||
| # Scrapy stuff: | |||||
| .scrapy | |||||
| # Sphinx documentation | |||||
| docs/_build/ | |||||
| # PyBuilder | |||||
| target/ | |||||
| # Jupyter Notebook | |||||
| .ipynb_checkpoints | |||||
| # IPython | |||||
| profile_default/ | |||||
| ipython_config.py | |||||
| # pyenv | |||||
| .python-version | |||||
| # celery beat schedule file | |||||
| celerybeat-schedule | |||||
| # SageMath parsed files | |||||
| *.sage.py | |||||
| # Environments | |||||
| .env | |||||
| .secrets | |||||
| .venv | |||||
| env/ | |||||
| venv/ | |||||
| ENV/ | |||||
| env.bak/ | |||||
| venv.bak/ | |||||
| # Spyder project settings | |||||
| .spyderproject | |||||
| .spyproject | |||||
| # Rope project settings | |||||
| .ropeproject | |||||
| # mkdocs documentation | |||||
| /site | |||||
| # mypy | |||||
| .mypy_cache/ | |||||
| .dmypy.json | |||||
| dmypy.json | |||||
| # Pyre type checker | |||||
| .pyre/ | |||||
| @ -0,0 +1,9 @@ | |||||
| repos: | |||||
| - repo: https://github.com/psf/black | |||||
| rev: "22.6.0" | |||||
| hooks: | |||||
| - id: black | |||||
| - repo: https://gitlab.com/pycqa/flake8 | |||||
| rev: "3.8.4" | |||||
| hooks: | |||||
| - id: flake8 | |||||
| @ -0,0 +1,70 @@ | |||||
| # Transpose | |||||
| A tool for moving and symlinking directories to a central location | |||||
| ## Inspiration | |||||
| I've been using linux as my main machine for a while and wanted a central directory to backup as backing up my entire `HOME` directory was a mess. I found moving directories and symlinking worked great. I created a simple project when learning python (I called symlinker) and used it for a while but found it annoying to configure and work with. | |||||
| I recently found I could use a tool like this for a SteamDeck and decided to start from scratch with better code and easier to use. | |||||
| This is the current result, although it still needs a lot of work as I'm sure I'm not doing things particularly well. | |||||
| ## Installation | |||||
| TODO | |||||
| ## Configuration | |||||
| There are a few environment variables that can be defined to override defaults | |||||
| ``` | |||||
| TRANSPOSE_STORE_PATH="$XDG_DATA_HOME/transpose" | |||||
| TRANSPOSE_CACHE_FILENAME=".transpose.json" | |||||
| ``` | |||||
| ## Usage | |||||
| ### Storing a Directory | |||||
| Storing a directory will: | |||||
| 1. Move a `target` to `$STORE_PATH/{name}` | |||||
| 2. Symlink `target` to `$STORE_PATH/{name}` | |||||
| 3. Create a cache file at `$STORE_PATH/{name}/.transpose.json` to store the original target path | |||||
| ``` | |||||
| transpose store "My Documents" /home/user/Documents | |||||
| ``` | |||||
| The above will (assuming using all the defaults): | |||||
| 1. Move `/home/user/Documents` to `$XDG_DATA_HOME/transpose/My Documents` | |||||
| 2. Symlink `/home/user/Documents` to `$XDG_DATA_HOME/transpose/My Documents` | |||||
| ### Restoring a Store Directory | |||||
| Restoring a directory will: | |||||
| 1. Remove the old symlink in the `original_path` of the cache file, `$STORE_PATH/{name}/.transpose.json` | |||||
| 2. Move the stored directory to the `original_path` | |||||
| ``` | |||||
| transpose restore "$XDG_DATA_HOME/transpose/My Documents" | |||||
| ``` | |||||
| The above will (assuming all the defaults): | |||||
| 1. Remove the symlink at `/home/user/Documents` (from cache file) | |||||
| 2. Move `$XDG_DATA_HOME/transpose/My Documents` to `/home/user/Documents` | |||||
| ## License | |||||
| TODO | |||||
| @ -0,0 +1,15 @@ | |||||
| from os import environ | |||||
| from pydantic import BaseSettings | |||||
| default_xdg_path = environ.get("XDG_DATA_HOME", f"{environ['HOME']}/.local/share") | |||||
| class Config(BaseSettings): | |||||
| store_path: str = f"{default_xdg_path}/transpose" | |||||
| cache_filename: str = ".transpose.json" | |||||
| class Config: | |||||
| env_file = ".env" | |||||
| env_file_encoding = "utf-8" | |||||
| env_nested_delimiter = "__" | |||||
| env_prefix = "TRANSPOSE_" | |||||
| @ -0,0 +1,63 @@ | |||||
| import argparse | |||||
| import os | |||||
| from config import Config | |||||
| from transpose import Transpose, version | |||||
| config = Config() | |||||
| def main() -> None: | |||||
| args = parse_arguments() | |||||
| t = Transpose( | |||||
| target_path=args.target_path, | |||||
| store_path=config.store_path, | |||||
| cache_filename=config.cache_filename, | |||||
| ) | |||||
| if args.action == "restore": | |||||
| t.restore() | |||||
| elif args.action == "store": | |||||
| t.store(name=args.name) | |||||
| def parse_arguments(): | |||||
| base_parser = argparse.ArgumentParser(add_help=False) | |||||
| parser = argparse.ArgumentParser( | |||||
| parents=[base_parser], | |||||
| description=""" | |||||
| Move and symlink a path for easier management | |||||
| """, | |||||
| ) | |||||
| parser.add_argument("--version", action="version", version=f"Transpose {version}") | |||||
| subparsers = parser.add_subparsers(help="Transpose Action", dest="action") | |||||
| restore_parser = subparsers.add_parser( | |||||
| "restore", | |||||
| help="Move a transposed directory back to it's original location", | |||||
| parents=[base_parser], | |||||
| ) | |||||
| restore_parser.add_argument( | |||||
| "target_path", | |||||
| help="The path to the directory to restore", | |||||
| ) | |||||
| store_parser = subparsers.add_parser( | |||||
| "store", help="Move target and create symlink in place", parents=[base_parser] | |||||
| ) | |||||
| store_parser.add_argument( | |||||
| "name", | |||||
| help="The name of the directory that will be created in the store path", | |||||
| ) | |||||
| store_parser.add_argument( | |||||
| "target_path", | |||||
| help="The path to the directory to be stored", | |||||
| ) | |||||
| return parser.parse_args() | |||||
| if __name__ == "__main__": | |||||
| main() | |||||
| @ -0,0 +1,24 @@ | |||||
| [tool.poetry] | |||||
| name = "transpose" | |||||
| version = "0.9.0" | |||||
| description = "Move and symlink a path" | |||||
| authors = ["Ryan Reed"] | |||||
| [tool.poetry.dependencies] | |||||
| python = "^3.7" | |||||
| rich = "*" | |||||
| click = "^8.1.3" | |||||
| [tool.poetry.dev-dependencies] | |||||
| black = "==22.6" | |||||
| flake8 = "==3.8.4" | |||||
| pre-commit = "*" | |||||
| pydantic = "*" | |||||
| pytest = "*" | |||||
| pytest-sugar = "*" | |||||
| python-dotenv = "*" | |||||
| [build-system] | |||||
| requires = ["poetry>=0.12"] | |||||
| build-backend = "poetry.masonry.api" | |||||
| @ -0,0 +1,2 @@ | |||||
| [pytest] | |||||
| testpaths = tests | |||||
| @ -0,0 +1,116 @@ | |||||
| import json | |||||
| import os | |||||
| from pathlib import Path, PurePath | |||||
| from contextlib import contextmanager | |||||
| from tempfile import TemporaryDirectory | |||||
| from config import Config | |||||
| from transpose import version | |||||
| from transpose.utils import check_path, create_cache, get_cache, move, remove, symlink | |||||
| TARGET_DIR = "source" | |||||
| STORE_DIR = "destination" | |||||
| SYMLINK_DIR = "symlink_test" | |||||
| CACHE_FILE_CONTENTS = {"version": version, "original_path": TARGET_DIR} | |||||
| config = Config() | |||||
| @contextmanager | |||||
| def setup(): | |||||
| old_dir = os.getcwd() | |||||
| with TemporaryDirectory("tests-temp") as td: | |||||
| try: | |||||
| os.chdir(td) | |||||
| os.mkdir(TARGET_DIR) | |||||
| os.mkdir(STORE_DIR) | |||||
| os.symlink(TARGET_DIR, SYMLINK_DIR) | |||||
| cache_path = Path(PurePath(TARGET_DIR, config.cache_filename)) | |||||
| with open(str(cache_path), "w") as f: | |||||
| json.dump(CACHE_FILE_CONTENTS, f) | |||||
| yield | |||||
| finally: | |||||
| os.chdir(old_dir) | |||||
| @setup() | |||||
| def test_check_path(): | |||||
| existing_dir = Path(TARGET_DIR) | |||||
| nonexisting_dir = Path("nonexistent") | |||||
| symlink_dir = Path(SYMLINK_DIR) | |||||
| cache_path = Path(PurePath(TARGET_DIR, config.cache_filename)) | |||||
| assert check_path(existing_dir) is True | |||||
| assert check_path(nonexisting_dir) is False | |||||
| assert check_path(symlink_dir, is_symlink=True) is True | |||||
| assert check_path(cache_path) is False | |||||
| @setup() | |||||
| def test_cache_create(): | |||||
| cache_file = "test_cache_file.json" | |||||
| cache_path = Path(PurePath(TARGET_DIR, cache_file)) | |||||
| original_path = Path("/tmp/some/random/path") | |||||
| create_cache(cache_path=cache_path, original_path=original_path) | |||||
| cache = json.load(open(cache_path, "r")) | |||||
| assert cache_path.exists() | |||||
| assert cache["original_path"] == str(original_path) | |||||
| assert cache["version"] == version | |||||
| @setup() | |||||
| def test_cache_get(): | |||||
| cache_path = Path(PurePath(TARGET_DIR, config.cache_filename)) | |||||
| cache = get_cache(cache_path) | |||||
| assert cache["version"] == CACHE_FILE_CONTENTS["version"] | |||||
| assert cache["original_path"] == CACHE_FILE_CONTENTS["original_path"] | |||||
| @setup() | |||||
| def test_file_move(): | |||||
| source_path = Path(TARGET_DIR) | |||||
| destination_path = Path(STORE_DIR) | |||||
| move(source=source_path.absolute(), destination=destination_path.absolute()) | |||||
| assert not source_path.exists() | |||||
| assert destination_path.exists() | |||||
| @setup() | |||||
| def test_file_remove(): | |||||
| cache_path = Path(PurePath(TARGET_DIR, config.cache_filename)) | |||||
| symlink_filepath = Path(PurePath(TARGET_DIR, SYMLINK_DIR)) | |||||
| target_filepath = Path(TARGET_DIR) | |||||
| remove(path=cache_path) | |||||
| remove(path=symlink_filepath) | |||||
| remove(path=target_filepath) | |||||
| assert not cache_path.exists() # Should be able to remove files | |||||
| assert not symlink_filepath.exists() # Should be able to remove symlinks | |||||
| assert target_filepath.exists() # Should not be able to remove directories | |||||
| @setup() | |||||
| def test_file_symlink(): | |||||
| symlink_name = "test_link" | |||||
| symlink_filepath = Path(symlink_name) | |||||
| target_filepath = Path(TARGET_DIR) | |||||
| symlink(target_path=target_filepath, symlink_path=symlink_filepath) | |||||
| assert target_filepath.exists() | |||||
| assert symlink_filepath.is_symlink() | |||||
| assert symlink_filepath.readlink() == target_filepath | |||||
| @ -0,0 +1,8 @@ | |||||
| from .logger import create_logger | |||||
| version_info = (0, 9, 0) | |||||
| version = ".".join(str(c) for c in version_info) | |||||
| logger = create_logger(__package__) | |||||
| from .transpose import Transpose # noqa: E402 | |||||
| @ -0,0 +1,6 @@ | |||||
| import logging | |||||
| def create_logger(logger_name: str) -> logging: | |||||
| logger = logging.getLogger(logger_name) | |||||
| return logger | |||||
| @ -0,0 +1,129 @@ | |||||
| from pathlib import Path, PurePath | |||||
| from .utils import check_path, create_cache, get_cache, move, remove, symlink | |||||
| class Transpose: | |||||
| def __init__( | |||||
| self, target_path: str, store_path: str, cache_filename: str = None | |||||
| ) -> None: | |||||
| self.target_path = Path(target_path) | |||||
| self.store_path = Path(store_path) | |||||
| if not cache_filename: | |||||
| cache_filename = ".transpose.json" | |||||
| self.cache_filename = cache_filename | |||||
| self.cache_path = Path(PurePath(self.target_path, cache_filename)) | |||||
| def restore(self) -> None: | |||||
| """ | |||||
| Restores a previously Transpose managed directory to it's previous location. | |||||
| Performs: | |||||
| 1. Verify `cache_file` exists | |||||
| 2. Verify `target_path` exists | |||||
| 3. Verify if `original_path` in cache is a symlink | |||||
| a. Remove if true | |||||
| 4. Verify `original_path` doesn't exist | |||||
| 5. Move `target_path` to `original_path` based on cache file settings | |||||
| Args: | |||||
| None | |||||
| Returns: | |||||
| None | |||||
| Raises: | |||||
| ValueError: Any paths not existing | |||||
| RuntimeError: Any error during the actual path changes | |||||
| """ | |||||
| if not self.cache_path.exists(): | |||||
| raise ValueError( | |||||
| f"Cache file does not exist indicating target is not managed by Transpose: {self.cache_path}" | |||||
| ) | |||||
| if not self.target_path.exists(): | |||||
| raise ValueError(f"Target path does not exist: {self.target_path}") | |||||
| cache = get_cache(self.cache_path) | |||||
| original_path = Path(cache["original_path"]) | |||||
| if original_path.is_symlink(): | |||||
| try: | |||||
| remove(original_path) | |||||
| except: # noqa: E722 # TODO | |||||
| raise RuntimeError( | |||||
| f"Failed to remove symlink in original path: {original_path}" | |||||
| ) | |||||
| elif original_path.exists(): | |||||
| raise ValueError( | |||||
| f"Original path in cache file already exists: {original_path}" | |||||
| ) | |||||
| try: | |||||
| move(source=self.target_path, destination=original_path) | |||||
| except: # noqa: E722 # TODO | |||||
| raise RuntimeError( | |||||
| f"Failed to move target to original location: {self.target_path} -> {original_path}" | |||||
| ) | |||||
| new_cache_path = Path(PurePath(original_path, self.cache_filename)) | |||||
| try: | |||||
| remove(new_cache_path) | |||||
| except: # noqa: E722 # TODO | |||||
| raise RuntimeError( | |||||
| f"Failed to remove previous cache file: {new_cache_path}" | |||||
| ) | |||||
| def store(self, name: str) -> None: | |||||
| """ | |||||
| Moves a directory to a central location and creates a symlink to the old path. | |||||
| Performs: | |||||
| 1. Verify `target_path` exists | |||||
| 2. Verify `store_path` exists | |||||
| 3. Create the cache file | |||||
| 4. Move the `target_path` to `store_path/name` | |||||
| 5. Create symlink `target_path` -> `store_path/name` | |||||
| Args: | |||||
| name: The directory name to give the new location | |||||
| Returns: | |||||
| None | |||||
| Raises | |||||
| ValueError: Any paths not existing | |||||
| RuntimeError: Any error during the actual path changes | |||||
| """ | |||||
| new_location = Path(PurePath(self.store_path, name)) | |||||
| if not check_path(path=self.target_path): | |||||
| raise ValueError( | |||||
| f"Target path, {self.target_path}, does not exist. Cannot continue." | |||||
| ) | |||||
| if check_path(path=new_location): | |||||
| raise ValueError( | |||||
| f"Store path, {new_location}, already exists. Cannot continue." | |||||
| ) | |||||
| try: | |||||
| create_cache( | |||||
| cache_path=self.cache_path, | |||||
| original_path=self.target_path, | |||||
| ) | |||||
| except: # noqa: E722 # TODO | |||||
| raise RuntimeError("Failed to create cache file: {self.cache_path}") | |||||
| try: | |||||
| move(source=self.target_path, destination=new_location) | |||||
| except: # noqa: E722 # TODO | |||||
| raise RuntimeError( | |||||
| f"Failed to move target to store path: {self.target_path} -> {self.new_location}" | |||||
| ) | |||||
| try: | |||||
| symlink(target_path=new_location, symlink_path=self.target_path) | |||||
| except: # noqa: E722 # TODO | |||||
| raise RuntimeError( | |||||
| f"Failed to symlink store path to target: {new_location} -> {self.target_path}" | |||||
| ) | |||||
| @ -0,0 +1,102 @@ | |||||
| from pathlib import Path | |||||
| from typing import Dict | |||||
| import json | |||||
| 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)} | |||||
| 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: | |||||
| """ | |||||
| Move a file using pathlib | |||||
| Args: | |||||
| source: Path to original source of the file/directory | |||||
| destination: Path to new destination | |||||
| Returns: | |||||
| None | |||||
| """ | |||||
| source.rename(destination) | |||||
| def remove(path: Path) -> None: | |||||
| """ | |||||
| Remove a file or symlink | |||||
| Does not support directories as a precaution and lack of need | |||||
| Args: | |||||
| path: Path to the file or symlink | |||||
| Returns: | |||||
| None | |||||
| """ | |||||
| if path.is_symlink() or path.is_file(): | |||||
| path.unlink() | |||||
| def symlink(target_path: Path, symlink_path: Path) -> None: | |||||
| """ | |||||
| Symlinks a file or directory | |||||
| Args: | |||||
| target_path: Path to the target that is being symlinked to | |||||
| symlink_path: Path to the symlink | |||||
| Returns: | |||||
| None | |||||
| """ | |||||
| symlink_path.symlink_to(target_path) | |||||