1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
'''fileops.py
Performs file operations such as pack/unpack,
ensuring directories exist,... imports snakeoils osutils
functions for use throughout catalyst.
'''
import glob
import os
import shutil
from stat import ST_UID, ST_GID, ST_MODE
from snakeoil.osutils import ensure_dirs as snakeoil_ensure_dirs
from catalyst import log
from catalyst.support import CatalystError
def ensure_dirs(path, gid=-1, uid=-1, mode=0o755, minimal=True,
failback=None, fatal=False):
'''Wrapper to snakeoil.osutil's ensure_dirs()
This additionally allows for failures to run
cleanup or other code and/or raise fatal errors.
:param path: directory to ensure exists on disk
:param gid: a valid GID to set any created directories to
:param uid: a valid UID to set any created directories to
:param mode: permissions to set any created directories to
:param minimal: boolean controlling whether or not the specified mode
must be enforced, or is the minimal permissions necessary. For example,
if mode=0o755, minimal=True, and a directory exists with mode 0707,
this will restore the missing group perms resulting in 757.
:param failback: function to run in the event of a failed attemp
to create the directory.
:return: True if the directory could be created/ensured to have those
permissions, False if not.
'''
succeeded = snakeoil_ensure_dirs(
path, gid=gid, uid=uid, mode=mode, minimal=minimal)
if not succeeded:
if failback:
failback()
if fatal:
raise CatalystError(
"Failed to create directory: %s" % path, print_traceback=True)
return succeeded
def clear_dir(target, mode=0o755, remove=False):
'''Universal directory clearing function
@target: string, path to be cleared or removed
@mode: integer, desired mode to set the directory to
@remove: boolean, passed through to clear_dir()
@return boolean
'''
log.debug('start: %s', target)
if not target:
log.debug('no target... returning')
return False
mystat = None
if os.path.isdir(target) and not os.path.islink(target):
log.notice('Emptying directory: %s', target)
# stat the dir, delete the dir, recreate the dir and set
# the proper perms and ownership
try:
log.debug('os.stat()')
mystat = os.stat(target)
log.debug('shutil.rmtree()')
shutil.rmtree(target)
except Exception:
log.error('clear_dir failed', exc_info=True)
return False
elif os.path.exists(target):
log.debug("Clearing (unlinking) non-directory: %s", target)
os.unlink(target)
else:
log.debug("Conditions not met to clear: %s", target)
log.debug(" isdir: %s", os.path.isdir(target))
log.debug(" islink: %s", os.path.islink(target))
log.debug(" exists: %s", os.path.exists(target))
if not remove:
log.debug('ensure_dirs()')
ensure_dirs(target, mode=mode)
if mystat:
os.chown(target, mystat[ST_UID], mystat[ST_GID])
os.chmod(target, mystat[ST_MODE])
log.debug('DONE, returning True')
return True
def clear_path(target_path):
"""Nuke |target_path| regardless of it being a dir, file or glob."""
targets = glob.iglob(target_path, recursive=True)
for path in targets:
clear_dir(path, remove=True)
def move_path(src, dest):
'''Move a source target to a new destination
:param src: source path to move
:param dest: destination path to move it to
:returns: boolean
'''
log.debug('Start move_path(%s, %s)', src, dest)
if os.path.isdir(src) and not os.path.islink(src):
if os.path.exists(dest):
log.warning('Removing existing target destination: %s', dest)
if not clear_dir(dest, remove=True):
return False
log.debug('Moving source...')
try:
shutil.move(src, dest)
except Exception:
log.error('move_path failed', exc_info=True)
return False
return True
return False
|
GitWeb