Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import datetime
2import io
3import logging
4import os
5import os.path as osp
6import shutil
7import stat
8import tempfile
9from functools import lru_cache
11from fsspec.compression import compr
12from fsspec.core import get_compression
13from fsspec.utils import isfilelike, stringify_path
15from ..spec import AbstractFileSystem
17logger = logging.getLogger("fsspec.local")
20class LocalFileSystem(AbstractFileSystem):
21 """Interface to files on local storage
23 Parameters
24 ----------
25 auto_mkdir: bool
26 Whether, when opening a file, the directory containing it should
27 be created (if it doesn't already exist). This is assumed by pyarrow
28 code.
29 """
31 root_marker = "/"
32 protocol = "file", "local"
33 local_file = True
35 def __init__(self, auto_mkdir=False, **kwargs):
36 super().__init__(**kwargs)
37 self.auto_mkdir = auto_mkdir
39 @property
40 def fsid(self):
41 return "local"
43 def mkdir(self, path, create_parents=True, **kwargs):
44 path = self._strip_protocol(path)
45 if self.exists(path):
46 raise FileExistsError(path)
47 if create_parents:
48 self.makedirs(path, exist_ok=True)
49 else:
50 os.mkdir(path, **kwargs)
52 def makedirs(self, path, exist_ok=False):
53 path = self._strip_protocol(path)
54 os.makedirs(path, exist_ok=exist_ok)
56 def rmdir(self, path):
57 path = self._strip_protocol(path)
58 os.rmdir(path)
60 def ls(self, path, detail=False, **kwargs):
61 path = self._strip_protocol(path)
62 path_info = self.info(path)
63 infos = []
64 if path_info["type"] == "directory":
65 with os.scandir(path) as it:
66 for f in it:
67 try:
68 # Only get the info if requested since it is a bit expensive (the stat call inside)
69 # The strip_protocol is also used in info() and calls make_path_posix to always return posix paths
70 info = self.info(f) if detail else self._strip_protocol(f.path)
71 infos.append(info)
72 except FileNotFoundError:
73 pass
74 else:
75 infos = [path_info] if detail else [path_info["name"]]
77 return infos
79 def info(self, path, **kwargs):
80 if isinstance(path, os.DirEntry):
81 # scandir DirEntry
82 out = path.stat(follow_symlinks=False)
83 link = path.is_symlink()
84 if path.is_dir(follow_symlinks=False):
85 t = "directory"
86 elif path.is_file(follow_symlinks=False):
87 t = "file"
88 else:
89 t = "other"
91 size = out.st_size
92 if link:
93 try:
94 out2 = path.stat(follow_symlinks=True)
95 size = out2.st_size
96 except OSError:
97 size = 0
98 path = self._strip_protocol(path.path)
99 else:
100 # str or path-like
101 path = self._strip_protocol(path)
102 out = os.stat(path, follow_symlinks=False)
103 link = stat.S_ISLNK(out.st_mode)
104 if link:
105 out = os.stat(path, follow_symlinks=True)
106 size = out.st_size
107 if stat.S_ISDIR(out.st_mode):
108 t = "directory"
109 elif stat.S_ISREG(out.st_mode):
110 t = "file"
111 else:
112 t = "other"
114 # Check for the 'st_birthtime' attribute, which is not always present; fallback to st_ctime
115 created_time = getattr(out, "st_birthtime", out.st_ctime)
117 result = {
118 "name": path,
119 "size": size,
120 "type": t,
121 "created": created_time,
122 "islink": link,
123 }
124 for field in ["mode", "uid", "gid", "mtime", "ino", "nlink"]:
125 result[field] = getattr(out, f"st_{field}")
126 if link:
127 result["destination"] = os.readlink(path)
128 return result
130 def lexists(self, path, **kwargs):
131 return osp.lexists(self._strip_protocol(path))
133 def cp_file(self, path1, path2, **kwargs):
134 path1 = self._strip_protocol(path1)
135 path2 = self._strip_protocol(path2)
136 if self.auto_mkdir:
137 self.makedirs(self._parent(path2), exist_ok=True)
138 if self.isfile(path1):
139 shutil.copyfile(path1, path2)
140 elif self.isdir(path1):
141 self.mkdirs(path2, exist_ok=True)
142 else:
143 raise FileNotFoundError(path1)
145 def isfile(self, path):
146 path = self._strip_protocol(path)
147 return os.path.isfile(path)
149 def isdir(self, path):
150 path = self._strip_protocol(path)
151 return os.path.isdir(path)
153 def get_file(self, path1, path2, callback=None, **kwargs):
154 if isfilelike(path2):
155 with open(self._strip_protocol(path1), "rb") as f:
156 shutil.copyfileobj(f, path2)
157 else:
158 return self.cp_file(path1, path2, **kwargs)
160 def put_file(self, path1, path2, callback=None, mode="overwrite", **kwargs):
161 if mode == "create" and self.exists(path2):
162 raise FileExistsError(path2)
163 return self.cp_file(path1, path2, **kwargs)
165 def mv(self, path1, path2, recursive: bool = True, **kwargs):
166 """Move files/directories
167 For the specific case of local, all ops on directories are recursive and
168 the recursive= kwarg is ignored.
169 """
170 path1 = self._strip_protocol(path1)
171 path2 = self._strip_protocol(path2)
173 if self.auto_mkdir:
174 self.makedirs(self._parent(path2), exist_ok=True)
176 shutil.move(path1, path2)
178 def link(self, src, dst, **kwargs):
179 src = self._strip_protocol(src)
180 dst = self._strip_protocol(dst)
181 os.link(src, dst, **kwargs)
183 def symlink(self, src, dst, **kwargs):
184 src = self._strip_protocol(src)
185 dst = self._strip_protocol(dst)
186 os.symlink(src, dst, **kwargs)
188 def islink(self, path) -> bool:
189 return os.path.islink(self._strip_protocol(path))
191 def rm_file(self, path):
192 os.remove(self._strip_protocol(path))
194 def rm(self, path, recursive=False, maxdepth=None):
195 if not isinstance(path, list):
196 path = [path]
198 for p in path:
199 p = self._strip_protocol(p)
200 if self.isdir(p) and not self.islink(p):
201 if not recursive:
202 raise ValueError("Cannot delete directory, set recursive=True")
203 if osp.abspath(p) == os.getcwd():
204 raise ValueError("Cannot delete current working directory")
205 if maxdepth is None:
206 shutil.rmtree(p)
207 else:
208 # only delete down to maxdepth; a directory still holding
209 # deeper files is kept
210 paths = self.expand_path(p, recursive=True, maxdepth=maxdepth)
211 for sub in reversed(paths):
212 if not self.isdir(sub) or self.islink(sub):
213 os.remove(sub)
214 elif not os.listdir(sub):
215 os.rmdir(sub)
216 else:
217 os.remove(p)
219 def unstrip_protocol(self, name):
220 protocol = self.protocol if isinstance(self.protocol, str) else self.protocol[0]
221 name = self._strip_protocol(name) # normalise for local/win/...
222 return f"{protocol}://{name}"
224 def _open(self, path, mode="rb", block_size=None, **kwargs):
225 path = self._strip_protocol(path)
226 if self.auto_mkdir and ("w" in mode or "x" in mode or "a" in mode):
227 self.makedirs(self._parent(path), exist_ok=True)
228 return LocalFileOpener(path, mode, fs=self, **kwargs)
230 def touch(self, path, truncate=True, **kwargs):
231 path = self._strip_protocol(path)
232 if self.auto_mkdir:
233 self.makedirs(self._parent(path), exist_ok=True)
234 if self.exists(path):
235 os.utime(path, None)
236 else:
237 open(path, "a").close()
238 if truncate:
239 os.truncate(path, 0)
241 def created(self, path):
242 info = self.info(path=path)
243 return datetime.datetime.fromtimestamp(info["created"], tz=datetime.UTC)
245 def modified(self, path):
246 info = self.info(path=path)
247 return datetime.datetime.fromtimestamp(info["mtime"], tz=datetime.UTC)
249 @classmethod
250 def _parent(cls, path):
251 path = cls._strip_protocol(path)
252 if os.sep == "/":
253 # posix native
254 return path.rsplit("/", 1)[0] or "/"
255 else:
256 # NT
257 path_ = path.rsplit("/", 1)[0]
258 if len(path_) <= 3:
259 if path_[1:2] == ":":
260 # nt root (something like c:/)
261 return path_[0] + ":/"
262 # More cases may be required here
263 return path_
265 @classmethod
266 def _strip_protocol(cls, path):
267 path = stringify_path(path)
268 protos = (cls.protocol,) if isinstance(cls.protocol, str) else cls.protocol
269 prefixes = (protocol + sep for protocol in protos for sep in ("://", ":"))
270 for prefix in prefixes:
271 if path.startswith(prefix):
272 path = path.removeprefix(prefix)
273 break
275 path = make_path_posix(path)
276 if os.sep != "/":
277 # This code-path is a stripped down version of
278 # > drive, path = ntpath.splitdrive(path)
279 if path[1:2] == ":":
280 # Absolute drive-letter path, e.g. X:\Windows
281 # Relative path with drive, e.g. X:Windows
282 drive, path = path[:2], path[2:]
283 elif path[:2] == "//":
284 # UNC drives, e.g. \\server\share or \\?\UNC\server\share
285 # Device drives, e.g. \\.\device or \\?\device
286 if (index1 := path.find("/", 2)) == -1 or (
287 index2 := path.find("/", index1 + 1)
288 ) == -1:
289 drive, path = path, ""
290 else:
291 drive, path = path[:index2], path[index2:]
292 else:
293 # Relative path, e.g. Windows
294 drive = ""
296 path = path.rstrip("/") or cls.root_marker
297 return drive + path
299 else:
300 return path.rstrip("/") or cls.root_marker
302 def _isfilestore(self):
303 # Inheriting from DaskFileSystem makes this False (S3, etc. were)
304 # the original motivation. But we are a posix-like file system.
305 # See https://github.com/dask/dask/issues/5526
306 return True
308 def chmod(self, path, mode):
309 path = self._strip_protocol(path)
310 return os.chmod(path, mode)
313def make_path_posix(path):
314 """Make path generic and absolute for current OS"""
315 if not isinstance(path, str):
316 if isinstance(path, (list, set, tuple)):
317 return type(path)(make_path_posix(p) for p in path)
318 else:
319 path = stringify_path(path)
320 if not isinstance(path, str):
321 raise TypeError(f"could not convert {path!r} to string")
322 if os.sep == "/":
323 # Native posix
324 if path.startswith("/"):
325 # most common fast case for posix
326 return path
327 elif path.startswith("~"):
328 return osp.expanduser(path)
329 elif path.startswith("./"):
330 path = path[2:]
331 elif path == ".":
332 path = ""
333 return f"{os.getcwd()}/{path}"
334 else:
335 # NT handling
336 if path[0:1] == "/" and path[2:3] == ":":
337 # path is like "/c:/local/path"
338 path = path[1:]
339 if path[1:2] == ":":
340 # windows full path like "C:\\local\\path"
341 if len(path) == 2 or (len(path) == 3 and path[2] in "/\\"):
342 # nt root (something like c:/). A three-character "c:x" is not a
343 # root: it names "x" relative to that drive, so it falls through
344 # instead of being collapsed to the drive root.
345 return path[0] + ":/"
346 path = path.replace("\\", "/")
347 return path
348 elif path[0:1] == "~":
349 return make_path_posix(osp.expanduser(path))
350 elif path.startswith(("\\\\", "//")):
351 # windows UNC/DFS-style paths
352 return "//" + path[2:].replace("\\", "/")
353 elif path.startswith(("\\", "/")):
354 # windows relative path with root
355 path = path.replace("\\", "/")
356 return f"{osp.splitdrive(os.getcwd())[0]}{path}"
357 else:
358 path = path.replace("\\", "/")
359 if path.startswith("./"):
360 path = path[2:]
361 elif path == ".":
362 path = ""
363 return f"{make_path_posix(os.getcwd())}/{path}"
366def trailing_sep(path):
367 """Return True if the path ends with a path separator.
369 A forward slash is always considered a path separator, even on Operating
370 Systems that normally use a backslash.
371 """
372 # TODO: if all incoming paths were posix-compliant then separator would
373 # always be a forward slash, simplifying this function.
374 # See https://github.com/fsspec/filesystem_spec/pull/1250
375 return path.endswith(os.sep) or (os.altsep is not None and path.endswith(os.altsep))
378@lru_cache(maxsize=1)
379def get_umask(mask: int = 0o666) -> int:
380 """Get the current umask.
382 Follows https://stackoverflow.com/a/44130549 to get the umask.
383 Temporarily sets the umask to the given value, and then resets it to the
384 original value.
385 """
386 value = os.umask(mask)
387 os.umask(value)
388 return value
391class LocalFileOpener(io.IOBase):
392 def __init__(
393 self, path, mode, autocommit=True, fs=None, compression=None, **kwargs
394 ):
395 logger.debug("open file: %s", path)
396 self.path = path
397 self.mode = mode
398 self.fs = fs
399 self.f = None
400 self.autocommit = autocommit
401 self.compression = get_compression(path, compression)
402 self.blocksize = io.DEFAULT_BUFFER_SIZE
403 self._open()
405 def _open(self):
406 if self.f is None or self.f.closed:
407 if self.autocommit or "w" not in self.mode:
408 self.f = open(self.path, mode=self.mode)
409 if self.compression:
410 compress = compr[self.compression]
411 self.f = compress(self.f, mode=self.mode)
412 else:
413 # TODO: check if path is writable?
414 i, name = tempfile.mkstemp()
415 os.close(i) # we want normal open and normal buffered file
416 self.temp = name
417 self.f = open(name, mode=self.mode)
418 if "r" in self.mode:
419 self.size = self.f.seek(0, 2)
420 self.f.seek(0)
421 self.f.size = self.size
423 def _fetch_range(self, start, end):
424 # probably only used by cached FS
425 if "r" not in self.mode:
426 raise ValueError
427 self._open()
428 self.f.seek(start)
429 return self.f.read(end - start)
431 def __setstate__(self, state):
432 self.f = None
433 loc = state.pop("loc", None)
434 self.__dict__.update(state)
435 if "r" in state["mode"]:
436 self.f = None
437 self._open()
438 self.f.seek(loc)
440 def __getstate__(self):
441 d = self.__dict__.copy()
442 d.pop("f")
443 if "r" in self.mode:
444 d["loc"] = self.f.tell()
445 else:
446 if not self.f.closed:
447 raise ValueError("Cannot serialise open write-mode local file")
448 return d
450 def commit(self):
451 if self.autocommit:
452 raise RuntimeError("Can only commit if not already set to autocommit")
453 if not self.f.closed:
454 # a compression wrapper (e.g., GzipFile) does not close the file
455 # object it was given, so buffered bytes and any trailer may still
456 # be pending here. Windows also refuses to rename an open file.
457 self.f.close()
458 try:
459 shutil.move(self.temp, self.path)
460 except PermissionError as e:
461 # shutil.move raises PermissionError if os.rename
462 # and the default copy2 fallback with shutil.copystats fail.
463 # The file should be there nonetheless, but without copied permissions.
464 # If it doesn't exist, there was no permission to create the file.
465 if not os.path.exists(self.path):
466 raise e
467 else:
468 # If PermissionError is not raised, permissions can be set.
469 try:
470 mask = 0o666
471 os.chmod(self.path, mask & ~get_umask(mask))
472 except RuntimeError:
473 pass
475 def discard(self):
476 if self.autocommit:
477 raise RuntimeError("Cannot discard if set to autocommit")
478 if not self.f.closed:
479 self.f.close()
480 os.remove(self.temp)
482 def readable(self) -> bool:
483 return True
485 def writable(self) -> bool:
486 return "r" not in self.mode
488 def read(self, *args, **kwargs):
489 return self.f.read(*args, **kwargs)
491 def write(self, *args, **kwargs):
492 return self.f.write(*args, **kwargs)
494 def tell(self, *args, **kwargs):
495 return self.f.tell(*args, **kwargs)
497 def seek(self, *args, **kwargs):
498 return self.f.seek(*args, **kwargs)
500 def seekable(self, *args, **kwargs):
501 return self.f.seekable(*args, **kwargs)
503 def readline(self, *args, **kwargs):
504 return self.f.readline(*args, **kwargs)
506 def readlines(self, *args, **kwargs):
507 return self.f.readlines(*args, **kwargs)
509 def close(self):
510 return self.f.close()
512 def truncate(self, size=None) -> int:
513 return self.f.truncate(size)
515 @property
516 def closed(self):
517 return self.f.closed
519 def fileno(self):
520 return self.raw.fileno()
522 def flush(self) -> None:
523 self.f.flush()
525 def __iter__(self):
526 return self.f.__iter__()
528 def __getattr__(self, item):
529 return getattr(self.f, item)
531 def __enter__(self):
532 self._incontext = True
533 return self
535 def __exit__(self, exc_type, exc_value, traceback):
536 self._incontext = False
537 self.f.__exit__(exc_type, exc_value, traceback)