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.
Expand source code
"""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 collections import namedtuple
from fnmatch import fnmatch
from pathlib import Path
import sys
from . import config
# Expose `emanate.version.__version__` as `emanate.__version__`.
from .version import __version__ # noqa: F401
#: Emanate's maintainer.
__author__ = "Ellen Marie Dash"
# __version__ is defined in version.py.
class FilePair(namedtuple('FilePair', ['src', 'dest'])):
"""Pairs of source/destination file paths."""
def print_add(self):
"""Print a message when creating a link."""
print("{!r} -> {!r}".format(str(self.src), str(self.dest)))
def print_del(self):
"""Print a message when deleting a link."""
print("{!r}".format(str(self.dest)))
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):
"""Add a link."""
self.dest.symlink_to(self.src)
return self.src.samefile(self.dest)
class Execution(list):
"""Describe an Emanate execution.
Callable once, useful to provide "dry-run"
functionality or report changes back to the user.
"""
def __init__(self, func, printer, iterable):
"""Prepare an Emanate execution."""
self.func = func
self.printer = printer
super().__init__(iterable)
def run(self):
"""Run a prepared execution."""
for args in self:
if self.func(args):
self.printer(args)
def dry(self):
"""Print a dry-run of an execution."""
for args in self:
self.printer(args)
class Emanate:
"""Provide 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).
"""
def __init__(self, *configs, src=None):
"""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).
"""
self.config = config.merge(
config.defaults(src),
*configs,
)
self.dest = self.config.destination.resolve()
@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):
"""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.config.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.config.source)
dest_path.mkdir(exist_ok=True)
return False
return True
def confirm_replace(self, dest_file):
"""Prompt the user before replacing a file.
The prompt is skipped if the `confirm` configuration option is False.
"""
prompt = "{!r} already exists. Replace it?".format(str(dest_file))
if not self.config.confirm:
return True
result = None
while result not in ["y", "n", "\n"]:
print("{} [Y/n] ".format(prompt), end="", flush=True)
result = sys.stdin.read(1).lower()
return result != "n"
def _add_symlink(self, pair):
_, dest = pair
# If the file exists and _isn't_ the symbolic link we're
# trying to make, prompt the user to determine what to do.
if dest.exists():
# If the user said no, skip the file.
if not self.confirm_replace(dest):
return False
Emanate.backup(dest)
return pair.add_symlink()
@staticmethod
def backup(dest_file):
"""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):
all_files = Path(self.config.source).glob("**/*")
for file in filter(self.valid_file, all_files):
src = file.absolute()
dest = self.dest / file.relative_to(self.config.source)
yield FilePair(src, dest)
def create(self):
"""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):
"""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 …
emanate.version
Classes
class Emanate (*configs, src=None)
-
Provide 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).
Expand source code
class Emanate: """Provide 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). """ def __init__(self, *configs, src=None): """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). """ self.config = config.merge( config.defaults(src), *configs, ) self.dest = self.config.destination.resolve() @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): """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.config.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.config.source) dest_path.mkdir(exist_ok=True) return False return True def confirm_replace(self, dest_file): """Prompt the user before replacing a file. The prompt is skipped if the `confirm` configuration option is False. """ prompt = "{!r} already exists. Replace it?".format(str(dest_file)) if not self.config.confirm: return True result = None while result not in ["y", "n", "\n"]: print("{} [Y/n] ".format(prompt), end="", flush=True) result = sys.stdin.read(1).lower() return result != "n" def _add_symlink(self, pair): _, dest = pair # If the file exists and _isn't_ the symbolic link we're # trying to make, prompt the user to determine what to do. if dest.exists(): # If the user said no, skip the file. if not self.confirm_replace(dest): return False Emanate.backup(dest) return pair.add_symlink() @staticmethod def backup(dest_file): """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): all_files = Path(self.config.source).glob("**/*") for file in filter(self.valid_file, all_files): src = file.absolute() dest = self.dest / file.relative_to(self.config.source) yield FilePair(src, dest) def create(self): """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): """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)
Static methods
def backup(dest_file)
-
Rename the file so we can safely write to the original path.
Expand source code
@staticmethod def backup(dest_file): """Rename the file so we can safely write to the original path.""" new_name = str(dest_file) + ".emanate" dest_file.rename(new_name)
Methods
def clean(self)
-
Remove symbolic links.
Expand source code
def clean(self): """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)
-
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): """Prompt the user before replacing a file. The prompt is skipped if the `confirm` configuration option is False. """ prompt = "{!r} already exists. Replace it?".format(str(dest_file)) if not self.config.confirm: return True result = None while result not in ["y", "n", "\n"]: print("{} [Y/n] ".format(prompt), end="", flush=True) result = sys.stdin.read(1).lower() return result != "n"
def create(self)
-
Create symbolic links.
Expand source code
def create(self): """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)
-
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): """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.config.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.config.source) dest_path.mkdir(exist_ok=True) return False return True
class Execution (func, printer, iterable)
-
Describe an Emanate execution.
Callable once, useful to provide "dry-run" functionality or report changes back to the user.
Prepare an Emanate execution.
Expand source code
class Execution(list): """Describe an Emanate execution. Callable once, useful to provide "dry-run" functionality or report changes back to the user. """ def __init__(self, func, printer, iterable): """Prepare an Emanate execution.""" self.func = func self.printer = printer super().__init__(iterable) def run(self): """Run a prepared execution.""" for args in self: if self.func(args): self.printer(args) def dry(self): """Print a dry-run of an execution.""" for args in self: self.printer(args)
Ancestors
- builtins.list
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: self.printer(args)
def run(self)
-
Run a prepared execution.
Expand source code
def run(self): """Run a prepared execution.""" for args in self: if self.func(args): self.printer(args)
class FilePair (src, dest)
-
Pairs of source/destination file paths.
Expand source code
class FilePair(namedtuple('FilePair', ['src', 'dest'])): """Pairs of source/destination file paths.""" def print_add(self): """Print a message when creating a link.""" print("{!r} -> {!r}".format(str(self.src), str(self.dest))) def print_del(self): """Print a message when deleting a link.""" print("{!r}".format(str(self.dest))) 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): """Add a link.""" self.dest.symlink_to(self.src) return self.src.samefile(self.dest)
Ancestors
- builtins.tuple
Methods
def add_symlink(self)
-
Add a link.
Expand source code
def add_symlink(self): """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("{!r} -> {!r}".format(str(self.src), str(self.dest)))
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("{!r}".format(str(self.dest)))