Skip to content

prefect.utilities.filesystem

Utilities for working with file systems

create_default_ignore_file(path)

Creates default ignore file in the provided path if one does not already exist; returns boolean specifying whether a file was created.

Source code in src/prefect/utilities/filesystem.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
def create_default_ignore_file(path: str) -> bool:
    """
    Creates default ignore file in the provided path if one does not already exist; returns boolean specifying
    whether a file was created.
    """
    path = pathlib.Path(path)
    ignore_file = path / ".prefectignore"
    if ignore_file.exists():
        return False
    default_file = pathlib.Path(prefect.__module_path__) / ".prefectignore"
    with ignore_file.open(mode="w") as f:
        f.write(default_file.read_text())
    return True

filename(path)

Extract the file name from a path with remote file system support

Source code in src/prefect/utilities/filesystem.py 
76
77
78
79
80
81
82
83
def filename(path: str) -> str:
    """Extract the file name from a path with remote file system support"""
    try:
        of: OpenFile = fsspec.open(path)
        sep = of.fs.sep
    except (ImportError, AttributeError):
        sep = "\\" if "\\" in path else "/"
    return path.split(sep)[-1]

filter_files(root='.', ignore_patterns=None, include_dirs=True)

This function accepts a root directory path and a list of file patterns to ignore, and returns a list of files that excludes those that should be ignored.

The specification matches that of .gitignore files.

Source code in src/prefect/utilities/filesystem.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def filter_files(
    root: str = ".", ignore_patterns: Optional[list] = None, include_dirs: bool = True
) -> set:
    """
    This function accepts a root directory path and a list of file patterns to ignore, and returns
    a list of files that excludes those that should be ignored.

    The specification matches that of [.gitignore files](https://git-scm.com/docs/gitignore).
    """
    spec = pathspec.PathSpec.from_lines("gitwildmatch", ignore_patterns or [])
    ignored_files = {p.path for p in spec.match_tree_entries(root)}
    if include_dirs:
        all_files = {p.path for p in pathspec.util.iter_tree_entries(root)}
    else:
        all_files = set(pathspec.util.iter_tree_files(root))
    included_files = all_files - ignored_files
    return included_files

get_open_file_limit()

Get the maximum number of open files allowed for the current process

Source code in src/prefect/utilities/filesystem.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def get_open_file_limit() -> int:
    """Get the maximum number of open files allowed for the current process"""

    try:
        if os.name == "nt":
            import ctypes

            return ctypes.cdll.ucrtbase._getmaxstdio()
        else:
            import resource

            soft_limit, _ = resource.getrlimit(resource.RLIMIT_NOFILE)
            return soft_limit
    except Exception:
        # Catch all exceptions, as ctypes can raise several errors
        # depending on what went wrong. Return a safe default if we
        # can't get the limit from the OS.
        return 200

is_local_path(path)

Check if the given path points to a local or remote file system

Source code in src/prefect/utilities/filesystem.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def is_local_path(path: Union[str, pathlib.Path, OpenFile]):
    """Check if the given path points to a local or remote file system"""
    if isinstance(path, str):
        try:
            of = fsspec.open(path)
        except ImportError:
            # The path is a remote file system that uses a lib that is not installed
            return False
    elif isinstance(path, pathlib.Path):
        return True
    elif isinstance(path, OpenFile):
        of = path
    else:
        raise TypeError(f"Invalid path of type {type(path).__name__!r}")

    return type(of.fs) == LocalFileSystem

relative_path_to_current_platform(path_str)

Converts a relative path generated on any platform to a relative path for the current platform.

Source code in src/prefect/utilities/filesystem.py 
120
121
122
123
124
125
126
def relative_path_to_current_platform(path_str: str) -> Path:
    """
    Converts a relative path generated on any platform to a relative path for the
    current platform.
    """

    return Path(PureWindowsPath(path_str).as_posix())

tmpchdir(path)

Change current-working directories for the duration of the context

Source code in src/prefect/utilities/filesystem.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@contextmanager
def tmpchdir(path: str):
    """
    Change current-working directories for the duration of the context
    """
    path = os.path.abspath(path)
    if os.path.isfile(path) or (not os.path.exists(path) and not path.endswith("/")):
        path = os.path.dirname(path)

    owd = os.getcwd()

    with chdir_lock:
        try:
            os.chdir(path)
            yield path
        finally:
            os.chdir(owd)

to_display_path(path, relative_to=None)

Convert a path to a displayable path. The absolute path or relative path to the current (or given) directory will be returned, whichever is shorter.

Source code in src/prefect/utilities/filesystem.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def to_display_path(
    path: Union[pathlib.Path, str], relative_to: Union[pathlib.Path, str] = None
) -> str:
    """
    Convert a path to a displayable path. The absolute path or relative path to the
    current (or given) directory will be returned, whichever is shorter.
    """
    path, relative_to = (
        pathlib.Path(path).resolve(),
        pathlib.Path(relative_to or ".").resolve(),
    )
    relative_path = str(path.relative_to(relative_to))
    absolute_path = str(path)
    return relative_path if len(relative_path) < len(absolute_path) else absolute_path