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(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(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, **kwargs):
161 return self.cp_file(path1, path2, **kwargs)
163 def mv(self, path1, path2, recursive: bool = True, **kwargs):
164 """Move files/directories
165 For the specific case of local, all ops on directories are recursive and
166 the recursive= kwarg is ignored.
167 """
168 path1 = self._strip_protocol(path1)
169 path2 = self._strip_protocol(path2)
171 if self.auto_mkdir:
172 self.makedirs(self._parent(path2), exist_ok=True)
174 shutil.move(path1, path2)
176 def link(self, src, dst, **kwargs):
177 src = self._strip_protocol(src)
178 dst = self._strip_protocol(dst)
179 os.link(src, dst, **kwargs)
181 def symlink(self, src, dst, **kwargs):
182 src = self._strip_protocol(src)
183 dst = self._strip_protocol(dst)
184 os.symlink(src, dst, **kwargs)
186 def islink(self, path) -> bool:
187 return os.path.islink(self._strip_protocol(path))
189 def rm_file(self, path):
190 os.remove(self._strip_protocol(path))
192 def rm(self, path, recursive=False, maxdepth=None):
193 if not isinstance(path, list):
194 path = [path]
196 for p in path:
197 p = self._strip_protocol(p)
198 if self.isdir(p) and not self.islink(p):
199 if not recursive:
200 raise ValueError("Cannot delete directory, set recursive=True")
201 if osp.abspath(p) == os.getcwd():
202 raise ValueError("Cannot delete current working directory")
203 shutil.rmtree(p)
204 else:
205 os.remove(p)
207 def unstrip_protocol(self, name):
208 protocol = self.protocol if isinstance(self.protocol, str) else self.protocol[0]
209 name = self._strip_protocol(name) # normalise for local/win/...
210 return f"{protocol}://{name}"
212 def _open(self, path, mode="rb", block_size=None, **kwargs):
213 path = self._strip_protocol(path)
214 if self.auto_mkdir and ("w" in mode or "x" in mode or "a" in mode):
215 self.makedirs(self._parent(path), exist_ok=True)
216 return LocalFileOpener(path, mode, fs=self, **kwargs)
218 def touch(self, path, truncate=True, **kwargs):
219 path = self._strip_protocol(path)
220 if self.auto_mkdir:
221 self.makedirs(self._parent(path), exist_ok=True)
222 if self.exists(path):
223 os.utime(path, None)
224 else:
225 open(path, "a").close()
226 if truncate:
227 os.truncate(path, 0)
229 def created(self, path):
230 info = self.info(path=path)
231 return datetime.datetime.fromtimestamp(
232 info["created"], tz=datetime.timezone.utc
233 )
235 def modified(self, path):
236 info = self.info(path=path)
237 return datetime.datetime.fromtimestamp(info["mtime"], tz=datetime.timezone.utc)
239 @classmethod
240 def _parent(cls, path):
241 path = cls._strip_protocol(path)
242 if os.sep == "/":
243 # posix native
244 return path.rsplit("/", 1)[0] or "/"
245 else:
246 # NT
247 path_ = path.rsplit("/", 1)[0]
248 if len(path_) <= 3:
249 if path_[1:2] == ":":
250 # nt root (something like c:/)
251 return path_[0] + ":/"
252 # More cases may be required here
253 return path_
255 @classmethod
256 def _strip_protocol(cls, path):
257 path = stringify_path(path)
258 protos = (cls.protocol,) if isinstance(cls.protocol, str) else cls.protocol
259 prefixes = (protocol + sep for protocol in protos for sep in ("://", ":"))
260 for prefix in prefixes:
261 if path.startswith(prefix):
262 path = path.removeprefix(prefix)
263 break
265 path = make_path_posix(path)
266 if os.sep != "/":
267 # This code-path is a stripped down version of
268 # > drive, path = ntpath.splitdrive(path)
269 if path[1:2] == ":":
270 # Absolute drive-letter path, e.g. X:\Windows
271 # Relative path with drive, e.g. X:Windows
272 drive, path = path[:2], path[2:]
273 elif path[:2] == "//":
274 # UNC drives, e.g. \\server\share or \\?\UNC\server\share
275 # Device drives, e.g. \\.\device or \\?\device
276 if (index1 := path.find("/", 2)) == -1 or (
277 index2 := path.find("/", index1 + 1)
278 ) == -1:
279 drive, path = path, ""
280 else:
281 drive, path = path[:index2], path[index2:]
282 else:
283 # Relative path, e.g. Windows
284 drive = ""
286 path = path.rstrip("/") or cls.root_marker
287 return drive + path
289 else:
290 return path.rstrip("/") or cls.root_marker
292 def _isfilestore(self):
293 # Inheriting from DaskFileSystem makes this False (S3, etc. were)
294 # the original motivation. But we are a posix-like file system.
295 # See https://github.com/dask/dask/issues/5526
296 return True
298 def chmod(self, path, mode):
299 path = stringify_path(path)
300 return os.chmod(path, mode)
303def make_path_posix(path):
304 """Make path generic and absolute for current OS"""
305 if not isinstance(path, str):
306 if isinstance(path, (list, set, tuple)):
307 return type(path)(make_path_posix(p) for p in path)
308 else:
309 path = stringify_path(path)
310 if not isinstance(path, str):
311 raise TypeError(f"could not convert {path!r} to string")
312 if os.sep == "/":
313 # Native posix
314 if path.startswith("/"):
315 # most common fast case for posix
316 return path
317 elif path.startswith("~"):
318 return osp.expanduser(path)
319 elif path.startswith("./"):
320 path = path[2:]
321 elif path == ".":
322 path = ""
323 return f"{os.getcwd()}/{path}"
324 else:
325 # NT handling
326 if path[0:1] == "/" and path[2:3] == ":":
327 # path is like "/c:/local/path"
328 path = path[1:]
329 if path[1:2] == ":":
330 # windows full path like "C:\\local\\path"
331 if len(path) == 2 or (len(path) == 3 and path[2] in "/\\"):
332 # nt root (something like c:/). A three-character "c:x" is not a
333 # root: it names "x" relative to that drive, so it falls through
334 # instead of being collapsed to the drive root.
335 return path[0] + ":/"
336 path = path.replace("\\", "/")
337 return path
338 elif path[0:1] == "~":
339 return make_path_posix(osp.expanduser(path))
340 elif path.startswith(("\\\\", "//")):
341 # windows UNC/DFS-style paths
342 return "//" + path[2:].replace("\\", "/")
343 elif path.startswith(("\\", "/")):
344 # windows relative path with root
345 path = path.replace("\\", "/")
346 return f"{osp.splitdrive(os.getcwd())[0]}{path}"
347 else:
348 path = path.replace("\\", "/")
349 if path.startswith("./"):
350 path = path[2:]
351 elif path == ".":
352 path = ""
353 return f"{make_path_posix(os.getcwd())}/{path}"
356def trailing_sep(path):
357 """Return True if the path ends with a path separator.
359 A forward slash is always considered a path separator, even on Operating
360 Systems that normally use a backslash.
361 """
362 # TODO: if all incoming paths were posix-compliant then separator would
363 # always be a forward slash, simplifying this function.
364 # See https://github.com/fsspec/filesystem_spec/pull/1250
365 return path.endswith(os.sep) or (os.altsep is not None and path.endswith(os.altsep))
368@lru_cache(maxsize=1)
369def get_umask(mask: int = 0o666) -> int:
370 """Get the current umask.
372 Follows https://stackoverflow.com/a/44130549 to get the umask.
373 Temporarily sets the umask to the given value, and then resets it to the
374 original value.
375 """
376 value = os.umask(mask)
377 os.umask(value)
378 return value
381class LocalFileOpener(io.IOBase):
382 def __init__(
383 self, path, mode, autocommit=True, fs=None, compression=None, **kwargs
384 ):
385 logger.debug("open file: %s", path)
386 self.path = path
387 self.mode = mode
388 self.fs = fs
389 self.f = None
390 self.autocommit = autocommit
391 self.compression = get_compression(path, compression)
392 self.blocksize = io.DEFAULT_BUFFER_SIZE
393 self._open()
395 def _open(self):
396 if self.f is None or self.f.closed:
397 if self.autocommit or "w" not in self.mode:
398 self.f = open(self.path, mode=self.mode)
399 if self.compression:
400 compress = compr[self.compression]
401 self.f = compress(self.f, mode=self.mode)
402 else:
403 # TODO: check if path is writable?
404 i, name = tempfile.mkstemp()
405 os.close(i) # we want normal open and normal buffered file
406 self.temp = name
407 self.f = open(name, mode=self.mode)
408 if "w" not in self.mode:
409 self.size = self.f.seek(0, 2)
410 self.f.seek(0)
411 self.f.size = self.size
413 def _fetch_range(self, start, end):
414 # probably only used by cached FS
415 if "r" not in self.mode:
416 raise ValueError
417 self._open()
418 self.f.seek(start)
419 return self.f.read(end - start)
421 def __setstate__(self, state):
422 self.f = None
423 loc = state.pop("loc", None)
424 self.__dict__.update(state)
425 if "r" in state["mode"]:
426 self.f = None
427 self._open()
428 self.f.seek(loc)
430 def __getstate__(self):
431 d = self.__dict__.copy()
432 d.pop("f")
433 if "r" in self.mode:
434 d["loc"] = self.f.tell()
435 else:
436 if not self.f.closed:
437 raise ValueError("Cannot serialise open write-mode local file")
438 return d
440 def commit(self):
441 if self.autocommit:
442 raise RuntimeError("Can only commit if not already set to autocommit")
443 if not self.f.closed:
444 # a compression wrapper (e.g., GzipFile) does not close the file
445 # object it was given, so buffered bytes and any trailer may still
446 # be pending here. Windows also refuses to rename an open file.
447 self.f.close()
448 try:
449 shutil.move(self.temp, self.path)
450 except PermissionError as e:
451 # shutil.move raises PermissionError if os.rename
452 # and the default copy2 fallback with shutil.copystats fail.
453 # The file should be there nonetheless, but without copied permissions.
454 # If it doesn't exist, there was no permission to create the file.
455 if not os.path.exists(self.path):
456 raise e
457 else:
458 # If PermissionError is not raised, permissions can be set.
459 try:
460 mask = 0o666
461 os.chmod(self.path, mask & ~get_umask(mask))
462 except RuntimeError:
463 pass
465 def discard(self):
466 if self.autocommit:
467 raise RuntimeError("Cannot discard if set to autocommit")
468 os.remove(self.temp)
470 def readable(self) -> bool:
471 return True
473 def writable(self) -> bool:
474 return "r" not in self.mode
476 def read(self, *args, **kwargs):
477 return self.f.read(*args, **kwargs)
479 def write(self, *args, **kwargs):
480 return self.f.write(*args, **kwargs)
482 def tell(self, *args, **kwargs):
483 return self.f.tell(*args, **kwargs)
485 def seek(self, *args, **kwargs):
486 return self.f.seek(*args, **kwargs)
488 def seekable(self, *args, **kwargs):
489 return self.f.seekable(*args, **kwargs)
491 def readline(self, *args, **kwargs):
492 return self.f.readline(*args, **kwargs)
494 def readlines(self, *args, **kwargs):
495 return self.f.readlines(*args, **kwargs)
497 def close(self):
498 return self.f.close()
500 def truncate(self, size=None) -> int:
501 return self.f.truncate(size)
503 @property
504 def closed(self):
505 return self.f.closed
507 def fileno(self):
508 return self.raw.fileno()
510 def flush(self) -> None:
511 self.f.flush()
513 def __iter__(self):
514 return self.f.__iter__()
516 def __getattr__(self, item):
517 return getattr(self.f, item)
519 def __enter__(self):
520 self._incontext = True
521 return self
523 def __exit__(self, exc_type, exc_value, traceback):
524 self._incontext = False
525 self.f.__exit__(exc_type, exc_value, traceback)