Package emanate

Emanate symbolic link manager.

Emanate is a command-line utility and Python library for managing symbolic links in a fashion similar to Stow or Effuse.

Given a source and destination directory, Emanate can create or remove symbolic links from the destination to each file in the source, mirroring the directory structure and creating directories as needed.

"""Emanate symbolic link manager.

Emanate is a command-line utility and Python library for managing symbolic
links in a fashion similar to Stow or Effuse.

Given a `source` and `destination` directory, Emanate can create or remove
symbolic links from the destination to each file in the source, mirroring
the directory structure and creating directories as needed.
"""

from dataclasses import dataclass
from fnmatch import fnmatch
from pathlib import Path
from typing import Any, Callable, Iterable
import sys
from .config import Config

try:
    from importlib import metadata
except ImportError:
    import importlib_metadata as metadata  # type: ignore

# Set the module-level dunders suggested in PEP8
__author__ = metadata.metadata(__name__)['author']
__version__ = metadata.version(__name__)


@dataclass(frozen=True)
class FilePair:
    """Pairs of source/destination file paths."""

    src: Path
    dest: Path

    def print_add(self):
        """Print a message when creating a link."""
        print(f"{str(self.src)!r} -> {str(self.dest)!r}")

    def print_del(self):
        """Print a message when deleting a link."""
        print(f"{str(self.dest)!r}")

    def del_symlink(self):
        """Delete a link."""
        if self.dest.samefile(self.src):
            self.dest.unlink()

        return not self.dest.exists()

    def add_symlink(self) -> bool:
        """Add a link."""
        self.dest.symlink_to(self.src)
        return self.src.samefile(self.dest)


@dataclass(frozen=True)
class Execution:
    """Describe an Emanate execution.

    The user passes functions defining the operation that is applied, and a
    “printer” that's called upon changes; this is useful to provide "dry-run"
    functionality or report changes back to the user.
    """

    func: 'Callable[[FilePair], bool]'
    printer: 'Callable[[FilePair], Any]'
    ops: 'Iterable[FilePair]'

    def run(self):
        """Run a prepared execution.

        Callable only once per Execution object.
        """
        for args in self.ops:
            if self.func(args):
                self.printer(args)

    def dry(self):
        """Print a dry-run of an execution."""
        for args in self.ops:
            self.printer(args)


