edq.util.dirent

  1"""
  2Operations relating to directory entries (dirents).
  3
  4These operations are designed for clarity and compatibility, not performance.
  5
  6Only directories, files, and links will be handled.
  7Other types of dirents may result in an error being raised.
  8
  9In general, all recursive operations do not follow symlinks by default and instead treat the link as a file.
 10"""
 11
 12import atexit
 13import os
 14import shutil
 15import tempfile
 16import typing
 17import uuid
 18
 19import edq.util.constants
 20import edq.util.hash
 21
 22DEFAULT_ENCODING: str = edq.util.constants.DEFAULT_ENCODING
 23""" The default encoding that will be used when reading and writing. """
 24
 25DEPTH_LIMIT: int = 10000
 26
 27def exists(path: str) -> bool:
 28    """
 29    Check if a path exists.
 30    This will transparently call os.path.lexists(),
 31    which will include broken links.
 32    """
 33
 34    return os.path.lexists(path)
 35
 36def get_temp_path(prefix: str = '', suffix: str = '', rm: bool = True) -> str:
 37    """
 38    Get a path to a valid (but not currently existing) temp dirent.
 39    If rm is True, then the dirent will be attempted to be deleted on exit
 40    (no error will occur if the path is not there).
 41    """
 42
 43    path = None
 44    while ((path is None) or exists(path)):
 45        path = os.path.join(tempfile.gettempdir(), prefix + str(uuid.uuid4()) + suffix)
 46
 47    path = os.path.realpath(path)
 48
 49    if (rm):
 50        atexit.register(remove, path)
 51
 52    return path
 53
 54def get_temp_dir(prefix: str = '', suffix: str = '', rm: bool = True) -> str:
 55    """
 56    Get a temp directory.
 57    The directory will exist when returned.
 58    """
 59
 60    path = get_temp_path(prefix = prefix, suffix = suffix, rm = rm)
 61    mkdir(path)
 62    return path
 63
 64def mkdir(raw_path: str) -> None:
 65    """
 66    Make a directory (including any required parent directories).
 67    Does not complain if the directory (or parents) already exist
 68    (this includes if the directory or parents are links to directories).
 69    """
 70
 71    path = os.path.abspath(raw_path)
 72
 73    if (exists(path)):
 74        if (os.path.isdir(path)):
 75            return
 76
 77        raise ValueError(f"Target of mkdir already exists, and is not a dir: '{raw_path}'.")
 78
 79    _check_parent_dirs(raw_path)
 80
 81    os.makedirs(path, exist_ok = True)
 82
 83def _check_parent_dirs(raw_path: str) -> None:
 84    """
 85    Check all parents to ensure that they are all dirs (or don't exist).
 86    This is naturally handled by os.makedirs(),
 87    but the error messages are not consistent between POSIX and Windows.
 88    """
 89
 90    path = os.path.abspath(raw_path)
 91
 92    parent_path = path
 93    for _ in range(DEPTH_LIMIT):
 94        new_parent_path = os.path.dirname(parent_path)
 95        if (parent_path == new_parent_path):
 96            # We have reached root (are our own parent).
 97            return
 98
 99        parent_path = new_parent_path
