Coverage for /usr/local/lib/python3.12/site-packages/prefect/utilities/filesystem.py: 23%
85 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 10:48 +0000
1"""
2Utilities for working with file systems
3"""
5import os 1a
6import pathlib 1a
7import threading 1a
8from collections.abc import Iterable 1a
9from contextlib import contextmanager 1a
10from pathlib import Path, PureWindowsPath 1a
11from typing import Any, AnyStr, Optional, Union, cast 1a
13# fsspec has no stubs, see https://github.com/fsspec/filesystem_spec/issues/625
14import fsspec # type: ignore 1a
15import pathspec 1a
16from fsspec.core import OpenFile # type: ignore 1a
17from fsspec.implementations.local import LocalFileSystem # type: ignore 1a
19import prefect 1a
22def create_default_ignore_file(path: str) -> bool: 1a
23 """
24 Creates default ignore file in the provided path if one does not already exist; returns boolean specifying
25 whether a file was created.
26 """
27 _path = pathlib.Path(path)
28 ignore_file = _path / ".prefectignore"
29 if ignore_file.exists():
30 return False
31 default_file = pathlib.Path(prefect.__module_path__) / ".prefectignore"
32 with ignore_file.open(mode="w") as f:
33 f.write(default_file.read_text())
34 return True
37def filter_files( 1a
38 root: str = ".",
39 ignore_patterns: Optional[Iterable[AnyStr]] = None,
40 include_dirs: bool = True,
41) -> set[str]:
42 """
43 This function accepts a root directory path and a list of file patterns to ignore, and returns
44 a list of files that excludes those that should be ignored.
46 The specification matches that of [.gitignore files](https://git-scm.com/docs/gitignore).
47 """
48 spec = pathspec.PathSpec.from_lines("gitwildmatch", ignore_patterns or [])
49 ignored_files = {p.path for p in spec.match_tree_entries(root)}
50 if include_dirs:
51 all_files = {p.path for p in pathspec.util.iter_tree_entries(root)}
52 else:
53 all_files = set(pathspec.util.iter_tree_files(root))
54 included_files = all_files - ignored_files
55 return included_files
58chdir_lock: threading.Lock = threading.Lock() 1a
61def _normalize_path(path: Union[str, Path]) -> str: 1a
62 """
63 Normalize a path, handling UNC paths on Windows specially.
64 """
65 path = Path(path)
67 # Handle UNC paths on Windows differently
68 if os.name == "nt" and str(path).startswith("\\\\"):
69 # For UNC paths, use absolute() instead of resolve()
70 # to avoid the Windows path resolution issues
71 return str(path.absolute())
72 else:
73 try:
74 # For non-UNC paths, try resolve() first
75 return str(path.resolve())
76 except OSError:
77 # Fallback to absolute() if resolve() fails
78 return str(path.absolute())
81@contextmanager 1a
82def tmpchdir(path: str): 1a
83 """
84 Change current-working directories for the duration of the context,
85 with special handling for UNC paths on Windows.
86 """
87 path = _normalize_path(path)
89 if os.path.isfile(path) or (not os.path.exists(path) and not path.endswith("/")):
90 path = os.path.dirname(path)
92 owd = os.getcwd()
94 with chdir_lock:
95 try:
96 # On Windows with UNC paths, we need to handle the directory change carefully
97 if os.name == "nt" and path.startswith("\\\\"):
98 # Use os.path.abspath to handle UNC paths
99 os.chdir(os.path.abspath(path))
100 else:
101 os.chdir(path)
102 yield path
103 finally:
104 os.chdir(owd)
107def filename(path: str) -> str: 1a
108 """Extract the file name from a path with remote file system support"""
109 try:
110 of: OpenFile = cast(OpenFile, fsspec.open(path)) # type: ignore # no typing stubs available
111 sep = cast(str, of.fs.sep) # type: ignore # no typing stubs available
112 except (ImportError, AttributeError):
113 sep = "\\" if "\\" in path else "/"
114 return path.split(sep)[-1]
117def is_local_path(path: Union[str, pathlib.Path, Any]) -> bool: 1a
118 """Check if the given path points to a local or remote file system"""
119 if isinstance(path, str):
120 try:
121 of = cast(OpenFile, fsspec.open(path)) # type: ignore # no typing stubs available
122 except ImportError:
123 # The path is a remote file system that uses a lib that is not installed
124 return False
125 elif isinstance(path, pathlib.Path):
126 return True
127 else:
128 of = path
130 return isinstance(of.fs, LocalFileSystem)
133def to_display_path( 1a
134 path: Union[pathlib.Path, str],
135 relative_to: Optional[Union[pathlib.Path, str]] = None,
136) -> str:
137 """
138 Convert a path to a displayable path. The absolute path or relative path to the
139 current (or given) directory will be returned, whichever is shorter.
140 """
141 path, relative_to = (
142 pathlib.Path(path).resolve(),
143 pathlib.Path(relative_to or ".").resolve(),
144 )
145 relative_path = str(path.relative_to(relative_to))
146 absolute_path = str(path)
147 return relative_path if len(relative_path) < len(absolute_path) else absolute_path
150def relative_path_to_current_platform(path_str: str) -> Path: 1a
151 """
152 Converts a relative path generated on any platform to a relative path for the
153 current platform.
154 """
156 return Path(PureWindowsPath(path_str).as_posix())
159def get_open_file_limit() -> int: 1a
160 """Get the maximum number of open files allowed for the current process"""
162 try:
163 if os.name == "nt":
164 import ctypes
166 return ctypes.cdll.ucrtbase._getmaxstdio()
167 else:
168 import resource
170 soft_limit, _ = resource.getrlimit(resource.RLIMIT_NOFILE)
171 return soft_limit
172 except Exception:
173 # Catch all exceptions, as ctypes can raise several errors
174 # depending on what went wrong. Return a safe default if we
175 # can't get the limit from the OS.
176 return 200