class Emanate:
    """Provides the core functionality of Emanate.

    This class is configurable at initialization-time, by passing it a number
    of configuration objects, supporting programmatic use (from a configuration
    management tool, for instance) as well as wrapping it in a human interface
    (see emanate.main for a simple example).
    """

    config: Config

    def __init__(self, *configs: Config):
        """Construct an Emanate instance from configuration dictionaries.

        The default values (as provided by Config.defaults()) are implicitly
        the first configuration object; latter configurations override earlier
        configurations (see Config.merge).

        The configs must define a source directory.
        """
        explicit_configs = Config.merge(*configs)
        self.conf = Config.defaults(explicit_configs.get('source')).merge(
            explicit_configs,
        )

    @property
    def dest(self) -> Path:
        return self.conf.destination

    @staticmethod
    def _is_dir(path_obj):
        """Check whether a given path is a directory, but never raise an
        exception (such as Path(x).is_dir() may do).
        """
        try:
            return path_obj.is_dir()
        except OSError:
            return False

    def valid_file(self, path_obj: Path) -> bool:
        """Check whether a given path is covered by an ignore glob.

        As a side effect, if the path is a directory, it is created
        in the destination directory.
        """
        path = str(path_obj.absolute())
        ignore_patterns = []
        for pattern in self.conf.ignore:
            ignore_patterns.append(pattern)
            # If it's a directory, also ignore its contents.
            if Emanate._is_dir(pattern):
                ignore_patterns.append(pattern / "*")

        if any(fnmatch(path, str(pattern)) for pattern in ignore_patterns):
            return False

        if path_obj.is_dir():
            dest_path = self.dest / path_obj.relative_to(self.conf.source)
            dest_path.mkdir(exist_ok=True)
            return False

        return True

    def confirm_replace(self, dest_file: Path) -> bool:
        """Prompt the user before replacing a file.

        The prompt is skipped if the `confirm` configuration option is False.
        """
        prompt = f"{str(dest_file)!r} already exists. Replace it?"

        if not self.conf.confirm:
            return True

        result = None
        while result not in ["y", "n", "\n"]:
            print(f"{prompt} [Y/n] ", end="", flush=True)
            result = sys.stdin.read(1).lower()

        return result != "n"

    def _add_symlink(self, pair: FilePair) -> bool:
        # If the file exists and _isn't_ the symbolic link we're
        # trying to make, prompt the user to determine what to do.
        if pair.dest.exists():
            # If the user said no, skip the file.
            if not self.confirm_replace(pair.dest):
                return False
            Emanate.backup(pair.dest)

        return pair.add_symlink()

    @staticmethod
    def backup(dest_file: Path):
        """Rename the file so we can safely write to the original path."""
        new_name = str(dest_file) + ".emanate"
        dest_file.rename(new_name)

    def _files(self) -> Iterable[FilePair]:
        all_files = Path(self.conf.source).glob("**/*")
        for file in filter(self.valid_file, all_files):
            src  = file.absolute()
            dest = self.dest / file.relative_to(self.conf.source)
            yield FilePair(src, dest)

    def create(self) -> Execution:
        """Create symbolic links."""
        # Ignore files that are already linked.
        gen = filter(lambda p: not (p.dest.exists() and p.src.samefile(p.dest)),
                     self._files())

        return Execution(self._add_symlink,
                         FilePair.print_add,
                         gen)

    def clean(self) -> Execution:
        """Remove symbolic links."""
        # Skip non-existing files.
        gen = filter(lambda p: p.dest.exists(), self._files())

        return Execution(FilePair.del_symlink,
                         FilePair.print_del,
                         gen)

Sub-modules

emanate.cli

Command-line interface for the Emanate symbolic link manager …

emanate.config

Emanate's configuration module …

Classes

class Emanate (*configs: Config)

Provides the core functionality of Emanate.

This class is configurable at initialization-time, by passing it a number of configuration objects, supporting programmatic use (from a configuration management tool, for instance) as well as wrapping it in a human interface (see emanate.main for a simple example).

Construct an Emanate instance from configuration dictionaries.

The default values (as provided by Config.defaults()) are implicitly the first configuration object; latter configurations override earlier configurations (see Config.merge).

The configs must define a source directory.

Expand source code

