3 """Utilities for working with files."""
11 from typing import Optional
13 from os.path import isfile, join, exists
15 logger = logging.getLogger(__name__)
18 def create_path_if_not_exist(path, on_error=None):
20 Attempts to create path if it does not exist. If on_error is
21 specified, it is called with an exception if one occurs, otherwise
22 exception is rethrown.
26 >>> path = os.path.join("/tmp", str(uuid.uuid4()), str(uuid.uuid4()))
27 >>> os.path.exists(path)
29 >>> create_path_if_not_exist(path)
30 >>> os.path.exists(path)
33 logger.debug(f"Creating path {path}")
34 previous_umask = os.umask(0)
39 if ex.errno != errno.EEXIST and not os.path.isdir(path):
40 if on_error is not None:
45 os.umask(previous_umask)
48 def does_file_exist(filename: str) -> bool:
49 return os.path.exists(filename) and os.path.isfile(filename)
52 def does_directory_exist(dirname: str) -> bool:
53 return os.path.exists(dirname) and os.path.isdir(dirname)
56 def does_path_exist(pathname: str) -> bool:
57 return os.path.exists(pathname)
60 def get_file_size(filename: str) -> int:
61 return os.path.getsize(filename)
64 def is_normal_file(filename: str) -> bool:
65 return os.path.isfile(filename)
68 def is_directory(filename: str) -> bool:
69 return os.path.isdir(filename)
72 def is_symlink(filename: str) -> bool:
73 return os.path.islink(filename)
76 def is_same_file(file1: str, file2: str) -> bool:
77 return os.path.samefile(file1, file2)
80 def get_file_raw_timestamps(filename: str) -> Optional[os.stat_result]:
82 return os.stat(filename)
83 except Exception as e:
88 def get_file_raw_timestamp(filename: str, extractor) -> Optional[float]:
89 tss = get_file_raw_timestamps(filename)
95 def get_file_raw_atime(filename: str) -> Optional[float]:
96 return get_file_raw_timestamp(filename, lambda x: x.st_atime)
99 def get_file_raw_mtime(filename: str) -> Optional[float]:
100 return get_file_raw_timestamp(filename, lambda x: x.st_mtime)
103 def get_file_raw_ctime(filename: str) -> Optional[float]:
104 return get_file_raw_timestamp(filename, lambda x: x.st_ctime)
107 def get_file_md5(filename: str) -> str:
108 file_hash = hashlib.md5()
109 with open(filename, "rb") as f:
112 file_hash.update(chunk)
114 return file_hash.hexdigest()
117 def set_file_raw_atime(filename: str, atime: float):
118 mtime = get_file_raw_mtime(filename)
119 os.utime(filename, (atime, mtime))
122 def set_file_raw_mtime(filename: str, mtime: float):
123 atime = get_file_raw_atime(filename)
124 os.utime(filename, (atime, mtime))
127 def set_file_raw_atime_and_mtime(filename: str, ts: float = None):
129 os.utime(filename, (ts, ts))
131 os.utime(filename, None)
134 def convert_file_timestamp_to_datetime(
135 filename: str, producer
136 ) -> Optional[datetime.datetime]:
137 ts = producer(filename)
139 return datetime.datetime.fromtimestamp(ts)
143 def get_file_atime_as_datetime(filename: str) -> Optional[datetime.datetime]:
144 return convert_file_timestamp_to_datetime(filename, get_file_raw_atime)
147 def get_file_mtime_as_datetime(filename: str) -> Optional[datetime.datetime]:
148 return convert_file_timestamp_to_datetime(filename, get_file_raw_mtime)
151 def get_file_ctime_as_datetime(filename: str) -> Optional[datetime.datetime]:
152 return convert_file_timestamp_to_datetime(filename, get_file_raw_ctime)
155 def get_file_timestamp_age_seconds(filename: str, extractor) -> Optional[int]:
157 ts = get_file_raw_timestamps(filename)
160 result = extractor(ts)
164 def get_file_atime_age_seconds(filename: str) -> Optional[int]:
165 return get_file_timestamp_age_seconds(filename, lambda x: x.st_atime)
168 def get_file_ctime_age_seconds(filename: str) -> Optional[int]:
169 return get_file_timestamp_age_seconds(filename, lambda x: x.st_ctime)
172 def get_file_mtime_age_seconds(filename: str) -> Optional[int]:
173 return get_file_timestamp_age_seconds(filename, lambda x: x.st_mtime)
176 def get_file_timestamp_timedelta(
177 filename: str, extractor
178 ) -> Optional[datetime.timedelta]:
179 age = get_file_timestamp_age_seconds(filename, extractor)
181 return datetime.timedelta(seconds=float(age))
185 def get_file_atime_timedelta(filename: str) -> Optional[datetime.timedelta]:
186 return get_file_timestamp_timedelta(filename, lambda x: x.st_atime)
189 def get_file_ctime_timedelta(filename: str) -> Optional[datetime.timedelta]:
190 return get_file_timestamp_timedelta(filename, lambda x: x.st_ctime)
193 def get_file_mtime_timedelta(filename: str) -> Optional[datetime.timedelta]:
194 return get_file_timestamp_timedelta(filename, lambda x: x.st_mtime)
197 def describe_file_timestamp(
198 filename: str, extractor, *, brief=False
200 from datetime_utils import describe_duration, describe_duration_briefly
201 age = get_file_timestamp_age_seconds(filename, extractor)
205 return describe_duration_briefly(age)
207 return describe_duration(age)
210 def describe_file_atime(filename: str, *, brief=False) -> Optional[str]:
211 return describe_file_timestamp(filename, lambda x: x.st_atime, brief=brief)
214 def describe_file_ctime(filename: str, *, brief=False) -> Optional[str]:
215 return describe_file_timestamp(filename, lambda x: x.st_ctime, brief=brief)
218 def describe_file_mtime(filename: str, *, brief=False) -> Optional[str]:
219 return describe_file_timestamp(filename, lambda x: x.st_mtime, brief=brief)
222 def expand_globs(in_filename: str):
223 for filename in glob.glob(in_filename):
227 def get_files(directory: str):
228 for filename in os.listdir(directory):
229 full_path = join(directory, filename)
230 if isfile(full_path) and exists(full_path):
234 def get_directories(directory: str):
235 for d in os.listdir(directory):
236 full_path = join(directory, d)
237 if not isfile(full_path) and exists(full_path):
241 def get_files_recursive(directory: str):
242 for filename in get_files(directory):
244 for subdir in get_directories(directory):
245 for file_or_directory in get_files_recursive(subdir):
246 yield file_or_directory