Skip to content

Utils

General utilities.

makedirs(path, exist_ok=False, mode=None)

Make the specified directory and any parent directories.

Source code in dvc_task/utils.py 
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
def makedirs(path: str, exist_ok: bool = False, mode: Optional[int] = None):
    """Make the specified directory and any parent directories."""
    if mode is None:
        os.makedirs(path, exist_ok=exist_ok)
        return

    # Modified version of os.makedirs() with support for extended mode
    # (e.g. S_ISGID)
    head, tail = os.path.split(path)
    if not tail:
        head, tail = os.path.split(head)
    if head and tail and not os.path.exists(head):
        try:
            makedirs(head, exist_ok=exist_ok, mode=mode)
        except FileExistsError:
            # Defeats race condition when another thread created the path
            pass
        cdir = os.curdir
        if tail == cdir:  # foo/newdir/. exists if foo/newdir exists
            return
    try:
        os.mkdir(path, mode)
    except OSError:
        # Cannot rely on checking for EEXIST, since the operating system
        # could give priority to other errors like EACCES or EROFS
        if not exist_ok or not os.path.isdir(path):
            raise

    try:
        os.chmod(path, mode)
    except OSError:
        logger.debug("failed to chmod '%o' '%s'", mode, path, exc_info=True)

remove(path)

Remove the specified path.

Source code in dvc_task/utils.py 
35
36
37
38
39
40
41
42
43
44
45
def remove(path: str):
    """Remove the specified path."""
    logger.debug("Removing '%s'", path)
    try:
        if os.path.isdir(path):
            shutil.rmtree(path, onerror=_chmod)
        else:
            _unlink(path, _chmod)
    except OSError as exc:
        if exc.errno != errno.ENOENT:
            raise

unc_path(path)

Return UNC formatted path.

Returns the unmodified path on posix platforms.

Source code in dvc_task/utils.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def unc_path(path: str) -> str:
    """Return UNC formatted path.

    Returns the unmodified path on posix platforms.
    """
    # Celery/Kombu URLs only take absolute filesystem paths
    # (UNC paths on windows)
    path = os.path.abspath(path)
    if os.name != "nt":
        return path
    drive, tail = os.path.splitdrive(path)
    if drive.endswith(":"):
        return f"\\\\?\\{drive}{tail}"
    return f"{drive}{tail}"