mirror of
ssh://git.janware.com/srv/git/janware/proj/jw-python
synced 2026-01-15 09:53:32 +01:00
Jan Lindemann
47f4de938f
commit_tmpfile() is a better os.rename which doesn't do anything if source and target files are equal, and logs if does something. Signed-off-by: Jan Lindemann <jan@janware.com>
74 lines
2.2 KiB
Python
74 lines
2.2 KiB
Python
import os, errno
|
|
import atexit
|
|
import tempfile
|
|
import inspect
|
|
from jwutils import log
|
|
|
|
_tmpfiles = set()
|
|
|
|
def _cleanup():
|
|
for f in _tmpfiles:
|
|
silentremove(f)
|
|
|
|
def silentremove(filename): #export
|
|
try:
|
|
os.remove(filename)
|
|
except OSError as e:
|
|
if e.errno != errno.ENOENT:
|
|
raise # re-raise exception if a different error occurred
|
|
|
|
def pad(token, total_size, right_align = False):
|
|
add = total_size - len(token)
|
|
if add <= 0:
|
|
return token
|
|
space = ' ' * add
|
|
if right_align:
|
|
return space + token
|
|
return token + space
|
|
|
|
def atomic_store(contents, path): # export
|
|
if path[0:3] == '/dev':
|
|
with open(path, 'w') as outfile:
|
|
outfile.write(contents)
|
|
return
|
|
outfile = tempfile.NamedTemporaryFile(prefix=os.path.basename(path), delete=False, dir=os.path.dirname(path))
|
|
name = outfile.name
|
|
_tmpfiles.add(name)
|
|
outfile.write(contents)
|
|
outfile.close()
|
|
os.rename(name, path)
|
|
_tmpfiles.remove(name)
|
|
|
|
# see https://stackoverflow.com/questions/2020014
|
|
def object_builtin_name(o, full=True): # export
|
|
#if not full:
|
|
# return o.__class__.__name__
|
|
module = o.__class__.__module__
|
|
if module is None or module == str.__class__.__module__:
|
|
return o.__class__.__name__ # Avoid reporting __builtin__
|
|
return module + '.' + o.__class__.__name__
|
|
|
|
def get_derived_classes(mod, base): # export
|
|
members = inspect.getmembers(mod, inspect.isclass)
|
|
r = []
|
|
for name, c in members:
|
|
log.slog(log.DEBUG, "found ", name)
|
|
if inspect.isabstract(c):
|
|
log.slog(log.DEBUG, " is abstract")
|
|
continue
|
|
if not base in inspect.getmro(c):
|
|
log.slog(log.DEBUG, " is not derived from", base, "only", inspect.getmro(c))
|
|
continue
|
|
r.append(c)
|
|
return r
|
|
|
|
def commit_tmpfile(tmp, path): # export
|
|
caller = log.get_caller_pos()
|
|
if os.path.isfile(path) and filecmp.cmp(tmp, path):
|
|
log.slog(log.INFO, "{} is up to date".format(path), caller=caller)
|
|
os.unlink(tmp)
|
|
else:
|
|
log.slog(log.NOTICE, "saving {}".format(path), caller=caller)
|
|
os.rename(path + '.tmp', path)
|
|
|
|
atexit.register(_cleanup)
|