class Emanate:
    """Provides the core functionality of Emanate.

    This class is configurable at initialization-time, by passing it a number
    of configuration objects, supporting programmatic use (from a configuration
    management tool, for instance) as well as wrapping it in a human interface
    (see emanate.main for a simple example).
    """

    config: Config

    def __init__(self, *configs: Config):
        """Construct an Emanate instance from configuration dictionaries.

        The default values (as provided by Config.defaults()) are implicitly
        the first configuration object; latter configurations override earlier
        configurations (see Config.merge).

        The configs must define a source directory.
        """
        explicit_configs = Config.merge(*configs)
        self.conf = Config.defaults(explicit_configs.get('source')).merge(
            explicit_configs,
        )

    @property
    def dest(self) -> Path:
        return self.conf.destination

    @staticmethod
    def _is_dir(path_obj):
        """Check whether a given path is a directory, but never raise an
        exception (such as Path(x).is_dir() may do).
        """
        try:
            return path_obj.is_dir()
        except OSError:
            return False

    def valid_file(self, path_obj: Path) -> bool:
        """Check whether a given path is covered by an ignore glob.

        As a side effect, if the path is a directory, it is created
        in the destination directory.
        """
        path = str(path_obj.absolute())
        ignore_patterns = []
        for pattern in self.conf.ignore:
            ignore_patterns.append(pattern)
            # If it's a directory, also ignore its contents.
            if Emanate._is_dir(pattern):
                ignore_patterns.append(pattern / "*")

        if any(fnmatch(path, str(pattern)) for pattern in ignore_patterns):
            return False

        if path_obj.is_dir():
            dest_path = self.dest / path_obj.relative_to(self.conf.source)
            dest_path.mkdir(exist_ok=True)
            return False

        return True

    def confirm_replace(self, dest_file: Path) -> bool:
        """Prompt the user before replacing a file.

        The prompt is skipped if the `confirm` configuration option is False.
        """
        prompt = f"{str(dest_file)!r} already exists. Replace it?"

        if not self.conf.confirm:
            return True

        result = None
        while result not in ["y", "n", "\n"]:
            print(f"{prompt} [Y/n] ", end="", flush=True)
            result = sys.stdin.read(1).lower()

        return result != "n"

    def _add_symlink(self, pair: FilePair) -> bool:
        # If the file exists and _isn't_ the symbolic link we're
        # trying to make, prompt the user to determine what to do.
        if pair.dest.exists():
            # If the user said no, skip the file.
            if not self.confirm_replace(pair.dest):
                return False
            Emanate.backup(pair.dest)

        return pair.add_symlink()

    @staticmethod
    def backup(dest_file: Path):
        """Rename the file so we can safely write to the original path."""
        new_name = str(dest_file) + ".emanate"
        dest_file.rename(new_name)

    def _files(self) -> Iterable[FilePair]:
        all_files = Path(self.conf.source).glob("**/*")
        for file in filter(self.valid_file, all_files):
            src  = file.absolute()
            dest = self.dest / file.relative_to(self.conf.source)
            yield FilePair(src, dest)

    def create(self) -> Execution:
        """Create symbolic links."""
        # Ignore files that are already linked.
        gen = filter(lambda p: not (p.dest.exists() and p.src.samefile(p.dest)),
                     self._files())

        return Execution(self._add_symlink,
                         FilePair.print_add,
                         gen)

    def clean(self) -> Execution:
        """Remove symbolic links."""
        # Skip non-existing files.
        gen = filter(lambda p: p.dest.exists(), self._files())

        return Execution(FilePair.del_symlink,
                         FilePair.print_del,
                         gen)

Class variables

var config : Config

Static methods

def backup(dest_file: pathlib.Path)

Rename the file so we can safely write to the original path.

Expand source code

@staticmethod
def backup(dest_file: Path):
    """Rename the file so we can safely write to the original path."""
    new_name = str(dest_file) + ".emanate"
    dest_file.rename(new_name)

Instance variables

var dest : pathlib.Path

Expand source code

@property
def dest(self) -> Path:
    return self.conf.destination

Methods

def clean(self) ‑> Execution

Remove symbolic links.

Expand source code

def clean(self) -> Execution:
    """Remove symbolic links."""
    # Skip non-existing files.
    gen = filter(lambda p: p.dest.exists(), self._files())

    return Execution(FilePair.del_symlink,
                     FilePair.print_del,
                     gen)

def confirm_replace(self, dest_file: pathlib.Path) ‑> bool

Prompt the user before replacing a file.

The prompt is skipped if the confirm configuration option is False.

Expand source code

def confirm_replace(self, dest_file: Path) -> bool:
    """Prompt the user before replacing a file.

    The prompt is skipped if the `confirm` configuration option is False.
    """
    prompt = f"{str(dest_file)!r} already exists. Replace it?"

    if not self.conf.confirm:
        return True

    result = None
    while result not in ["y", "n", "\n"]:
        print(f"{prompt} [Y/n] ", end="", flush=True)
        result = sys.stdin.read(1).lower()

    return result != "n"

def create(self) ‑> Execution

Create symbolic links.

Expand source code

def create(self) -> Execution:
    """Create symbolic links."""
    # Ignore files that are already linked.
    gen = filter(lambda p: not (p.dest.exists() and p.src.samefile(p.dest)),
                 self._files())

    return Execution(self._add_symlink,
                     FilePair.print_add,
                     gen)

def valid_file(self, path_obj: pathlib.Path) ‑> bool