100
101        if (os.path.exists(parent_path) and (not os.path.isdir(parent_path))):
102            raise ValueError(f"Target of mkdir contains parent ('{os.path.basename(parent_path)}') that exists and is not a dir: '{raw_path}'.")
103
104    raise ValueError("Depth limit reached.")
105
106def remove(path: str) -> None:
107    """
108    Remove the given path.
109    The path can be of any type (dir, file, link),
110    and does not need to exist.
111    """
112
113    if (not exists(path)):
114        return
115
116    if (os.path.isfile(path) or os.path.islink(path)):
117        os.remove(path)
118    elif (os.path.isdir(path)):
119        shutil.rmtree(path)
120    else:
121        raise ValueError(f"Unknown type of dirent: '{path}'.")
122
123def same(a: str, b: str) -> bool:
124    """
125    Check if two paths represent the same dirent.
126    If either (or both) paths do not exist, false will be returned.
127    If either paths are links, they are resolved before checking
128    (so a link and the target file are considered the "same").
129    """
130
131    return (exists(a) and exists(b) and os.path.samefile(a, b))
132
133def move(raw_source: str, raw_dest: str, no_clobber: bool = False) -> None:
134    """
135    Move the source dirent to the given destination.
136    Any existing destination will be removed before moving.
137    """
138
139    source = os.path.abspath(raw_source)
140    dest = os.path.abspath(raw_dest)
141
142    if (not exists(source)):
143        raise ValueError(f"Source of move does not exist: '{raw_source}'.")
144
145    # If dest is a dir, then resolve the path.
146    if (os.path.isdir(dest)):
147        dest = os.path.abspath(os.path.join(dest, os.path.basename(source)))
148
149    # Skip if this is self.
150    if (same(source, dest)):
151        return
152
153    # Check for clobber.
154    if (exists(dest)):
155        if (no_clobber):
156            raise ValueError(f"Destination of move already exists: '{raw_dest}'.")
157
158        remove(dest)
159
160    # Create any required parents.
161    os.makedirs(os.path.dirname(dest), exist_ok = True)
162
163    shutil.move(source, dest)
164
165def copy(raw_source: str, raw_dest: str, no_clobber: bool = False) -> None:
166    """
167    Copy a dirent or directory to a destination.
168
169    The destination will be overwritten if it exists (and no_clobber is false).
170    For copying the contents of a directory INTO another directory, use copy_contents().
171
172    No copy is made if the source and dest refer to the same dirent.
173    """
174
175    source = os.path.abspath(raw_source)
176    dest = os.path.abspath(raw_dest)
177
178    if (same(source, dest)):
179        return
180
181    if (not exists(source)):
182        raise ValueError(f"Source of copy does not exist: '{raw_source}'.")
183
184    if (contains_path(source, dest)):
185        raise ValueError(f"Source of copy cannot contain the destination. Source: '{raw_source}', Destination: '{raw_dest}'.")
186
187    if (contains_path(dest, source)):
188        raise ValueError(f"Destination of copy cannot contain the source. Destination: '{raw_dest}', Source: '{raw_source}'.")
189
190    if (exists(dest)):
191        if (no_clobber):
192            raise ValueError(f"Destination of copy already exists: '{raw_dest}'.")
193
194        remove(dest)
195
196    mkdir(os.path.dirname(dest))
197
198    if (os.path.islink(source)):
199        # shutil.copy2() can generally handle (broken) links, but Windows is inconsistent (between 3.11 and 3.12) on link handling.
200        link_target = os.readlink(source)
201        os.symlink(link_target, dest)
202    elif (os.path.isfile(source)):
203        shutil.copy2(source, dest, follow_symlinks = False)
204    elif (os.path.isdir(source)):
205        mkdir(dest)
206
207        for child in sorted(os.listdir(source)):
208            copy(os.path.join(raw_source, child), os.path.join(raw_dest, child))
209    else:
210        raise ValueError(f"Source of copy is not a dir, fie, or link: '{raw_source}'.")
211
212def copy_contents(raw_source: str, raw_dest: str, no_clobber: bool = False) -> None:
213    """
214    Copy a file or the contents of a directory (excluding the top-level directory itself) into a destination.
215    If the destination exists, it must be a directory.
216
217    The source and destination should not be the same file.
218
219    For a file, this is equivalent to `mkdir -p dest && cp source dest`
220    For a dir, this is equivalent to `mkdir -p dest && cp -r source/* dest`
221    """
222
223    source = os.path.abspath(raw_source)
224    dest = os.path.abspath(raw_dest)
225
226    if (same(source, dest)):
227        raise ValueError(f"Source and destination of contents copy cannot be the same: '{raw_source}'.")
228
229    if (exists(dest) and (not os.path.isdir(dest))):
230        raise ValueError(f"Destination of contents copy exists and is not a dir: '{raw_dest}'.")
231
232    mkdir(dest)
233
234    if (os.path.isfile(source) or os.path.islink(source)):
235        copy(source, os.path.join(dest, os.path.basename(source)), no_clobber = no_clobber)
236    elif (os.path.isdir(source)):
237        for child in sorted(os.listdir(source)):
238            copy(os.path.join(raw_source, child), os.path.join(raw_dest, child), no_clobber = no_clobber)
239    else:
240        raise ValueError(f"Source of contents copy is not a dir, fie, or link: '{raw_source}'.")
241
242def read_file(raw_path: str, strip: bool = True, encoding: str = DEFAULT_ENCODING) -> str:
243    """ Read the contents of a file. """
244
245    path = os.path.abspath(raw_path)
246
247    if (not exists(path)):
248        raise ValueError(f"Source of read does not exist: '{raw_path}'.")
249
250    with open(path, 'r', encoding = encoding) as file:
251        contents = file.read()
252
253    if (strip):
254        contents = contents.strip()
255
256    return contents
257
258def write_file(
259        raw_path: str, contents: typing.Union[str, None],
260        strip: bool = True, newline: bool = True,
261        encoding: str = DEFAULT_ENCODING,
262        no_clobber: bool = False) -> None:
263    """
264    Write the contents of a file.
265    If clobbering, any existing dirent will be removed before write.
266    """
267
268    path = os.path.abspath(raw_path)
269
270    if (exists(path)):
271        if (no_clobber):
272            raise ValueError(f"Destination of write already exists: '{raw_path}'.")
273
274        remove(path)
275
276    if (contents is None):
277        contents = ''
278
279    if (strip):
280        contents = contents.strip()
281
282    if (newline):
283        contents += "\n"
284
285    with open(path, 'w', encoding = encoding) as file:
286        file.write(contents)
287
288def read_file_bytes(raw_path: str) -> bytes:
289    """ Read the contents of a file as bytes. """
290
291    path = os.path.abspath(raw_path)
292
293    if (not exists(path)):
294        raise ValueError(f"Source of read bytes does not exist: '{raw_path}'.")
295
296    with open(path, 'rb') as file:
297        return file.read()
298
299def write_file_bytes(
300        raw_path: str, contents: typing.Union[bytes, str, None],
301        no_clobber: bool = False) -> None:
302    """
303    Write the contents of a file as bytes.
304    If clobbering, any existing dirent will be removed before write.
305    """
306
307    if (contents is None):
308        contents = b''
309
310    if (isinstance(contents, str)):
311        contents = contents.encode(DEFAULT_ENCODING)
312
313    path = os.path.abspath(raw_path)
314
315    if (exists(path)):
316        if (no_clobber):
317            raise ValueError(f"Destination of write bytes already exists: '{raw_path}'.")
318
319        remove(path)
320
321    with open(path, 'wb') as file:
322        file.write(contents)
323
324def contains_path(parent: str, child: str) -> bool:
325    """
326    Check if the parent path contains the child path.
327    This is pure lexical analysis, no dirent stats are checked.
328    Will return false if the (absolute) paths are the same
329    (this function does not allow a path to contain itself).
330    """
331
332    if ((parent == '') or (child == '')):
333        return False
334
335    parent = os.path.abspath(parent)
336    child = os.path.abspath(child)
337
338    child = os.path.dirname(child)
339    for _ in range(DEPTH_LIMIT):
340        if (parent == child):
341            return True
342
343        new_child = os.path.dirname(child)
344        if (child == new_child):
345            return False
346
347        child = new_child
348
349    raise ValueError("Depth limit reached.")
350
351def hash_file(raw_path: str) -> str:
352    """
353    Compute the SHA256 hash of the file (see edq.util.hash.sha256_hex()).
354    Links will has their path (according to os.readlink()).
355    Directories will raise an exception.
356    """
357
358    path = os.path.abspath(raw_path)
359
360    contents: typing.Any = None
361
362    if (not exists(path)):
363        raise ValueError(f"Target of hash file does not exist: '{raw_path}'.")
364
365    if (os.path.islink(path)):
366        contents = os.readlink(path)
367    elif (os.path.isfile(path)):
368        contents = read_file_bytes(raw_path)
369    else:
370        raise ValueError(f"Target of hash file is not a file: '{raw_path}'.")
371
372    return edq.util.hash.sha256_hex(contents)
373
374def tree(raw_path: str, hash_files: bool = False) -> typing.Dict[str, typing.Union[None, str, typing.Dict[str, typing.Any]]]:
375    """
376    Return a tree structure that includes all descendants of the given dirent (including the dirent itself).
377    If `hash_files` is true, then the value of non-dir keys will be the SHA256 hash of the file (see hash_file()),
378    otherwise the value will be None.
379    """
380
381    path = os.path.abspath(raw_path)
382
383    if (not exists(path)):
384        raise ValueError(f"Target of tree does not exist: '{raw_path}'.")
385
386    return {
387        os.path.basename(path): _tree(path, hash_files, 0),
388    }
389
390def _tree(path: str, hash_files: bool, level: int) -> typing.Union[str, None, typing.Dict[str, typing.Any]]:
391    """ Recursive helper for tree(). """
392
393    if (level > DEPTH_LIMIT):
394        raise ValueError("Depth limit reached.")
395
396    if (not os.path.isdir(path)):
397        if (hash_files):
398            return hash_file(path)
399
400        return None
401
402    result = {}
403    for child in sorted(os.listdir(path)):
404        result[child] = _tree(os.path.join(path, child), hash_files, level + 1)
405
406    return result