mirrors/dotdrop
1
0
Fork 0
mirror of https://github.com/deadc0de6/dotdrop.git synced 2026-02-15 17:20:05 +00:00
Code Activity

linting

Browse Source
This commit is contained in:
deadc0de6
2021-04-30 20:01:02 +02:00
parent 5db5d51fb2
commit cb71bf299f
7 changed files with 532 additions and 480 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
dotdrop/__init__.py
View File

@@ -3,6 +3,7 @@ author: deadc0de6 (https://github.com/deadc0de6)
Copyright (c) 2017, deadc0de6
"""
# pylint: disable=C0415
import sys

126
dotdrop/cfg_yaml.py
View File

@@ -36,6 +36,7 @@ from dotdrop.exceptions import YamlException, UndefinedException
class CfgYaml:
"""yaml config file parser"""
# global entries
key_settings = Settings.key_yaml
@@ -437,7 +438,6 @@ class CfgYaml:
self._log.warn(warn)
self._dirty = False
self.cfg_updated = False
return True
def dump(self):
@@ -464,20 +464,20 @@ class CfgYaml:
self._check_minversion(minversion)
# normalize paths
p = self._norm_path(settings[self.key_settings_dotpath])
settings[self.key_settings_dotpath] = p
p = self._norm_path(settings[self.key_settings_workdir])
settings[self.key_settings_workdir] = p
p = [
self._norm_path(p)
for p in settings[Settings.key_filter_file]
paths = self._norm_path(settings[self.key_settings_dotpath])
settings[self.key_settings_dotpath] = paths
paths = self._norm_path(settings[self.key_settings_workdir])
settings[self.key_settings_workdir] = paths
paths = [
self._norm_path(path)
for path in settings[Settings.key_filter_file]
]
settings[Settings.key_filter_file] = p
p = [
self._norm_path(p)
for p in settings[Settings.key_func_file]
settings[Settings.key_filter_file] = paths
paths = [
self._norm_path(path)
for path in settings[Settings.key_func_file]
]
settings[Settings.key_func_file] = p
settings[Settings.key_func_file] = paths
if self._debug:
self._debug_dict('settings block:', settings)
return settings
@@ -1166,11 +1166,11 @@ class CfgYaml:
new = []
if not entries:
return new
for e in entries:
et = self._template_item(e)
if self._debug and e != et:
self._dbg('resolved: {} -> {}'.format(e, et))
new.append(et)
for entry in entries:
newe = self._template_item(entry)
if self._debug and entry != newe:
self._dbg('resolved: {} -> {}'.format(entry, newe))
new.append(newe)
return new
def _template_dict(self, entries):
@@ -1178,11 +1178,11 @@ class CfgYaml:
new = {}
if not entries:
return new
for k, v in entries.items():
vt = self._template_item(v)
if self._debug and v != vt:
self._dbg('resolved: {} -> {}'.format(v, vt))
new[k] = vt
for k, val in entries.items():
newv = self._template_item(val)
if self._debug and val != newv:
self._dbg('resolved: {} -> {}'.format(val, newv))
new[k] = newv
return new
def _template_dotfiles_entries(self):
@@ -1226,9 +1226,9 @@ class CfgYaml:
if self.key_all not in pdfs:
# take a subset of the dotfiles
newdotfiles = {}
for k, v in dotfiles.items():
for k, val in dotfiles.items():
if k in pdfs:
newdotfiles[k] = v
newdotfiles[k] = val
dotfiles = newdotfiles
for dotfile in dotfiles.values():
@@ -1246,26 +1246,29 @@ class CfgYaml:
var = self._enrich_vars(variables, self._profile)
# use a separated templategen to handle variables
# resolved outside the main config
t = Templategen(variables=var,
func_file=self.settings[Settings.key_func_file],
filter_file=self.settings[Settings.key_filter_file])
func_files = self.settings[Settings.key_func_file]
filter_files = self.settings[Settings.key_filter_file]
templ = Templategen(variables=var,
func_file=func_files,
filter_file=filter_files)
for k in variables.keys():
val = variables[k]
while Templategen.var_is_template(val):
val = t.generate_string(val)
val = templ.generate_string(val)
variables[k] = val
t.update_variables(variables)
templ.update_variables(variables)
if variables is self.variables:
self._redefine_templater()
def _get_profile_included_vars(self):
"""resolve profile included variables/dynvariables"""
for k, v in self.profiles.items():
if self.key_profile_include in v and v[self.key_profile_include]:
for _, val in self.profiles.items():
if self.key_profile_include in val and \
val[self.key_profile_include]:
new = []
for x in v[self.key_profile_include]:
new.append(self._tmpl.generate_string(x))
v[self.key_profile_include] = new
for entry in val[self.key_profile_include]:
new.append(self._tmpl.generate_string(entry))
val[self.key_profile_include] = new
# now get the included ones
pro_var = self._get_profile_included_item(self.key_profile_variables)
@@ -1291,7 +1294,8 @@ class CfgYaml:
"""
if not dic:
return
[dic.pop(k, None) for k in self._profilevarskeys]
for k in self._profilevarskeys:
dic.pop(k, None)
def _parse_extended_import_path(self, path_entry):
"""Parse an import path in a tuple (path, fatal_not_found)."""
@@ -1369,7 +1373,8 @@ class CfgYaml:
processed_paths = (self._process_path(p) for p in paths)
return list(chain.from_iterable(processed_paths))
def _merge_dict(self, high, low):
@classmethod
def _merge_dict(cls, high, low):
"""merge high and low dict"""
if not high:
high = {}
@@ -1377,7 +1382,8 @@ class CfgYaml:
low = {}
return {**low, **high}
def _get_entry(self, dic, key, mandatory=True):
@classmethod
def _get_entry(cls, dic, key, mandatory=True):
"""return copy of entry from yaml dictionary"""
if key not in dic:
if mandatory:
@@ -1393,19 +1399,19 @@ class CfgYaml:
def _clear_none(self, dic):
"""recursively delete all none/empty values in a dictionary."""
new = {}
for k, v in dic.items():
for k, val in dic.items():
if k == self.key_dotfile_src:
# allow empty dotfile src
new[k] = v
new[k] = val
continue
if k == self.key_dotfile_dst:
# allow empty dotfile dst
new[k] = v
new[k] = val
continue
newv = v
if isinstance(v, dict):
newv = val
if isinstance(val, dict):
# recursive travers dict
newv = self._clear_none(v)
newv = self._clear_none(val)
if not newv:
# no empty dict
continue
@@ -1418,7 +1424,8 @@ class CfgYaml:
new[k] = newv
return new
def _is_glob(self, path):
@classmethod
def _is_glob(cls, path):
"""Quick test if path is a glob."""
return '*' in path or '?' in path
@@ -1435,8 +1442,8 @@ class CfgYaml:
return path
path = os.path.expanduser(path)
if not os.path.isabs(path):
d = os.path.dirname(self._path)
ret = os.path.join(d, path)
dirn = os.path.dirname(self._path)
ret = os.path.join(dirn, path)
if self._debug:
msg = 'normalizing relative to cfg: {} -> {}'
self._dbg(msg.format(path, ret))
@@ -1446,30 +1453,31 @@ class CfgYaml:
self._dbg('normalizing: {} -> {}'.format(path, ret))
return ret
def _shell_exec_dvars(self, dic, keys=[]):
def _shell_exec_dvars(self, dic, keys=None):
"""shell execute dynvariables in-place"""
if not keys:
keys = dic.keys()
for k in keys:
v = dic[k]
ret, out = shell(v, debug=self._debug)
val = dic[k]
ret, out = shell(val, debug=self._debug)
if not ret:
err = 'var \"{}: {}\" failed: {}'.format(k, v, out)
err = 'var \"{}: {}\" failed: {}'.format(k, val, out)
self._log.err(err)
raise YamlException(err)
if self._debug:
self._dbg('{}: `{}` -> {}'.format(k, v, out))
self._dbg('{}: `{}` -> {}'.format(k, val, out))
dic[k] = out
def _check_minversion(self, minversion):
@classmethod
def _check_minversion(cls, minversion):
if not minversion:
return
try:
cur = tuple([int(x) for x in VERSION.split('.')])
cfg = tuple([int(x) for x in minversion.split('.')])
except Exception:
cur = ([int(x) for x in VERSION.split('.')])
cfg = ([int(x) for x in minversion.split('.')])
except Exception as exc:
err = 'bad version: \"{}\" VS \"{}\"'.format(VERSION, minversion)
raise YamlException(err)
raise YamlException(err) from exc
if cur < cfg:
err = 'current dotdrop version is too old for that config file.'
err += ' Please update.'
@@ -1495,8 +1503,8 @@ class CfgYaml:
self._dbg('{}:'.format(title))
if not elems:
return
for k, v in elems.items():
self._dbg('\t- \"{}\": {}'.format(k, v))
for k, val in elems.items():
self._dbg('\t- \"{}\": {}'.format(k, val))
def _dbg(self, content):
pre = os.path.basename(self._path)

550
dotdrop/dotdrop.py
View File

@@ -22,7 +22,8 @@ from dotdrop.utils import get_tmpdir, removepath, \
uniq_list, patch_ignores, dependencies_met, \
adapt_workers
from dotdrop.linktypes import LinkTypes
from dotdrop.exceptions import YamlException, UndefinedException
from dotdrop.exceptions import YamlException, \
UndefinedException, UnmetDependency
LOG = Logger()
TRANS_SUFFIX = 'trans'
@@ -32,7 +33,7 @@ TRANS_SUFFIX = 'trans'
###########################################################
def action_executor(o, actions, defactions, templater, post=False):
def action_executor(opts, actions, defactions, templater, post=False):
"""closure for action execution"""
def execute():
"""
@@ -40,70 +41,71 @@ def action_executor(o, actions, defactions, templater, post=False):
True, None if ok
False, errstring if issue
"""
s = 'pre' if not post else 'post'
actiontype = 'pre' if not post else 'post'
# execute default actions
for action in defactions:
if o.dry:
LOG.dry('would execute def-{}-action: {}'.format(s,
if opts.dry:
LOG.dry('would execute def-{}-action: {}'.format(actiontype,
action))
continue
LOG.dbg('executing def-{}-action: {}'.format(s, action))
ret = action.execute(templater=templater, debug=o.debug)
LOG.dbg('executing def-{}-action: {}'.format(actiontype, action))
ret = action.execute(templater=templater, debug=opts.debug)
if not ret:
err = 'def-{}-action \"{}\" failed'.format(s, action.key)
LOG.err(err)
err = 'def-{}-action \"{}\" failed'
LOG.err(err.format(actiontype, action.key))
return False, err
# execute actions
for action in actions:
if o.dry:
LOG.dry('would execute {}-action: {}'.format(s, action))
if opts.dry:
err = 'would execute {}-action: {}'
LOG.dry(err.format(actiontype, action))
continue
LOG.dbg('executing {}-action: {}'.format(s, action))
ret = action.execute(templater=templater, debug=o.debug)
LOG.dbg('executing {}-action: {}'.format(actiontype, action))
ret = action.execute(templater=templater, debug=opts.debug)
if not ret:
err = '{}-action \"{}\" failed'.format(s, action.key)
err = '{}-action \"{}\" failed'.format(actiontype, action.key)
LOG.err(err)
return False, err
return True, None
return execute
def _dotfile_update(o, path, key=False):
def _dotfile_update(opts, path, key=False):
"""
update a dotfile pointed by path
if key is false or by key (in path)
"""
updater = Updater(o.dotpath, o.variables, o.conf,
dry=o.dry, safe=o.safe, debug=o.debug,
ignore=o.update_ignore,
showpatch=o.update_showpatch,
ignore_missing_in_dotdrop=o.ignore_missing_in_dotdrop)
updater = Updater(opts.dotpath, opts.variables, opts.conf,
dry=opts.dry, safe=opts.safe, debug=opts.debug,
ignore=opts.update_ignore,
showpatch=opts.update_showpatch,
ignore_missing_in_dotdrop=opts.ignore_missing_in_dotdrop)
if key:
return updater.update_key(path)
return updater.update_path(path)
def _dotfile_compare(o, dotfile, tmp):
def _dotfile_compare(opts, dotfile, tmp):
"""
compare a dotfile
returns True if same
"""
t = _get_templater(o)
ignore_missing_in_dotdrop = o.ignore_missing_in_dotdrop or \
templ = _get_templater(opts)
ignore_missing_in_dotdrop = opts.ignore_missing_in_dotdrop or \
dotfile.ignore_missing_in_dotdrop
inst = Installer(create=o.create, backup=o.backup,
dry=o.dry, base=o.dotpath,
workdir=o.workdir, debug=o.debug,
backup_suffix=o.install_backup_suffix,
diff_cmd=o.diff_command)
comp = Comparator(diff_cmd=o.diff_command, debug=o.debug,
inst = Installer(create=opts.create, backup=opts.backup,
dry=opts.dry, base=opts.dotpath,
workdir=opts.workdir, debug=opts.debug,
backup_suffix=opts.install_backup_suffix,
diff_cmd=opts.diff_command)
comp = Comparator(diff_cmd=opts.diff_command, debug=opts.debug,
ignore_missing_in_dotdrop=ignore_missing_in_dotdrop)
# add dotfile variables
newvars = dotfile.get_dotfile_variables()
t.add_tmp_vars(newvars=newvars)
templ.add_tmp_vars(newvars=newvars)
# dotfiles does not exist / not installed
LOG.dbg('comparing {}'.format(dotfile))
@@ -118,14 +120,14 @@ def _dotfile_compare(o, dotfile, tmp):
tmpsrc = None
if dotfile.trans_r:
LOG.dbg('applying transformation before comparing')
tmpsrc = apply_trans(o.dotpath, dotfile, t, debug=o.debug)
tmpsrc = apply_trans(opts.dotpath, dotfile, templ, debug=opts.debug)
if not tmpsrc:
# could not apply trans
return False
src = tmpsrc
# is a symlink pointing to itself
asrc = os.path.join(o.dotpath, os.path.expanduser(src))
asrc = os.path.join(opts.dotpath, os.path.expanduser(src))
adst = os.path.expanduser(dotfile.dst)
if os.path.samefile(asrc, adst):
line = '=> compare {}: diffing with \"{}\"'
@@ -133,13 +135,13 @@ def _dotfile_compare(o, dotfile, tmp):
LOG.dbg('points to itself')
return True
ignores = list(set(o.compare_ignore + dotfile.cmpignore))
ignores = patch_ignores(ignores, dotfile.dst, debug=o.debug)
ignores = list(set(opts.compare_ignore + dotfile.cmpignore))
ignores = patch_ignores(ignores, dotfile.dst, debug=opts.debug)
insttmp = None
if dotfile.template and Templategen.is_template(src, ignore=ignores):
# install dotfile to temporary dir for compare
ret, err, insttmp = inst.install_to_temp(t, tmp, src, dotfile.dst,
ret, err, insttmp = inst.install_to_temp(templ, tmp, src, dotfile.dst,
is_template=True,
chmod=dotfile.chmod)
if not ret:
@@ -151,28 +153,29 @@ def _dotfile_compare(o, dotfile, tmp):
src = insttmp
# compare
# need to be executed before cleaning
diff = comp.compare(src, dotfile.dst, ignore=ignores)
# clean tmp transformed dotfile if any
if tmpsrc:
tmpsrc = os.path.join(o.dotpath, tmpsrc)
tmpsrc = os.path.join(opts.dotpath, tmpsrc)
if os.path.exists(tmpsrc):
removepath(tmpsrc, LOG)
# clean tmp template dotfile if any
if insttmp:
if os.path.exists(insttmp):
if insttmp and os.path.exists(insttmp):
removepath(insttmp, LOG)
if diff != '':
# print diff results
line = '=> compare {}: diffing with \"{}\"'
LOG.log(line.format(dotfile.key, dotfile.dst))
if o.compare_fileonly:
if opts.compare_fileonly:
LOG.raw('<files are different>')
else:
LOG.emph(diff)
return False
# no difference
line = '=> compare {}: diffing with \"{}\"'
LOG.dbg(line.format(dotfile.key, dotfile.dst))
@@ -180,33 +183,33 @@ def _dotfile_compare(o, dotfile, tmp):
return True
def _dotfile_install(o, dotfile, tmpdir=None):
def _dotfile_install(opts, dotfile, tmpdir=None):
"""
install a dotfile
returns <success, dotfile key, err>
"""
# installer
inst = _get_install_installer(o, tmpdir=tmpdir)
inst = _get_install_installer(opts, tmpdir=tmpdir)
# templater
t = _get_templater(o)
templ = _get_templater(opts)
# add dotfile variables
newvars = dotfile.get_dotfile_variables()
t.add_tmp_vars(newvars=newvars)
templ.add_tmp_vars(newvars=newvars)
preactions = []
if not o.install_temporary:
if not opts.install_temporary:
preactions.extend(dotfile.get_pre_actions())
defactions = o.install_default_actions_pre
pre_actions_exec = action_executor(o, preactions, defactions,
t, post=False)
defactions = opts.install_default_actions_pre
pre_actions_exec = action_executor(opts, preactions, defactions,
templ, post=False)
LOG.dbg('installing dotfile: \"{}\"'.format(dotfile.key))
LOG.dbg(dotfile.prt())
ignores = list(set(o.install_ignore + dotfile.instignore))
ignores = patch_ignores(ignores, dotfile.dst, debug=o.debug)
ignores = list(set(opts.install_ignore + dotfile.instignore))
ignores = patch_ignores(ignores, dotfile.dst, debug=opts.debug)
is_template = dotfile.template and Templategen.is_template(
dotfile.src,
@@ -214,29 +217,29 @@ def _dotfile_install(o, dotfile, tmpdir=None):
)
if hasattr(dotfile, 'link') and dotfile.link == LinkTypes.LINK:
# link
r, err = inst.install(t, dotfile.src, dotfile.dst,
ret, err = inst.install(templ, dotfile.src, dotfile.dst,
dotfile.link,
actionexec=pre_actions_exec,
is_template=is_template,
ignore=ignores,
chmod=dotfile.chmod,
force_chmod=o.install_force_chmod)
force_chmod=opts.install_force_chmod)
elif hasattr(dotfile, 'link') and \
dotfile.link == LinkTypes.LINK_CHILDREN:
# link_children
r, err = inst.install(t, dotfile.src, dotfile.dst,
ret, err = inst.install(templ, dotfile.src, dotfile.dst,
dotfile.link,
actionexec=pre_actions_exec,
is_template=is_template,
chmod=dotfile.chmod,
ignore=ignores,
force_chmod=o.install_force_chmod)
force_chmod=opts.install_force_chmod)
else:
# nolink
src = dotfile.src
tmp = None
if dotfile.trans_r:
tmp = apply_trans(o.dotpath, dotfile, t, debug=o.debug)
tmp = apply_trans(opts.dotpath, dotfile, templ, debug=opts.debug)
if not tmp:
return False, dotfile.key, None
src = tmp
@@ -245,92 +248,92 @@ def _dotfile_install(o, dotfile, tmpdir=None):
src,
ignore=ignores,
)
r, err = inst.install(t, src, dotfile.dst,
ret, err = inst.install(templ, src, dotfile.dst,
LinkTypes.NOLINK,
actionexec=pre_actions_exec,
noempty=dotfile.noempty,
ignore=ignores,
is_template=is_template,
chmod=dotfile.chmod,
force_chmod=o.install_force_chmod)
force_chmod=opts.install_force_chmod)
if tmp:
tmp = os.path.join(o.dotpath, tmp)
tmp = os.path.join(opts.dotpath, tmp)
if os.path.exists(tmp):
removepath(tmp, LOG)
# check result of installation
if r:
if ret:
# dotfile was installed
if not o.install_temporary:
defactions = o.install_default_actions_post
if not opts.install_temporary:
defactions = opts.install_default_actions_post
postactions = dotfile.get_post_actions()
post_actions_exec = action_executor(o, postactions, defactions,
t, post=True)
post_actions_exec = action_executor(opts, postactions, defactions,
templ, post=True)
post_actions_exec()
else:
# dotfile was NOT installed
if o.install_force_action:
if opts.install_force_action:
# pre-actions
LOG.dbg('force pre action execution ...')
pre_actions_exec()
# post-actions
LOG.dbg('force post action execution ...')
defactions = o.install_default_actions_post
defactions = opts.install_default_actions_post
postactions = dotfile.get_post_actions()
post_actions_exec = action_executor(o, postactions, defactions,
t, post=True)
post_actions_exec = action_executor(opts, postactions, defactions,
templ, post=True)
post_actions_exec()
return r, dotfile.key, err
return ret, dotfile.key, err
def cmd_install(o):
def cmd_install(opts):
"""install dotfiles for this profile"""
dotfiles = o.dotfiles
prof = o.conf.get_profile()
dotfiles = opts.dotfiles
prof = opts.conf.get_profile()
adapt_workers(o, LOG)
adapt_workers(opts, LOG)
pro_pre_actions = prof.get_pre_actions() if prof else []
pro_post_actions = prof.get_post_actions() if prof else []
if o.install_keys:
if opts.install_keys:
# filtered dotfiles to install
uniq = uniq_list(o.install_keys)
uniq = uniq_list(opts.install_keys)
dotfiles = [d for d in dotfiles if d.key in uniq]
if not dotfiles:
msg = 'no dotfile to install for this profile (\"{}\")'
LOG.warn(msg.format(o.profile))
LOG.warn(msg.format(opts.profile))
return False
# the installer
tmpdir = None
if o.install_temporary:
if opts.install_temporary:
tmpdir = get_tmpdir()
installed = []
# execute profile pre-action
LOG.dbg('run {} profile pre actions'.format(len(pro_pre_actions)))
t = _get_templater(o)
ret, err = action_executor(o, pro_pre_actions, [], t, post=False)()
templ = _get_templater(opts)
ret, _ = action_executor(opts, pro_pre_actions, [], templ, post=False)()
if not ret:
return False
# install each dotfile
if o.workers > 1:
if opts.workers > 1:
# in parallel
LOG.dbg('run with {} workers'.format(o.workers))
ex = futures.ThreadPoolExecutor(max_workers=o.workers)
LOG.dbg('run with {} workers'.format(opts.workers))
ex = futures.ThreadPoolExecutor(max_workers=opts.workers)
wait_for = []
for dotfile in dotfiles:
j = ex.submit(_dotfile_install, o, dotfile, tmpdir=tmpdir)
j = ex.submit(_dotfile_install, opts, dotfile, tmpdir=tmpdir)
wait_for.append(j)
# check result
for f in futures.as_completed(wait_for):
r, key, err = f.result()
if r:
for fut in futures.as_completed(wait_for):
tmpret, key, err = fut.result()
if tmpret:
installed.append(key)
elif err:
LOG.err('installing \"{}\" failed: {}'.format(key,
@@ -338,42 +341,43 @@ def cmd_install(o):
else:
# sequentially
for dotfile in dotfiles:
r, key, err = _dotfile_install(o, dotfile, tmpdir=tmpdir)
tmpret, key, err = _dotfile_install(opts, dotfile, tmpdir=tmpdir)
# check result
if r:
if tmpret:
installed.append(key)
elif err:
LOG.err('installing \"{}\" failed: {}'.format(key,
err))
# execute profile post-action
if len(installed) > 0 or o.install_force_action:
if len(installed) > 0 or opts.install_force_action:
msg = 'run {} profile post actions'
LOG.dbg(msg.format(len(pro_post_actions)))
ret, err = action_executor(o, pro_post_actions, [], t, post=False)()
ret, _ = action_executor(opts, pro_post_actions,
[], templ, post=False)()
if not ret:
return False
LOG.dbg('install done: installed \"{}\"'.format(','.join(installed)))
if o.install_temporary:
if opts.install_temporary:
LOG.log('\ninstalled to tmp \"{}\".'.format(tmpdir))
LOG.log('\n{} dotfile(s) installed.'.format(len(installed)))
return True
def cmd_compare(o, tmp):
def cmd_compare(opts, tmp):
"""compare dotfiles and return True if all identical"""
dotfiles = o.dotfiles
dotfiles = opts.dotfiles
if not dotfiles:
msg = 'no dotfile defined for this profile (\"{}\")'
LOG.warn(msg.format(o.profile))
LOG.warn(msg.format(opts.profile))
return True
# compare only specific files
selected = dotfiles
if o.compare_focus:
selected = _select(o.compare_focus, dotfiles)
if opts.compare_focus:
selected = _select(opts.compare_focus, dotfiles)
if len(selected) < 1:
LOG.log('\nno dotfile to compare')
@@ -381,20 +385,20 @@ def cmd_compare(o, tmp):
same = True
cnt = 0
if o.workers > 1:
if opts.workers > 1:
# in parallel
LOG.dbg('run with {} workers'.format(o.workers))
ex = futures.ThreadPoolExecutor(max_workers=o.workers)
LOG.dbg('run with {} workers'.format(opts.workers))
ex = futures.ThreadPoolExecutor(max_workers=opts.workers)
wait_for = []
for dotfile in selected:
j = ex.submit(_dotfile_compare, o, dotfile, tmp)
wait_for.append(j)
# check result
for f in futures.as_completed(wait_for):
if not dotfile.src and not dotfile.dst:
# ignore fake dotfile
continue
if not f.result():
j = ex.submit(_dotfile_compare, opts, dotfile, tmp)
wait_for.append(j)
# check result
for fut in futures.as_completed(wait_for):
if not fut.result():
same = False
cnt += 1
else:
@@ -403,7 +407,7 @@ def cmd_compare(o, tmp):
if not dotfile.src and not dotfile.dst:
# ignore fake dotfile
continue
if not _dotfile_compare(o, dotfile, tmp):
if not _dotfile_compare(opts, dotfile, tmp):
same = False
cnt += 1
@@ -411,31 +415,32 @@ def cmd_compare(o, tmp):
return same
def cmd_update(o):
def cmd_update(opts):
"""update the dotfile(s) from path(s) or key(s)"""
cnt = 0
paths = o.update_path
iskey = o.update_iskey
paths = opts.update_path
iskey = opts.update_iskey
if o.profile not in [p.key for p in o.profiles]:
LOG.err('no such profile \"{}\"'.format(o.profile))
if opts.profile not in [p.key for p in opts.profiles]:
LOG.err('no such profile \"{}\"'.format(opts.profile))
return False
adapt_workers(o, LOG)
adapt_workers(opts, LOG)
if not paths:
# update the entire profile
if iskey:
LOG.dbg('update by keys: {}'.format(paths))
paths = [d.key for d in o.dotfiles]
paths = [d.key for d in opts.dotfiles]
else:
LOG.dbg('update by paths: {}'.format(paths))
paths = [d.dst for d in o.dotfiles]
msg = 'Update all dotfiles for profile \"{}\"'.format(o.profile)
if o.safe and not LOG.ask(msg):
paths = [d.dst for d in opts.dotfiles]
msg = 'Update all dotfiles for profile \"{}\"'.format(opts.profile)
if opts.safe and not LOG.ask(msg):
LOG.log('\n{} file(s) updated.'.format(cnt))
return False
# check there's something to do
if not paths:
LOG.log('\nno dotfile to update')
return True
@@ -443,84 +448,87 @@ def cmd_update(o):
LOG.dbg('dotfile to update: {}'.format(paths))
# update each dotfile
if o.workers > 1:
if opts.workers > 1:
# in parallel
LOG.dbg('run with {} workers'.format(o.workers))
ex = futures.ThreadPoolExecutor(max_workers=o.workers)
LOG.dbg('run with {} workers'.format(opts.workers))
ex = futures.ThreadPoolExecutor(max_workers=opts.workers)
wait_for = []
for path in paths:
j = ex.submit(_dotfile_update, o, path, key=iskey)
j = ex.submit(_dotfile_update, opts, path, key=iskey)
wait_for.append(j)
# check result
for f in futures.as_completed(wait_for):
if f.result():
for fut in futures.as_completed(wait_for):
if fut.result():
cnt += 1
else:
# sequentially
for path in paths:
if _dotfile_update(o, path, key=iskey):
if _dotfile_update(opts, path, key=iskey):
cnt += 1
LOG.log('\n{} file(s) updated.'.format(cnt))
return cnt == len(paths)
def cmd_importer(o):
def cmd_importer(opts):
"""import dotfile(s) from paths"""
ret = True
cnt = 0
paths = o.import_path
importer = Importer(o.profile, o.conf, o.dotpath, o.diff_command,
dry=o.dry, safe=o.safe, debug=o.debug,
keepdot=o.keepdot, ignore=o.import_ignore)
paths = opts.import_path
importer = Importer(opts.profile, opts.conf,
opts.dotpath, opts.diff_command,
dry=opts.dry, safe=opts.safe,
debug=opts.debug,
keepdot=opts.keepdot,
ignore=opts.import_ignore)
for path in paths:
r = importer.import_path(path, import_as=o.import_as,
import_link=o.import_link,
import_mode=o.import_mode)
if r < 0:
tmpret = importer.import_path(path, import_as=opts.import_as,
import_link=opts.import_link,
import_mode=opts.import_mode)
if tmpret < 0:
ret = False
elif r > 0:
elif tmpret > 0:
cnt += 1
if o.dry:
if opts.dry:
LOG.dry('new config file would be:')
LOG.raw(o.conf.dump())
LOG.raw(opts.conf.dump())
else:
o.conf.save()
opts.conf.save()
LOG.log('\n{} file(s) imported.'.format(cnt))
return ret
def cmd_list_profiles(o):
def cmd_list_profiles(opts):
"""list all profiles"""
LOG.emph('Available profile(s):\n')
for p in o.profiles:
if o.profiles_grepable:
fmt = '{}'.format(p.key)
for profile in opts.profiles:
if opts.profiles_grepable:
fmt = '{}'.format(profile.key)
LOG.raw(fmt)
else:
LOG.sub(p.key, end='')
LOG.log(' ({} dotfiles)'.format(len(p.dotfiles)))
LOG.sub(profile.key, end='')
LOG.log(' ({} dotfiles)'.format(len(profile.dotfiles)))
LOG.log('')
def cmd_files(o):
def cmd_files(opts):
"""list all dotfiles for a specific profile"""
if o.profile not in [p.key for p in o.profiles]:
LOG.warn('unknown profile \"{}\"'.format(o.profile))
if opts.profile not in [p.key for p in opts.profiles]:
LOG.warn('unknown profile \"{}\"'.format(opts.profile))
return
what = 'Dotfile(s)'
if o.files_templateonly:
if opts.files_templateonly:
what = 'Template(s)'
LOG.emph('{} for profile \"{}\":\n'.format(what, o.profile))
for dotfile in o.dotfiles:
if o.files_templateonly:
src = os.path.join(o.dotpath, dotfile.src)
LOG.emph('{} for profile \"{}\":\n'.format(what, opts.profile))
for dotfile in opts.dotfiles:
if opts.files_templateonly:
src = os.path.join(opts.dotpath, dotfile.src)
if not Templategen.is_template(src):
continue
if o.files_grepable:
if opts.files_grepable:
fmt = '{},dst:{},src:{},link:{}'
fmt = fmt.format(dotfile.key, dotfile.dst,
dotfile.src, dotfile.link.name.lower())
@@ -539,26 +547,26 @@ def cmd_files(o):
LOG.log('')
def cmd_detail(o):
def cmd_detail(opts):
"""list details on all files for all dotfile entries"""
if o.profile not in [p.key for p in o.profiles]:
LOG.warn('unknown profile \"{}\"'.format(o.profile))
if opts.profile not in [p.key for p in opts.profiles]:
LOG.warn('unknown profile \"{}\"'.format(opts.profile))
return
dotfiles = o.dotfiles
if o.detail_keys:
dotfiles = opts.dotfiles
if opts.detail_keys:
# filtered dotfiles to install
uniq = uniq_list(o.details_keys)
uniq = uniq_list(opts.details_keys)
dotfiles = [d for d in dotfiles if d.key in uniq]
LOG.emph('dotfiles details for profile \"{}\":\n'.format(o.profile))
for d in dotfiles:
_detail(o.dotpath, d)
LOG.emph('dotfiles details for profile \"{}\":\n'.format(opts.profile))
for dotfile in dotfiles:
_detail(opts.dotpath, dotfile)
LOG.log('')
def cmd_remove(o):
def cmd_remove(opts):
"""remove dotfile from dotpath and from config"""
paths = o.remove_path
iskey = o.remove_iskey
paths = opts.remove_path
iskey = opts.remove_iskey
if not paths:
LOG.log('no dotfile to remove')
@@ -569,13 +577,13 @@ def cmd_remove(o):
for key in paths:
if not iskey:
# by path
dotfiles = o.conf.get_dotfile_by_dst(key)
dotfiles = opts.conf.get_dotfile_by_dst(key)
if not dotfiles:
LOG.warn('{} ignored, does not exist'.format(key))
continue
else:
# by key
dotfile = o.conf.get_dotfile(key)
dotfile = opts.conf.get_dotfile(key)
if not dotfile:
LOG.warn('{} ignored, does not exist'.format(key))
continue
@@ -592,46 +600,46 @@ def cmd_remove(o):
LOG.dbg('removing {}'.format(key))
# make sure is part of the profile
if dotfile.key not in [d.key for d in o.dotfiles]:
if dotfile.key not in [d.key for d in opts.dotfiles]:
msg = '{} ignored, not associated to this profile'
LOG.warn(msg.format(key))
continue
profiles = o.conf.get_profiles_by_dotfile_key(k)
profiles = opts.conf.get_profiles_by_dotfile_key(k)
pkeys = ','.join([p.key for p in profiles])
if o.dry:
if opts.dry:
LOG.dry('would remove {} from {}'.format(dotfile, pkeys))
continue
msg = 'Remove \"{}\" from all these profiles: {}'.format(k, pkeys)
if o.safe and not LOG.ask(msg):
if opts.safe and not LOG.ask(msg):
return False
LOG.dbg('remove dotfile: {}'.format(dotfile))
for profile in profiles:
if not o.conf.del_dotfile_from_profile(dotfile, profile):
if not opts.conf.del_dotfile_from_profile(dotfile, profile):
return False
if not o.conf.del_dotfile(dotfile):
if not opts.conf.del_dotfile(dotfile):
return False
# remove dotfile from dotpath
dtpath = os.path.join(o.dotpath, dotfile.src)
dtpath = os.path.join(opts.dotpath, dotfile.src)
removepath(dtpath, LOG)
# remove empty directory
parent = os.path.dirname(dtpath)
# remove any empty parent up to dotpath
while parent != o.dotpath:
while parent != opts.dotpath:
if os.path.isdir(parent) and not os.listdir(parent):
msg = 'Remove empty dir \"{}\"'.format(parent)
if o.safe and not LOG.ask(msg):
if opts.safe and not LOG.ask(msg):
break
removepath(parent, LOG)
parent = os.path.dirname(parent)
removed.append(dotfile)
if o.dry:
if opts.dry:
LOG.dry('new config file would be:')
LOG.raw(o.conf.dump())
LOG.raw(opts.conf.dump())
else:
o.conf.save()
opts.conf.save()
if removed:
LOG.log('\nFollowing dotfile(s) are not tracked anymore:')
entries = ['- \"{}\" (was tracked as \"{}\")'.format(r.dst, r.key)
@@ -647,25 +655,25 @@ def cmd_remove(o):
###########################################################
def _get_install_installer(o, tmpdir=None):
def _get_install_installer(opts, tmpdir=None):
"""get an installer instance for cmd_install"""
inst = Installer(create=o.create, backup=o.backup,
dry=o.dry, safe=o.safe,
base=o.dotpath, workdir=o.workdir,
diff=o.install_diff, debug=o.debug,
inst = Installer(create=opts.create, backup=opts.backup,
dry=opts.dry, safe=opts.safe,
base=opts.dotpath, workdir=opts.workdir,
diff=opts.install_diff, debug=opts.debug,
totemp=tmpdir,
showdiff=o.install_showdiff,
backup_suffix=o.install_backup_suffix,
diff_cmd=o.diff_command)
showdiff=opts.install_showdiff,
backup_suffix=opts.install_backup_suffix,
diff_cmd=opts.diff_command)
return inst
def _get_templater(o):
def _get_templater(opts):
"""get an templater instance"""
t = Templategen(base=o.dotpath, variables=o.variables,
func_file=o.func_file, filter_file=o.filter_file,
debug=o.debug)
return t
templ = Templategen(base=opts.dotpath, variables=opts.variables,
func_file=opts.func_file, filter_file=opts.filter_file,
debug=opts.debug)
return templ
def _detail(dotpath, dotfile):
@@ -684,24 +692,24 @@ def _detail(dotpath, dotfile):
LOG.sub('{} (template:{})'.format(path, template))
else:
for root, _, files in os.walk(path):
for f in files:
p = os.path.join(root, f)
for file in files:
fpath = os.path.join(root, file)
template = 'no'
if dotfile.template and Templategen.is_template(p):
if dotfile.template and Templategen.is_template(fpath):
template = 'yes'
LOG.sub('{} (template:{})'.format(p, template))
LOG.sub('{} (template:{})'.format(fpath, template))
def _select(selections, dotfiles):
selected = []
for selection in selections:
df = next(
dotfile = next(
(x for x in dotfiles
if os.path.expanduser(x.dst) == os.path.expanduser(selection)),
None
)
if df:
selected.append(df)
if dotfile:
selected.append(dotfile)
else:
LOG.err('no dotfile matches \"{}\"'.format(selection))
return selected
@@ -716,9 +724,9 @@ def apply_trans(dotpath, dotfile, templater, debug=False):
new_src = '{}.{}'.format(src, TRANS_SUFFIX)
trans = dotfile.trans_r
LOG.dbg('executing transformation: {}'.format(trans))
s = os.path.join(dotpath, src)
srcpath = os.path.join(dotpath, src)
temp = os.path.join(dotpath, new_src)
if not trans.transform(s, temp, templater=templater, debug=debug):
if not trans.transform(srcpath, temp, templater=templater, debug=debug):
msg = 'transformation \"{}\" failed for {}'
LOG.err(msg.format(trans.key, dotfile.key))
if new_src and os.path.exists(new_src):
@@ -731,97 +739,103 @@ def apply_trans(dotpath, dotfile, templater, debug=False):
# main
###########################################################
def _exec_command(opts):
"""execute command"""
ret = True
command = ''
try:
if opts.cmd_profiles:
# list existing profiles
command = 'profiles'
LOG.dbg('running cmd: {}'.format(command))
cmd_list_profiles(opts)
elif opts.cmd_files:
# list files for selected profile
command = 'files'
LOG.dbg('running cmd: {}'.format(command))
cmd_files(opts)
elif opts.cmd_install:
# install the dotfiles stored in dotdrop
command = 'install'
LOG.dbg('running cmd: {}'.format(command))
ret = cmd_install(opts)
elif opts.cmd_compare:
# compare local dotfiles with dotfiles stored in dotdrop
command = 'compare'
LOG.dbg('running cmd: {}'.format(command))
tmp = get_tmpdir()
ret = cmd_compare(opts, tmp)
# clean tmp directory
removepath(tmp, LOG)
elif opts.cmd_import:
# import dotfile(s)
command = 'import'
LOG.dbg('running cmd: {}'.format(command))
ret = cmd_importer(opts)
elif opts.cmd_update:
# update a dotfile
command = 'update'
LOG.dbg('running cmd: {}'.format(command))
ret = cmd_update(opts)
elif opts.cmd_detail:
# detail files
command = 'detail'
LOG.dbg('running cmd: {}'.format(command))
cmd_detail(opts)
elif opts.cmd_remove:
# remove dotfile
command = 'remove'
LOG.dbg('running cmd: {}'.format(command))
cmd_remove(opts)
except KeyboardInterrupt:
LOG.err('interrupted')
ret = False
return ret, command
def main():
"""entry point"""
# check dependencies are met
try:
dependencies_met()
except Exception as e:
LOG.err(e)
except UnmetDependency as exc:
LOG.err(exc)
return False
t0 = time.time()
time0 = time.time()
try:
o = Options()
except YamlException as e:
LOG.err('config error: {}'.format(str(e)))
opts = Options()
except YamlException as exc:
LOG.err('config error: {}'.format(str(exc)))
return False
except UndefinedException as e:
LOG.err('config error: {}'.format(str(e)))
except UndefinedException as exc:
LOG.err('config error: {}'.format(str(exc)))
return False
if o.debug:
LOG.debug = o.debug
if opts.debug:
LOG.debug = opts.debug
LOG.dbg('\n\n')
options_time = time.time() - t0
options_time = time.time() - time0
ret = True
t0 = time.time()
command = ''
try:
if o.cmd_profiles:
# list existing profiles
command = 'profiles'
LOG.dbg('running cmd: {}'.format(command))
cmd_list_profiles(o)
elif o.cmd_files:
# list files for selected profile
command = 'files'
LOG.dbg('running cmd: {}'.format(command))
cmd_files(o)
elif o.cmd_install:
# install the dotfiles stored in dotdrop
command = 'install'
LOG.dbg('running cmd: {}'.format(command))
ret = cmd_install(o)
elif o.cmd_compare:
# compare local dotfiles with dotfiles stored in dotdrop
command = 'compare'
LOG.dbg('running cmd: {}'.format(command))
tmp = get_tmpdir()
ret = cmd_compare(o, tmp)
# clean tmp directory
removepath(tmp, LOG)
elif o.cmd_import:
# import dotfile(s)
command = 'import'
LOG.dbg('running cmd: {}'.format(command))
ret = cmd_importer(o)
elif o.cmd_update:
# update a dotfile
command = 'update'
LOG.dbg('running cmd: {}'.format(command))
ret = cmd_update(o)
elif o.cmd_detail:
# detail files
command = 'detail'
LOG.dbg('running cmd: {}'.format(command))
cmd_detail(o)
elif o.cmd_remove:
# remove dotfile
command = 'remove'
LOG.dbg('running cmd: {}'.format(command))
cmd_remove(o)
except KeyboardInterrupt:
LOG.err('interrupted')
ret = False
cmd_time = time.time() - t0
time0 = time.time()
ret, command = _exec_command(opts)
cmd_time = time.time() - time0
LOG.dbg('done executing command \"{}\"'.format(command))
LOG.dbg('options loaded in {}'.format(options_time))
LOG.dbg('command executed in {}'.format(cmd_time))
if ret and o.conf.save():
if ret and opts.conf.save():
LOG.log('config file updated')
LOG.dbg('return {}'.format(ret))

4
dotdrop/exceptions.py
View File

@@ -12,3 +12,7 @@ class YamlException(Exception):
class UndefinedException(Exception):
"""exception in templating"""
class UnmetDependency(Exception):
"""unmet dependency"""

194
dotdrop/installer.py
View File

@@ -17,6 +17,7 @@ from dotdrop.exceptions import UndefinedException
class Installer:
"""dotfile installer"""
def __init__(self, base='.', create=True, backup=True,
dry=False, safe=False, workdir='~/.config/dotdrop',
@@ -65,7 +66,7 @@ class Installer:
def install(self, templater, src, dst, linktype,
actionexec=None, noempty=False,
ignore=[], is_template=True,
ignore=None, is_template=True,
chmod=None, force_chmod=False):
"""
install src to dst
@@ -93,7 +94,7 @@ class Installer:
return True, None
msg = 'installing \"{}\" to \"{}\" (link: {})'
self.log.dbg(msg.format(src, dst, str(linktype)))
src, dst, cont, err = self._check_paths(src, dst, chmod)
src, dst, cont, err = self._check_paths(src, dst)
if not cont:
return self._log_install(cont, err)
@@ -108,10 +109,13 @@ class Installer:
# install to temporary dir
# and ignore any actions
if self.totemp:
r, err, _ = self.install_to_temp(templater, self.totemp,
src, dst, is_template=is_template,
chmod=chmod, ignore=ignore)
return self._log_install(r, err)
ret, err, _ = self.install_to_temp(templater,
self.totemp,
src, dst,
is_template=is_template,
chmod=chmod,
ignore=ignore)
return self._log_install(ret, err)
isdir = os.path.isdir(src)
self.log.dbg('install {} to {}'.format(src, dst))
@@ -120,20 +124,18 @@ class Installer:
if linktype == LinkTypes.NOLINK:
# normal file
if isdir:
r, err = self._copy_dir(templater, src, dst,
ret, err = self._copy_dir(templater, src, dst,
actionexec=actionexec,
noempty=noempty, ignore=ignore,
is_template=is_template,
chmod=chmod)
is_template=is_template)
else:
r, err = self._copy_file(templater, src, dst,
ret, err = self._copy_file(templater, src, dst,
actionexec=actionexec,
noempty=noempty, ignore=ignore,
is_template=is_template,
chmod=chmod)
is_template=is_template)
elif linktype == LinkTypes.LINK:
# symlink
r, err = self._link(templater, src, dst,
ret, err = self._link(templater, src, dst,
actionexec=actionexec,
is_template=is_template)
elif linktype == LinkTypes.LINK_CHILDREN:
@@ -142,17 +144,17 @@ class Installer:
msg = 'symlink children of {} to {}'
self.log.dbg(msg.format(src, dst))
err = 'source dotfile is not a directory: {}'.format(src)
r = False
ret = False
else:
r, err = self._link_children(templater, src, dst,
ret, err = self._link_children(templater, src, dst,
actionexec=actionexec,
is_template=is_template,
ignore=ignore)
self.log.dbg('before chmod: {} err:{}'.format(r, err))
self.log.dbg('before chmod: {} err:{}'.format(ret, err))
if self.dry:
return self._log_install(r, err)
return self._log_install(ret, err)
# handle chmod
# - on success (r, not err)
@@ -160,7 +162,7 @@ class Installer:
# but not when
# - error (not r, err)
# - aborted (not r, err)
if (r or (not r and not err)):
if (ret or (not ret and not err)):
if not chmod:
chmod = utils.get_file_perm(src)
dstperms = utils.get_file_perm(dst)
@@ -168,21 +170,21 @@ class Installer:
# apply mode
msg = 'chmod {} to {:o}'.format(dst, chmod)
if not force_chmod and self.safe and not self.log.ask(msg):
r = False
ret = False
err = 'aborted'
else:
if not self.comparing:
self.log.sub('chmod {} to {:o}'.format(dst, chmod))
if utils.chmod(dst, chmod, debug=self.debug):
r = True
ret = True
else:
r = False
ret = False
err = 'chmod failed'
return self._log_install(r, err)
return self._log_install(ret, err)
def install_to_temp(self, templater, tmpdir, src, dst,
is_template=True, chmod=None, ignore=[]):
is_template=True, chmod=None, ignore=None):
"""
install a dotfile to a tempdir
@@ -198,9 +200,10 @@ class Installer:
- success, error-if-any, dotfile-installed-path
"""
self.log.dbg('tmp install {} (defined dst: {})'.format(src, dst))
src, dst, cont, err = self._check_paths(src, dst, chmod)
src, dst, cont, err = self._check_paths(src, dst)
if not cont:
return self._log_install(cont, err)
self._log_install(cont, err)
return cont, err, None
ret = False
tmpdst = ''
@@ -253,18 +256,18 @@ class Installer:
self.log.dbg('is a template')
self.log.dbg('install to {}'.format(self.workdir))
tmp = self._pivot_path(dst, self.workdir, striphome=True)
r, err = self.install(templater, src, tmp,
ret, err = self.install(templater, src, tmp,
LinkTypes.NOLINK,
actionexec=actionexec,
is_template=is_template)
if not r and not os.path.exists(tmp):
return r, err
if not ret and not os.path.exists(tmp):
return ret, err
src = tmp
r, err = self._symlink(src, dst, actionexec=actionexec)
return r, err
ret, err = self._symlink(src, dst, actionexec=actionexec)
return ret, err
def _link_children(self, templater, src, dst,
actionexec=None, is_template=True, ignore=[]):
actionexec=None, is_template=True, ignore=None):
"""
install link:link_children
@@ -318,11 +321,11 @@ class Installer:
self.log.dbg('install to {} and symlink'
.format(self.workdir))
tmp = self._pivot_path(subdst, self.workdir, striphome=True)
r, e = self.install(templater, subsrc, tmp,
ret2, err2 = self.install(templater, subsrc, tmp,
LinkTypes.NOLINK,
actionexec=actionexec,
is_template=is_template)
if not r and e and not os.path.exists(tmp):
if not ret2 and err2 and not os.path.exists(tmp):
continue
subsrc = tmp
@@ -369,8 +372,8 @@ class Installer:
overwrite = True
try:
utils.removepath(dst)
except OSError as e:
err = 'something went wrong with {}: {}'.format(src, e)
except OSError as exc:
err = 'something went wrong with {}: {}'.format(src, exc)
return False, err
if self.dry:
self.log.dry('would link {} to {}'.format(dst, src))
@@ -379,9 +382,9 @@ class Installer:
if not self._create_dirs(base):
err = 'error creating directory for {}'.format(dst)
return False, err
r, e = self._exec_pre_actions(actionexec)
if not r:
return False, e
ret, err = self._exec_pre_actions(actionexec)
if not ret:
return False, err
# re-check in case action created the file
if os.path.lexists(dst):
msg = 'Remove "{}" for link creation?'.format(dst)
@@ -389,8 +392,8 @@ class Installer:
return False, 'aborted'
try:
utils.removepath(dst)
except OSError as e:
err = 'something went wrong with {}: {}'.format(src, e)
except OSError as exc:
err = 'something went wrong with {}: {}'.format(src, exc)
return False, err
os.symlink(src, dst)
if not self.comparing:
@@ -399,8 +402,7 @@ class Installer:
def _copy_file(self, templater, src, dst,
actionexec=None, noempty=False,
ignore=[], is_template=True,
chmod=None):
ignore=None, is_template=True):
"""
install src to dst when is a file
@@ -441,8 +443,8 @@ class Installer:
saved = templater.add_tmp_vars(self._get_tmp_file_vars(src, dst))
try:
content = templater.generate(src)
except UndefinedException as e:
return False, str(e)
except UndefinedException as exc:
return False, str(exc)
finally:
templater.restore_vars(saved)
# test is empty
@@ -456,8 +458,7 @@ class Installer:
# write the file
ret, err = self._write(src, dst,
content=content,
actionexec=actionexec,
chmod=chmod)
actionexec=actionexec)
if ret and not err:
if not self.dry and not self.comparing:
self.log.sub('install {} to {}'.format(src, dst))
@@ -465,7 +466,7 @@ class Installer:
def _copy_dir(self, templater, src, dst,
actionexec=None, noempty=False,
ignore=[], is_template=True, chmod=None):
ignore=None, is_template=True):
"""
install src to dst when is a directory
@@ -486,17 +487,16 @@ class Installer:
# handle all files in dir
for entry in os.listdir(src):
f = os.path.join(src, entry)
fpath = os.path.join(src, entry)
self.log.dbg('deploy sub from {}: {}'.format(dst, entry))
if not os.path.isdir(f):
if not os.path.isdir(fpath):
# is file
res, err = self._copy_file(templater, f,
res, err = self._copy_file(templater, fpath,
os.path.join(dst, entry),
actionexec=actionexec,
noempty=noempty,
ignore=ignore,
is_template=is_template,
chmod=None)
is_template=is_template)
if not res and err:
# error occured
ret = res, err
@@ -506,13 +506,12 @@ class Installer:
ret = True, None
else:
# is directory
res, err = self._copy_dir(templater, f,
res, err = self._copy_dir(templater, fpath,
os.path.join(dst, entry),
actionexec=actionexec,
noempty=noempty,
ignore=ignore,
is_template=is_template,
chmod=None)
is_template=is_template)
if not res and err:
# error occured
ret = res, err
@@ -522,8 +521,33 @@ class Installer:
ret = True, None
return ret
@classmethod
def _write_content_to_file(cls, content, src, dst):
"""write content to file"""
if content:
# write content the file
try:
with open(dst, 'wb') as file:
file.write(content)
shutil.copymode(src, dst)
except NotADirectoryError as exc:
err = 'opening dest file: {}'.format(exc)
return False, err
except OSError as exc:
return False, str(exc)
except TypeError as exc:
return False, str(exc)
else:
# copy file
try:
shutil.copyfile(src, dst)
shutil.copymode(src, dst)
except OSError as exc:
return False, str(exc)
return True, None
def _write(self, src, dst, content=None,
actionexec=None, chmod=None):
actionexec=None):
"""
copy dotfile / write content to file
@@ -541,8 +565,8 @@ class Installer:
if os.path.lexists(dst):
try:
os.stat(dst)
except OSError as e:
if e.errno == errno.ENOENT:
except OSError as exc:
if exc.errno == errno.ENOENT:
# broken symlink
err = 'broken symlink {}'.format(dst)
return False, err
@@ -560,47 +584,38 @@ class Installer:
if not self.log.ask('Overwrite \"{}\"'.format(dst)):
return False, 'aborted'
overwrite = True
if self.backup and os.path.lexists(dst):
self._backup(dst)
# create hierarchy
base = os.path.dirname(dst)
if not self._create_dirs(base):
err = 'creating directory for {}'.format(dst)
return False, err
r, e = self._exec_pre_actions(actionexec)
if not r:
return False, e
# execute pre actions
ret, err = self._exec_pre_actions(actionexec)
if not ret:
return False, err
self.log.dbg('install file to \"{}\"'.format(dst))
# re-check in case action created the file
if self.safe and not overwrite and os.path.lexists(dst):
if not self.log.ask('Overwrite \"{}\"'.format(dst)):
if self.safe and not overwrite and \
os.path.lexists(dst) and \
not self.log.ask('Overwrite \"{}\"'.format(dst)):
self.log.warn('ignoring {}'.format(dst))
return False, 'aborted'
if content:
# write content the file
try:
with open(dst, 'wb') as f:
f.write(content)
shutil.copymode(src, dst)
except NotADirectoryError as e:
err = 'opening dest file: {}'.format(e)
return False, err
except Exception as e:
return False, str(e)
else:
# copy file
try:
shutil.copyfile(src, dst)
shutil.copymode(src, dst)
except Exception as e:
return False, str(e)
return True, None
# writing to file
return self._write_content_to_file(content, src, dst)
########################################################
# helpers
########################################################
def _get_tmp_file_vars(self, src, dst):
@classmethod
def _get_tmp_file_vars(cls, src, dst):
tmp = {}
tmp['_dotfile_sub_abs_src'] = src
tmp['_dotfile_sub_abs_dst'] = dst
@@ -615,10 +630,10 @@ class Installer:
if content:
tmp = utils.write_to_tmpfile(content)
src = tmp
r = utils.fastdiff(src, dst)
if r:
ret = utils.fastdiff(src, dst)
if ret:
self.log.dbg('content differ')
return r
return ret
def _show_diff_before_write(self, src, dst, content=None):
"""
@@ -691,7 +706,10 @@ class Installer:
return ret, err
def _log_install(self, boolean, err):
"""log installation process"""
"""
log installation process
returns success, error-if-any
"""
if not self.debug:
return boolean, err
if boolean:
@@ -703,7 +721,7 @@ class Installer:
self.log.dbg('install: IGNORED')
return boolean, err
def _check_paths(self, src, dst, chmod):
def _check_paths(self, src, dst):
"""
check and normalize param
returns <src>, <dst>, <continue>, <error>

66
dotdrop/templategen.py
View File

@@ -30,9 +30,10 @@ COMMENT_END = '@@#}'
class Templategen:
"""dotfile templater"""
def __init__(self, base='.', variables={},
func_file=[], filter_file=[], debug=False):
def __init__(self, base='.', variables=None,
func_file=None, filter_file=None, debug=False):
"""constructor
@base: directory path where to search for templates
@variables: dictionary of variables for templates
@@ -69,13 +70,13 @@ class Templategen:
self.log.dbg('load global functions:')
self._load_funcs_to_dic(jhelpers, self.env.globals)
if func_file:
for f in func_file:
self.log.dbg('load custom functions from {}'.format(f))
self._load_path_to_dic(f, self.env.globals)
for ffile in func_file:
self.log.dbg('load custom functions from {}'.format(ffile))
self._load_path_to_dic(ffile, self.env.globals)
if filter_file:
for f in filter_file:
self.log.dbg('load custom filters from {}'.format(f))
self._load_path_to_dic(f, self.env.filters)
for ffile in filter_file:
self.log.dbg('load custom filters from {}'.format(ffile))
self._load_path_to_dic(ffile, self.env.filters)
if self.debug:
self._debug_dict('template additional variables', variables)
@@ -89,9 +90,9 @@ class Templategen:
return ''
try:
return self._handle_file(src)
except UndefinedError as e:
err = 'undefined variable: {}'.format(e.message)
raise UndefinedException(err)
except UndefinedError as exc:
err = 'undefined variable: {}'.format(exc.message)
raise UndefinedException(err) from exc
def generate_string(self, string):
"""
@@ -103,11 +104,11 @@ class Templategen:
return ''
try:
return self.env.from_string(string).render(self.variables)
except UndefinedError as e:
err = 'undefined variable: {}'.format(e.message)
raise UndefinedException(err)
except UndefinedError as exc:
err = 'undefined variable: {}'.format(exc.message)
raise UndefinedException(err) from exc
def add_tmp_vars(self, newvars={}):
def add_tmp_vars(self, newvars=None):
"""add vars to the globals, make sure to call restore_vars"""
saved_variables = self.variables.copy()
if not newvars:
@@ -139,13 +140,15 @@ class Templategen:
self.log.dbg('load function \"{}\"'.format(name))
dic[name] = func
def _header(self, prepend=''):
@classmethod
def _header(cls, prepend=''):
"""add a comment usually in the header of a dotfile"""
return '{}{}'.format(prepend, utils.header())
def _handle_file(self, src):
"""generate the file content from template"""
try:
# pylint: disable=C0415
import magic
filetype = magic.from_file(src, mime=True)
self.log.dbg('using \"magic\" for filetype identification')
@@ -162,7 +165,8 @@ class Templategen:
return self._handle_bin_file(src)
return self._handle_text_file(src)
def _is_text(self, fileoutput):
@classmethod
def _is_text(cls, fileoutput):
"""return if `file -b` output is ascii text"""
out = fileoutput.lower()
if out.startswith('text'):
@@ -179,8 +183,8 @@ class Templategen:
path = os.path.normpath(path)
if not os.path.exists(path):
raise TemplateNotFound(path)
with open(path, 'r') as f:
content = f.read()
with open(path, 'r') as file:
content = file.read()
return content
def _handle_text_file(self, src):
@@ -199,18 +203,19 @@ class Templategen:
# this is dirty
if not src.startswith(self.base):
src = os.path.join(self.base, src)
with open(src, 'rb') as f:
content = f.read()
with open(src, 'rb') as file:
content = file.read()
return content
def _read_bad_encoded_text(self, path):
@classmethod
def _read_bad_encoded_text(cls, path):
"""decode non utf-8 data"""
with open(path, 'rb') as f:
data = f.read()
with open(path, 'rb') as file:
data = file.read()
return data.decode('utf-8', 'replace')
@staticmethod
def is_template(path, ignore=[]):
def is_template(path, ignore=None):
"""recursively check if any file is a template within path"""
path = os.path.expanduser(path)
@@ -250,10 +255,11 @@ class Templategen:
markers = [BLOCK_START, VAR_START, COMMENT_START]
patterns = [re.compile(marker.encode()) for marker in markers]
try:
with io.open(path, "r", encoding="utf-8") as f:
m = mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ)
with io.open(path, "r", encoding="utf-8") as file:
mapf = mmap.mmap(file.fileno(), 0,
access=mmap.ACCESS_READ)
for pattern in patterns:
if pattern.search(m):
if pattern.search(mapf):
return True
except UnicodeDecodeError:
# is binary so surely no template
@@ -267,5 +273,5 @@ class Templategen:
self.log.dbg('{}:'.format(title))
if not elems:
return
for k, v in elems.items():
self.log.dbg(' - \"{}\": {}'.format(k, v))
for k, val in elems.items():
self.log.dbg(' - \"{}\": {}'.format(k, val))

3
dotdrop/utils.py
View File

@@ -18,6 +18,7 @@ from shutil import rmtree, which
# local import
from dotdrop.logger import Logger
from dotdrop.exceptions import UnmetDependency
LOG = Logger()
STAR = '*'
@@ -316,7 +317,7 @@ def dependencies_met():
err = 'The tool \"{}\" was not found in the PATH!'
for dep in deps:
if not which(dep):
raise Exception(err.format(dep))
raise UnmetDependency(err.format(dep))
# check python deps
err = 'missing python module \"{}\"'