Check whether a given path is covered by an ignore glob.

As a side effect, if the path is a directory, it is created in the destination directory.

Expand source code

def valid_file(self, path_obj: Path) -> bool:
    """Check whether a given path is covered by an ignore glob.

    As a side effect, if the path is a directory, it is created
    in the destination directory.
    """
    path = str(path_obj.absolute())
    ignore_patterns = []
    for pattern in self.conf.ignore:
        ignore_patterns.append(pattern)
        # If it's a directory, also ignore its contents.
        if Emanate._is_dir(pattern):
            ignore_patterns.append(pattern / "*")

    if any(fnmatch(path, str(pattern)) for pattern in ignore_patterns):
        return False

    if path_obj.is_dir():
        dest_path = self.dest / path_obj.relative_to(self.conf.source)
        dest_path.mkdir(exist_ok=True)
        return False

    return True

class Execution (func: Callable[[FilePair], bool], printer: Callable[[FilePair], Any], ops: Iterable[FilePair])

Describe an Emanate execution.

The user passes functions defining the operation that is applied, and a “printer” that's called upon changes; this is useful to provide "dry-run" functionality or report changes back to the user.

Expand source code

class Execution:
    """Describe an Emanate execution.

    The user passes functions defining the operation that is applied, and a
    “printer” that's called upon changes; this is useful to provide "dry-run"
    functionality or report changes back to the user.
    """

    func: 'Callable[[FilePair], bool]'
    printer: 'Callable[[FilePair], Any]'
    ops: 'Iterable[FilePair]'

    def run(self):
        """Run a prepared execution.

        Callable only once per Execution object.
        """
        for args in self.ops:
            if self.func(args):
                self.printer(args)

    def dry(self):
        """Print a dry-run of an execution."""
        for args in self.ops:
            self.printer(args)

Class variables

var func : Callable[[FilePair], bool]

var ops : Iterable[FilePair]

var printer : Callable[[FilePair], Any]

Methods

def dry(self)

Print a dry-run of an execution.

Expand source code

def dry(self):
    """Print a dry-run of an execution."""
    for args in self.ops:
        self.printer(args)

def run(self)

Run a prepared execution.

Callable only once per Execution object.

Expand source code

def run(self):
    """Run a prepared execution.

    Callable only once per Execution object.
    """
    for args in self.ops:
        if self.func(args):
            self.printer(args)

class FilePair (src: pathlib.Path, dest: pathlib.Path)

Pairs of source/destination file paths.

Expand source code

class FilePair:
    """Pairs of source/destination file paths."""

    src: Path
    dest: Path

    def print_add(self):
        """Print a message when creating a link."""
        print(f"{str(self.src)!r} -> {str(self.dest)!r}")

    def print_del(self):
        """Print a message when deleting a link."""
        print(f"{str(self.dest)!r}")

    def del_symlink(self):
        """Delete a link."""
        if self.dest.samefile(self.src):
            self.dest.unlink()

        return not self.dest.exists()

    def add_symlink(self) -> bool:
        """Add a link."""
        self.dest.symlink_to(self.src)
        return self.src.samefile(self.dest)

Class variables

var dest : pathlib.Path

var src : pathlib.Path

Methods

def add_symlink(self) ‑> bool

Add a link.

Expand source code

def add_symlink(self) -> bool:
    """Add a link."""
    self.dest.symlink_to(self.src)
    return self.src.samefile(self.dest)

def del_symlink(self)

Delete a link.

Expand source code

def del_symlink(self):
    """Delete a link."""
    if self.dest.samefile(self.src):
        self.dest.unlink()

    return not self.dest.exists()

def print_add(self)

Print a message when creating a link.

Expand source code

def print_add(self):
    """Print a message when creating a link."""
    print(f"{str(self.src)!r} -> {str(self.dest)!r}")

def print_del(self)

Print a message when deleting a link.

Expand source code

def print_del(self):
    """Print a message when deleting a link."""
    print(f"{str(self.dest)!r}")