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(
243        raw_path: str,
244        strip: bool = True,
245        encoding: typing.Union[str, None] = DEFAULT_ENCODING,
246        ) -> str:
247    """ Read the contents of a file. """
248
249    if (encoding is None):
250        encoding = DEFAULT_ENCODING
251
252    path = os.path.abspath(raw_path)
253
254    if (not exists(path)):
255        raise ValueError(f"Source of read does not exist: '{raw_path}'.")
256
257    with open(path, 'r', encoding = encoding) as file:
258        contents = file.read()
259
260    if (strip):
261        contents = contents.strip()
262
263    return contents
264
265def write_file(
266        raw_path: str,
267        contents: typing.Union[str, None],
268        strip: bool = True,
269        newline: bool = True,
270        encoding: typing.Union[str, None] = DEFAULT_ENCODING,
271        no_clobber: bool = False,
272        ) -> None:
273    """
274    Write the contents of a file.
275    If clobbering, any existing dirent will be removed before write.
276    """
277
278    if (encoding is None):
279        encoding = DEFAULT_ENCODING
280
281    path = os.path.abspath(raw_path)
282
283    if (exists(path)):
284        if (no_clobber):
285            raise ValueError(f"Destination of write already exists: '{raw_path}'.")
286
287        remove(path)
288
289    if (contents is None):
290        contents = ''
291
292    if (strip):
293        contents = contents.strip()
294
295    if (newline):
296        contents += "\n"
297
298    with open(path, 'w', encoding = encoding) as file:
299        file.write(contents)
300
301def read_file_bytes(raw_path: str) -> bytes:
302    """ Read the contents of a file as bytes. """
303
304    path = os.path.abspath(raw_path)
305
306    if (not exists(path)):
307        raise ValueError(f"Source of read bytes does not exist: '{raw_path}'.")
308
309    with open(path, 'rb') as file:
310        return file.read()
311
312def write_file_bytes(
313        raw_path: str, contents: typing.Union[bytes, str, None],
314        no_clobber: bool = False) -> None:
315    """
316    Write the contents of a file as bytes.
317    If clobbering, any existing dirent will be removed before write.
318    """
319
320    if (contents is None):
321        contents = b''
322
323    if (isinstance(contents, str)):
324        contents = contents.encode(DEFAULT_ENCODING)
325
326    path = os.path.abspath(raw_path)
327
328    if (exists(path)):
329        if (no_clobber):
330            raise ValueError(f"Destination of write bytes already exists: '{raw_path}'.")
331
332        remove(path)
333
334    with open(path, 'wb') as file:
335        file.write(contents)
336
337def contains_path(parent: str, child: str) -> bool:
338    """
339    Check if the parent path contains the child path.
340    This is pure lexical analysis, no dirent stats are checked.
341    Will return false if the (absolute) paths are the same
342    (this function does not allow a path to contain itself).
343    """
344
345    if ((parent == '') or (child == '')):
346        return False
347
348    parent = os.path.abspath(parent)
349    child = os.path.abspath(child)
350
351    child = os.path.dirname(child)
352    for _ in range(DEPTH_LIMIT):
353        if (parent == child):
354            return True
355
356        new_child = os.path.dirname(child)
357        if (child == new_child):
358            return False
359
360        child = new_child
361
362    raise ValueError("Depth limit reached.")
363
364def hash_file(raw_path: str) -> str:
365    """
366    Compute the SHA256 hash of the file (see edq.util.hash.sha256_hex()).
367    Links will has their path (according to os.readlink()).
368    Directories will raise an exception.
369    """
370
371    path = os.path.abspath(raw_path)
372
373    contents: typing.Any = None
374
375    if (not exists(path)):
376        raise ValueError(f"Target of hash file does not exist: '{raw_path}'.")
377
378    if (os.path.islink(path)):
379        contents = os.readlink(path)
380    elif (os.path.isfile(path)):
381        contents = read_file_bytes(raw_path)
382    else:
383        raise ValueError(f"Target of hash file is not a file: '{raw_path}'.")
384
385    return edq.util.hash.sha256_hex(contents)
386
387def tree(raw_path: str, hash_files: bool = False) -> typing.Dict[str, typing.Union[None, str, typing.Dict[str, typing.Any]]]:
388    """
389    Return a tree structure that includes all descendants of the given dirent (including the dirent itself).
390    If `hash_files` is true, then the value of non-dir keys will be the SHA256 hash of the file (see hash_file()),
391    otherwise the value will be None.
392    """
393
394    path = os.path.abspath(raw_path)
395
396    if (not exists(path)):
397        raise ValueError(f"Target of tree does not exist: '{raw_path}'.")
398
399    return {
400        os.path.basename(path): _tree(path, hash_files, 0),
401    }
402
403def _tree(path: str, hash_files: bool, level: int) -> typing.Union[str, None, typing.Dict[str, typing.Any]]:
404    """ Recursive helper for tree(). """
405
406    if (level > DEPTH_LIMIT):
407        raise ValueError("Depth limit reached.")
408
409    if (not os.path.isdir(path)):
410        if (hash_files):
411            return hash_file(path)
412
413        return None
414
415    result = {}
416    for child in sorted(os.listdir(path)):
417        result[child] = _tree(os.path.join(path, child), hash_files, level + 1)
418
419    return result