Coverage for python/lsst/resources/file.py: 87%
254 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-31 01:33 -0700
1# This file is part of lsst-resources.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# Use of this source code is governed by a 3-clause BSD-style
10# license that can be found in the LICENSE file.
12from __future__ import annotations
14__all__ = ("FileResourcePath",)
16import contextlib
17import copy
18import datetime
19import logging
20import os
21import os.path
22import posixpath
23import re
24import shutil
25import stat
26import urllib.parse
27from collections.abc import Generator, Iterator
28from pathlib import Path
29from typing import TYPE_CHECKING
31from ._resourceHandles._baseResourceHandle import ResourceHandleProtocol
32from ._resourceHandles._fileResourceHandle import FileResourceHandle
33from ._resourcePath import ResourceInfo, ResourcePath
34from .utils import NoTransaction, ensure_directory_is_writeable, os2posix, posix2os
36try:
37 import fsspec
38 from fsspec.spec import AbstractFileSystem
39except ImportError:
40 # Hidden from type checkers so that the names above keep the types they
41 # have when fsspec is installed.
42 if not TYPE_CHECKING:
43 fsspec = None
44 AbstractFileSystem = type
46if TYPE_CHECKING:
47 from importlib.resources.abc import Traversable
49 from .utils import TransactionProtocol
52log = logging.getLogger(__name__)
55def _path_to_info(uri: str, path: str | Path | Traversable) -> ResourceInfo | None:
56 """Given a path to a local file, return a `ResourceInfo`."""
57 if isinstance(path, Path):
58 stat_result = path.stat()
59 elif isinstance(path, str): 59 ↛ 61line 59 didn't jump to line 61 because the condition on line 59 was always true
60 stat_result = os.stat(path)
61 elif (stat_method := getattr(path, "stat", None)) and callable(stat_method):
62 # Edge case triggered by importlib.resources.
63 stat_result = stat_method()
64 if not isinstance(stat_result, os.stat_result):
65 raise RuntimeError(f"Unexpected stat result from {path}.stat()")
66 else:
67 return None
69 return ResourceInfo(
70 uri=uri,
71 is_file=not stat.S_ISDIR(stat_result.st_mode),
72 size=0 if stat.S_ISDIR(stat_result.st_mode) else stat_result.st_size,
73 last_modified=datetime.datetime.fromtimestamp(stat_result.st_mtime, tz=datetime.UTC),
74 checksums={},
75 )
78class FileResourcePath(ResourcePath):
79 """Path for explicit ``file`` URI scheme."""
81 transferModes = ("copy", "link", "symlink", "hardlink", "relsymlink", "auto", "move")
82 transferDefault: str = "link"
84 # By definition refers to a local file
85 isLocal = True
87 # A warm existence check or removal here costs a few microseconds, almost
88 # all of it holding the GIL, so a batch has to be this large before
89 # spreading it over threads beats a plain loop in the calling thread.
90 # Measurements above this size are flat, so it also matches the 1000 keys
91 # an S3 bulk delete takes.
92 _chunk_size = 1000
94 @property
95 def ospath(self) -> str:
96 """Path component of the URI localized to current OS.
98 Will unquote URI path since a formal URI must include the quoting.
99 """
100 return urllib.parse.unquote(posix2os(self._uri.path))
102 def exists(self) -> bool:
103 """Indicate that the file exists."""
104 # Uses os.path.exists so if there is a soft link that points
105 # to a file that no longer exists this will return False
106 return os.path.exists(self.ospath)
108 def size(self) -> int:
109 """Return the size of the file in bytes."""
110 if not os.path.isdir(self.ospath):
111 stat = os.stat(self.ospath)
112 sz = stat.st_size
113 else:
114 sz = 0
115 return sz
117 def get_info(self) -> ResourceInfo:
118 """Return lightweight metadata about this file."""
119 info = _path_to_info(str(self), self.ospath)
120 if info is None: 120 ↛ 121line 120 didn't jump to line 121 because the condition on line 120 was never true
121 raise RuntimeError(f"Unexpected internal failure obtaining file info for {self}")
122 return info
124 def remove(self) -> None:
125 """Remove the resource."""
126 os.remove(self.ospath)
128 @contextlib.contextmanager
129 def _as_local(
130 self, multithreaded: bool = True, tmpdir: ResourcePath | None = None
131 ) -> Generator[ResourcePath]:
132 """Return the local path of the file.
134 This is an internal helper for ``as_local()``.
136 Parameters
137 ----------
138 multithreaded : `bool`, optional
139 Unused.
140 tmpdir : `ResourcePath` or `None`, optional
141 Unused.
143 Returns
144 -------
145 local_uri : `ResourcePath`
146 A local URI. In this case it will be itself.
147 """
148 yield self
150 def read(self, size: int = -1) -> bytes:
151 with open(self.ospath, "rb") as fh:
152 return fh.read(size)
154 def write(self, data: bytes, overwrite: bool = True) -> None:
155 dir = os.path.dirname(self.ospath)
156 if dir and not os.path.exists(dir):
157 _create_directories(dir)
158 mode = "wb" if overwrite else "xb"
159 with open(self.ospath, mode) as f:
160 f.write(data)
162 def mkdir(self) -> None:
163 """Make the directory associated with this URI.
165 An attempt will be made to create the directory even if the URI
166 looks like a file.
168 Raises
169 ------
170 NotADirectoryError:
171 Raised if a non-directory already exists.
172 """
173 try:
174 _create_directories(self.ospath)
175 except FileExistsError:
176 raise NotADirectoryError(f"{self.ospath} exists but is not a directory.") from None
178 def isdir(self) -> bool:
179 """Return whether this URI is a directory.
181 Returns
182 -------
183 isdir : `bool`
184 `True` if this URI is a directory or looks like a directory,
185 else `False`.
186 """
187 if self.dirLike is None:
188 # Cache state for next time.
189 self.dirLike = os.path.isdir(self.ospath)
190 return self.dirLike
192 def transfer_from(
193 self,
194 src: ResourcePath,
195 transfer: str,
196 overwrite: bool = False,
197 transaction: TransactionProtocol | None = None,
198 multithreaded: bool = True,
199 ) -> None:
200 """Transfer the current resource to a local file.
202 Parameters
203 ----------
204 src : `ResourcePath`
205 Source URI.
206 transfer : `str`
207 Mode to use for transferring the resource. Supports the following
208 options: copy, link, symlink, hardlink, relsymlink.
209 overwrite : `bool`, optional
210 Allow an existing file to be overwritten. Defaults to `False`.
211 transaction : `~lsst.resources.utils.TransactionProtocol`, optional
212 If a transaction is provided, undo actions will be registered.
213 multithreaded : `bool`, optional
214 Whether threads are allowed to be used or not.
215 """
216 # Fail early to prevent delays if remote resources are requested
217 if transfer not in self.transferModes:
218 raise ValueError(f"Transfer mode '{transfer}' not supported by URI scheme {self.scheme}")
220 # Existence checks can take time so only try if the log message
221 # will be issued.
222 if log.isEnabledFor(logging.DEBUG): 222 ↛ 235line 222 didn't jump to line 235 because the condition on line 222 was always true
223 log.debug(
224 "Transferring %s [exists: %s] -> %s [exists: %s] (transfer=%s)",
225 src,
226 src.exists(),
227 self,
228 self.exists(),
229 transfer,
230 )
232 # Short circuit if the URIs are identical. The inode comparison below
233 # only runs for a local source, so a non-local source that happens to
234 # name this same resource would otherwise be reported as a clash.
235 if self == src:
236 log.debug(
237 "Target and destination URIs are identical: %s, returning immediately."
238 " No further action required.",
239 self,
240 )
241 return
243 # The output location should not exist unless overwrite=True.
244 # Rather than use `exists()`, use os.stat since we might need
245 # the full answer later.
246 dest_stat: os.stat_result | None
247 try:
248 # Do not read through links of the file itself.
249 dest_stat = os.lstat(self.ospath)
250 except FileNotFoundError:
251 dest_stat = None
253 # It is possible that the source URI and target URI refer
254 # to the same file. This can happen for a number of reasons
255 # (such as soft links in the path, or they really are the same).
256 # In that case log a message and return as if the transfer
257 # completed (it technically did). A temporary file download
258 # can't be the same so the test can be skipped.
259 if dest_stat and src.isLocal and not src.isTemporary:
260 # Be consistent and use lstat here (even though realpath
261 # has been called). It does not harm.
262 local_src_stat = os.lstat(src.ospath)
263 if dest_stat.st_ino == local_src_stat.st_ino and dest_stat.st_dev == local_src_stat.st_dev:
264 log.debug(
265 "Destination URI %s is the same file as source URI %s, returning immediately."
266 " No further action required.",
267 self,
268 src,
269 )
270 return
272 if not overwrite and dest_stat:
273 raise FileExistsError(
274 f"Destination path '{self}' already exists. Transfer from {src} cannot be completed."
275 )
277 # Make the destination path absolute (but don't follow links since
278 # that would possibly cause us to end up in the wrong place if the
279 # file existed already as a soft link)
280 newFullPath = os.path.abspath(self.ospath)
281 outputDir = os.path.dirname(newFullPath)
283 # We do not have to special case FileResourcePath here because
284 # as_local handles that. If remote download, download it to the
285 # destination directory to allow an atomic rename but only if that
286 # directory exists because we do not want to create a directory
287 # but then end up with the download failing.
288 tmpdir = outputDir if os.path.exists(outputDir) else None
289 with src.as_local(multithreaded=multithreaded, tmpdir=tmpdir) as local_uri:
290 is_temporary = local_uri.isTemporary
291 local_src = local_uri.ospath
293 # Short circuit if the URIs are identical immediately.
294 if self == local_uri: 294 ↛ 295line 294 didn't jump to line 295 because the condition on line 294 was never true
295 log.debug(
296 "Target and destination URIs are identical: %s, returning immediately."
297 " No further action required.",
298 self,
299 )
300 return
302 # Default transfer mode depends on whether we have a temporary
303 # file or not.
304 if transfer == "auto":
305 transfer = self.transferDefault if not is_temporary else "copy"
307 if not os.path.exists(local_src):
308 if is_temporary:
309 if src == local_uri: 309 ↛ 313line 309 didn't jump to line 313 because the condition on line 309 was always true
310 msg = f"Local temporary file {src} has gone missing."
311 else:
312 # This will not happen in normal scenarios.
313 msg = f"Local file {local_uri} downloaded from {src} has gone missing"
314 else:
315 msg = f"Source URI {src} does not exist"
316 raise FileNotFoundError(msg)
318 # Follow soft links
319 local_src = os.path.realpath(os.path.normpath(local_src))
321 # Creating a symlink to a local copy of a remote resource
322 # should never work. Creating a hardlink will work but should
323 # not be allowed since it is highly unlikely that this is ever
324 # an intended option and depends on the local target being
325 # on the same file system as was used for the temporary file
326 # download.
327 # If a symlink is being requested for a local temporary file
328 # that is likely undesirable but should not be refused.
329 if is_temporary and src != local_uri and "link" in transfer:
330 raise RuntimeError(
331 f"Can not use local file system transfer mode {transfer} for remote resource ({src})"
332 )
333 elif is_temporary and src == local_uri and "symlink" in transfer:
334 log.debug(
335 "Using a symlink for a temporary resource may lead to unexpected downstream failures."
336 )
338 # For temporary files we can own them if we created it.
339 requested_transfer = transfer
340 if src != local_uri and is_temporary and transfer == "copy":
341 transfer = "move"
343 if not os.path.isdir(outputDir):
344 # Must create the directory -- this can not be rolled back
345 # since another transfer running concurrently may
346 # be relying on this existing.
347 _create_directories(outputDir)
349 if transaction is None:
350 # Use a no-op transaction to reduce code duplication
351 transaction = NoTransaction()
353 # For links the OS doesn't let us overwrite so if something does
354 # exist we have to remove it before we do the actual "transfer"
355 # below
356 if "link" in transfer and overwrite and dest_stat:
357 with contextlib.suppress(Exception):
358 # If this fails we ignore it since it's a problem
359 # that will manifest immediately below with a more relevant
360 # error message
361 self.remove()
363 if transfer == "move":
364 # If a rename works we try that since that is guaranteed to
365 # be atomic. If that fails we copy and rename. We do this
366 # in case other processes are trying to move to the same
367 # file and we want the "winner" to not be corrupted.
368 try:
369 with transaction.undoWith(f"move from {local_src}", os.rename, newFullPath, local_src):
370 os.rename(local_src, newFullPath)
371 except OSError:
372 with self.temporary_uri(prefix=self.parent(), suffix=self.getExtension()) as temp_copy:
373 shutil.copy(local_src, temp_copy.ospath)
374 with transaction.undoWith(
375 f"move from {local_src}",
376 shutil.move,
377 newFullPath,
378 local_src,
379 copy_function=shutil.copy,
380 ):
381 os.rename(temp_copy.ospath, newFullPath)
382 os.remove(local_src)
383 elif transfer == "copy":
384 # We want atomic copies so first copy to a temp location in
385 # the same output directory. This at least guarantees that
386 # if multiple processes are writing to the same file
387 # simultaneously the file we end up with will not be corrupt.
388 if overwrite:
389 with self.temporary_uri(prefix=self.parent(), suffix=self.getExtension()) as temp_copy:
390 shutil.copy(local_src, temp_copy.ospath)
391 with transaction.undoWith(f"copy from {local_src}", os.remove, newFullPath):
392 os.rename(temp_copy.ospath, newFullPath)
393 else:
394 # Create the file exclusively to ensure that no others are
395 # trying to write.
396 temp_path = newFullPath + ".transfer-tmp"
397 try:
398 with open(temp_path, "x"):
399 pass
400 except FileExistsError:
401 raise FileExistsError(
402 f"Another process is writing to '{self}'."
403 f" Transfer from {src} cannot be completed."
404 )
405 with transaction.undoWith(f"copy from {local_src}", os.remove, temp_path):
406 # Make sure file is writable, no matter the umask.
407 st = os.stat(temp_path)
408 os.chmod(temp_path, st.st_mode | stat.S_IWUSR)
409 shutil.copy(local_src, temp_path)
410 # Use link/remove to atomically and exclusively move the
411 # file into place (only one concurrent linker can win).
412 try:
413 os.link(temp_path, newFullPath)
414 except FileExistsError:
415 raise FileExistsError(
416 f"Another process wrote to '{self}'. Transfer from {src} cannot be completed."
417 )
418 finally:
419 os.remove(temp_path)
420 elif transfer == "link":
421 # Try hard link and if that fails use a symlink
422 with transaction.undoWith(f"link to {local_src}", os.remove, newFullPath):
423 try:
424 os.link(local_src, newFullPath)
425 except OSError:
426 # Read through existing symlinks
427 os.symlink(local_src, newFullPath)
428 elif transfer == "hardlink":
429 with transaction.undoWith(f"hardlink to {local_src}", os.remove, newFullPath):
430 os.link(local_src, newFullPath)
431 elif transfer == "symlink":
432 # Read through existing symlinks
433 with transaction.undoWith(f"symlink to {local_src}", os.remove, newFullPath):
434 os.symlink(local_src, newFullPath)
435 elif transfer == "relsymlink":
436 # This is a standard symlink but using a relative path
437 # Need the directory name to give to relative root
438 # A full file path confuses it into an extra ../
439 newFullPathRoot = os.path.dirname(newFullPath)
440 relPath = os.path.relpath(local_src, newFullPathRoot)
441 with transaction.undoWith(f"relsymlink to {local_src}", os.remove, newFullPath):
442 os.symlink(relPath, newFullPath)
443 else:
444 raise NotImplementedError(f"Transfer type '{transfer}' not supported.")
446 # This was an explicit move requested from a remote resource
447 # try to remove that remote resource. We check is_temporary because
448 # the local file would have been moved by shutil.move already.
449 if requested_transfer == "move" and is_temporary and src != local_uri:
450 # Transactions do not work here
451 src.remove()
453 def walk(
454 self, file_filter: str | re.Pattern | None = None
455 ) -> Iterator[list | tuple[ResourcePath, list[str], list[str]]]:
456 """Walk the directory tree returning matching files and directories.
458 Parameters
459 ----------
460 file_filter : `str` or `re.Pattern`, optional
461 Regex to filter out files from the list before it is returned.
463 Yields
464 ------
465 dirpath : `ResourcePath`
466 Current directory being examined.
467 dirnames : `list` of `str`
468 Names of subdirectories within dirpath.
469 filenames : `list` of `str`
470 Names of all the files within dirpath.
471 """
472 if not self.isdir():
473 raise ValueError("Can not walk a non-directory URI")
475 if isinstance(file_filter, str):
476 file_filter = re.compile(file_filter)
478 for root, dirs, files in os.walk(self.ospath, followlinks=True):
479 # Filter by the regex
480 if file_filter is not None:
481 files = [f for f in files if file_filter.search(f)]
482 # Rebuild from the parsed URI rather than from the OS path, so
483 # that a subclass keeps its own scheme and netloc. Constructing
484 # from a plain path always resolves to a file URI.
485 path = os2posix(root)
486 if self.quotePaths: 486 ↛ 488line 486 didn't jump to line 488 because the condition on line 486 was always true
487 path = urllib.parse.quote(path)
488 yield self.replace(path=path, forceDirectory=True), dirs, files
490 @classmethod
491 def _fixupPathUri(
492 cls,
493 parsed: urllib.parse.ParseResult,
494 root: ResourcePath | None = None,
495 forceAbsolute: bool = False,
496 forceDirectory: bool | None = None,
497 ) -> tuple[urllib.parse.ParseResult, bool | None]:
498 """Fix up relative paths in URI instances.
500 Parameters
501 ----------
502 parsed : `~urllib.parse.ParseResult`
503 The result from parsing a URI using `urllib.parse`.
504 root : `ResourcePath`, optional
505 Path to use as root when converting relative to absolute.
506 If `None`, it will be the current working directory. It is only
507 used if a file-scheme is used incorrectly with a relative path.
508 forceAbsolute : `bool`, ignored
509 Has no effect for this subclass. ``file`` URIs are always
510 absolute.
511 forceDirectory : `bool`, optional
512 If `True` forces the URI to end with a separator, otherwise given
513 URI is interpreted as is.
515 Returns
516 -------
517 modified : `~urllib.parse.ParseResult`
518 Update result if a URI is being handled.
519 dirLike : `bool` or `None`
520 `True` if given parsed URI has a trailing separator or
521 ``forceDirectory`` is `True`. Otherwise can return the given
522 value of ``forceDirectory``.
524 Notes
525 -----
526 Relative paths are explicitly not supported by RFC8089 but `urllib`
527 does accept URIs of the form ``file:relative/path.ext``. They need
528 to be turned into absolute paths before they can be used. This is
529 always done regardless of the ``forceAbsolute`` parameter.
530 """
531 # assume we are not dealing with a directory like URI
532 dirLike = forceDirectory
534 # file URI implies POSIX path separators so split as POSIX,
535 # then join as os, and convert to abspath. Do not handle
536 # home directories since "file" scheme is explicitly documented
537 # to not do tilde expansion.
538 sep = posixpath.sep
540 # Consistency check.
541 if forceDirectory is False and parsed.path.endswith(sep):
542 raise ValueError(
543 f"URI {parsed.geturl()} ends with {sep} but "
544 "forceDirectory parameter declares it to be a file."
545 )
547 # For an absolute path all we need to do is check if we need
548 # to force the directory separator
549 if posixpath.isabs(parsed.path):
550 if forceDirectory:
551 if not parsed.path.endswith(sep):
552 parsed = parsed._replace(path=parsed.path + sep)
553 dirLike = True
554 return copy.copy(parsed), dirLike
556 # Relative path so must fix it to be compliant with the standard
558 # Replacement values for the URI
559 replacements = {}
561 if root is None:
562 root_str = os.path.abspath(os.path.curdir)
563 else:
564 if root.scheme and root.scheme != "file": 564 ↛ 565line 564 didn't jump to line 565 because the condition on line 564 was never true
565 raise RuntimeError(f"The override root must be a file URI not {root.scheme}")
566 root_str = os.path.abspath(root.ospath)
568 replacements["path"] = posixpath.normpath(posixpath.join(os2posix(root_str), parsed.path))
570 # normpath strips trailing "/" so put it back if necessary
571 # Acknowledge that trailing separator exists.
572 if forceDirectory or (parsed.path.endswith(sep) and not replacements["path"].endswith(sep)):
573 replacements["path"] += sep
574 dirLike = True
576 # ParseResult is a NamedTuple so _replace is standard API
577 parsed = parsed._replace(**replacements)
579 if parsed.params or parsed.query: 579 ↛ 580line 579 didn't jump to line 580 because the condition on line 579 was never true
580 log.warning("Additional items unexpectedly encountered in file URI: %s", parsed.geturl())
582 return parsed, dirLike
584 @contextlib.contextmanager
585 def _openImpl(
586 self,
587 mode: str = "r",
588 *,
589 encoding: str | None = None,
590 ) -> Generator[ResourceHandleProtocol]:
591 with FileResourceHandle(mode=mode, log=log, uri=self, encoding=encoding) as buffer:
592 yield buffer
594 def to_fsspec(self) -> tuple[AbstractFileSystem, str]:
595 """Return an abstract file system and path that can be used by fsspec.
597 Returns
598 -------
599 fs : `fsspec.spec.AbstractFileSystem`
600 A file system object suitable for use with the returned path.
601 path : `str`
602 A path that can be opened by the file system object.
603 """
604 if fsspec is None: 604 ↛ 605line 604 didn't jump to line 605 because the condition on line 604 was never true
605 raise ImportError("fsspec is not available")
606 # fsspec does not like URL encodings in file URIs so pass it the os
607 # path instead.
608 return fsspec.url_to_fs(self.ospath)
611def _create_directories(name: str | bytes) -> None:
612 """Create a directory and all of its parent directories that don't yet
613 exist.
615 Parameters
616 ----------
617 name : `str` or `bytes`
618 Path to the directory to be created
620 Notes
621 -----
622 The code in this function is duplicated from the Python standard library
623 function os.makedirs with one change: if the user has set a process umask
624 that prevents us from creating/accessing files in the newly created
625 directories, the permissions of the directories are altered to allow
626 owner-write and owner-traverse so that they can be used.
627 """
628 # These are optional parameters in the original function, but they can be
629 # constant here.
630 mode = 0o777
631 exist_ok = True
633 head, tail = os.path.split(name)
634 if not tail:
635 head, tail = os.path.split(head)
636 if head and tail and not os.path.exists(head):
637 try:
638 _create_directories(head)
639 except FileExistsError:
640 # Defeats race condition when another thread created the path
641 pass
642 cdir: str | bytes = os.curdir
643 if isinstance(tail, bytes): 643 ↛ 644line 643 didn't jump to line 644 because the condition on line 643 was never true
644 cdir = bytes(os.curdir, "ASCII")
645 if tail == cdir: # xxx/newdir/. exists if xxx/newdir exists 645 ↛ 646line 645 didn't jump to line 646 because the condition on line 645 was never true
646 return
647 try:
648 os.mkdir(name, mode)
649 # This is the portion that is modified relative to the standard library
650 # version of the function.
651 ensure_directory_is_writeable(name)
652 # end modified portion
653 except OSError:
654 # Cannot rely on checking for EEXIST, since the operating system
655 # could give priority to other errors like EACCES or EROFS
656 if not exist_ok or not os.path.isdir(name):
657 raise