Coverage for python/lsst/resources/file.py: 89%
255 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-22 09:18 +0000
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 # A transfer costs orders of magnitude more than an existence check and
95 # scales with the file size, so batches stay small enough to balance.
96 _transfer_chunk_size = 25
98 @property
99 def ospath(self) -> str:
100 """Path component of the URI localized to current OS.
102 Will unquote URI path since a formal URI must include the quoting.
103 """
104 return urllib.parse.unquote(posix2os(self._uri.path))
106 def exists(self) -> bool:
107 """Indicate that the file exists."""
108 # Uses os.path.exists so if there is a soft link that points
109 # to a file that no longer exists this will return False
110 return os.path.exists(self.ospath)
112 def size(self) -> int:
113 """Return the size of the file in bytes."""
114 if not os.path.isdir(self.ospath):
115 stat = os.stat(self.ospath)
116 sz = stat.st_size
117 else:
118 sz = 0
119 return sz
121 def get_info(self) -> ResourceInfo:
122 """Return lightweight metadata about this file."""
123 info = _path_to_info(str(self), self.ospath)
124 if info is None: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true
125 raise RuntimeError(f"Unexpected internal failure obtaining file info for {self}")
126 return info
128 def remove(self) -> None:
129 """Remove the resource."""
130 os.remove(self.ospath)
132 @contextlib.contextmanager
133 def _as_local(
134 self, multithreaded: bool = True, tmpdir: ResourcePath | None = None
135 ) -> Generator[ResourcePath]:
136 """Return the local path of the file.
138 This is an internal helper for ``as_local()``.
140 Parameters
141 ----------
142 multithreaded : `bool`, optional
143 Unused.
144 tmpdir : `ResourcePath` or `None`, optional
145 Unused.
147 Returns
148 -------
149 local_uri : `ResourcePath`
150 A local URI. In this case it will be itself.
151 """
152 yield self
154 def read(self, size: int = -1) -> bytes:
155 with open(self.ospath, "rb") as fh:
156 return fh.read(size)
158 def write(self, data: bytes, overwrite: bool = True) -> None:
159 dir = os.path.dirname(self.ospath)
160 if dir and not os.path.exists(dir):
161 _create_directories(dir)
162 mode = "wb" if overwrite else "xb"
163 with open(self.ospath, mode) as f:
164 f.write(data)
166 def mkdir(self) -> None:
167 """Make the directory associated with this URI.
169 An attempt will be made to create the directory even if the URI
170 looks like a file.
172 Raises
173 ------
174 NotADirectoryError:
175 Raised if a non-directory already exists.
176 """
177 try:
178 _create_directories(self.ospath)
179 except FileExistsError:
180 raise NotADirectoryError(f"{self.ospath} exists but is not a directory.") from None
182 def isdir(self) -> bool:
183 """Return whether this URI is a directory.
185 Returns
186 -------
187 isdir : `bool`
188 `True` if this URI is a directory or looks like a directory,
189 else `False`.
190 """
191 if self.dirLike is None:
192 # Cache state for next time.
193 self.dirLike = os.path.isdir(self.ospath)
194 return self.dirLike
196 def transfer_from(
197 self,
198 src: ResourcePath,
199 transfer: str,
200 overwrite: bool = False,
201 transaction: TransactionProtocol | None = None,
202 multithreaded: bool = True,
203 ) -> None:
204 """Transfer the current resource to a local file.
206 Parameters
207 ----------
208 src : `ResourcePath`
209 Source URI.
210 transfer : `str`
211 Mode to use for transferring the resource. Supports the following
212 options: copy, link, symlink, hardlink, relsymlink.
213 overwrite : `bool`, optional
214 Allow an existing file to be overwritten. Defaults to `False`.
215 transaction : `~lsst.resources.utils.TransactionProtocol`, optional
216 If a transaction is provided, undo actions will be registered.
217 multithreaded : `bool`, optional
218 Whether threads are allowed to be used or not.
219 """
220 # Fail early to prevent delays if remote resources are requested
221 if transfer not in self.transferModes:
222 raise ValueError(f"Transfer mode '{transfer}' not supported by URI scheme {self.scheme}")
224 # Existence checks can take time so only try if the log message
225 # will be issued.
226 if log.isEnabledFor(logging.DEBUG): 226 ↛ 239line 226 didn't jump to line 239 because the condition on line 226 was always true
227 log.debug(
228 "Transferring %s [exists: %s] -> %s [exists: %s] (transfer=%s)",
229 src,
230 src.exists(),
231 self,
232 self.exists(),
233 transfer,
234 )
236 # Short circuit if the URIs are identical. The inode comparison below
237 # only runs for a local source, so a non-local source that happens to
238 # name this same resource would otherwise be reported as a clash.
239 if self == src:
240 log.debug(
241 "Target and destination URIs are identical: %s, returning immediately."
242 " No further action required.",
243 self,
244 )
245 return
247 # The output location should not exist unless overwrite=True.
248 # Rather than use `exists()`, use os.stat since we might need
249 # the full answer later.
250 dest_stat: os.stat_result | None
251 try:
252 # Do not read through links of the file itself.
253 dest_stat = os.lstat(self.ospath)
254 except FileNotFoundError:
255 dest_stat = None
257 # It is possible that the source URI and target URI refer
258 # to the same file. This can happen for a number of reasons
259 # (such as soft links in the path, or they really are the same).
260 # In that case log a message and return as if the transfer
261 # completed (it technically did). A temporary file download
262 # can't be the same so the test can be skipped.
263 if dest_stat and src.isLocal and not src.isTemporary:
264 # Be consistent and use lstat here (even though realpath
265 # has been called). It does not harm.
266 local_src_stat = os.lstat(src.ospath)
267 if dest_stat.st_ino == local_src_stat.st_ino and dest_stat.st_dev == local_src_stat.st_dev:
268 log.debug(
269 "Destination URI %s is the same file as source URI %s, returning immediately."
270 " No further action required.",
271 self,
272 src,
273 )
274 return
276 if not overwrite and dest_stat:
277 raise FileExistsError(
278 f"Destination path '{self}' already exists. Transfer from {src} cannot be completed."
279 )
281 # Make the destination path absolute (but don't follow links since
282 # that would possibly cause us to end up in the wrong place if the
283 # file existed already as a soft link)
284 newFullPath = os.path.abspath(self.ospath)
285 outputDir = os.path.dirname(newFullPath)
287 # We do not have to special case FileResourcePath here because
288 # as_local handles that. If remote download, download it to the
289 # destination directory to allow an atomic rename but only if that
290 # directory exists because we do not want to create a directory
291 # but then end up with the download failing.
292 tmpdir = outputDir if os.path.exists(outputDir) else None
293 with src.as_local(multithreaded=multithreaded, tmpdir=tmpdir) as local_uri:
294 is_temporary = local_uri.isTemporary
295 local_src = local_uri.ospath
297 # Short circuit if the URIs are identical immediately.
298 if self == local_uri: 298 ↛ 299line 298 didn't jump to line 299 because the condition on line 298 was never true
299 log.debug(
300 "Target and destination URIs are identical: %s, returning immediately."
301 " No further action required.",
302 self,
303 )
304 return
306 # Default transfer mode depends on whether we have a temporary
307 # file or not.
308 if transfer == "auto":
309 transfer = self.transferDefault if not is_temporary else "copy"
311 if not os.path.exists(local_src):
312 if is_temporary:
313 if src == local_uri: 313 ↛ 317line 313 didn't jump to line 317 because the condition on line 313 was always true
314 msg = f"Local temporary file {src} has gone missing."
315 else:
316 # This will not happen in normal scenarios.
317 msg = f"Local file {local_uri} downloaded from {src} has gone missing"
318 else:
319 msg = f"Source URI {src} does not exist"
320 raise FileNotFoundError(msg)
322 # Follow soft links
323 local_src = os.path.realpath(os.path.normpath(local_src))
325 # Creating a symlink to a local copy of a remote resource
326 # should never work. Creating a hardlink will work but should
327 # not be allowed since it is highly unlikely that this is ever
328 # an intended option and depends on the local target being
329 # on the same file system as was used for the temporary file
330 # download.
331 # If a symlink is being requested for a local temporary file
332 # that is likely undesirable but should not be refused.
333 if is_temporary and src != local_uri and "link" in transfer:
334 raise RuntimeError(
335 f"Can not use local file system transfer mode {transfer} for remote resource ({src})"
336 )
337 elif is_temporary and src == local_uri and "symlink" in transfer:
338 log.debug(
339 "Using a symlink for a temporary resource may lead to unexpected downstream failures."
340 )
342 # For temporary files we can own them if we created it.
343 requested_transfer = transfer
344 if src != local_uri and is_temporary and transfer == "copy":
345 transfer = "move"
347 if not os.path.isdir(outputDir):
348 # Must create the directory -- this can not be rolled back
349 # since another transfer running concurrently may
350 # be relying on this existing.
351 _create_directories(outputDir)
353 if transaction is None:
354 # Use a no-op transaction to reduce code duplication
355 transaction = NoTransaction()
357 # For links the OS doesn't let us overwrite so if something does
358 # exist we have to remove it before we do the actual "transfer"
359 # below
360 if "link" in transfer and overwrite and dest_stat:
361 with contextlib.suppress(Exception):
362 # If this fails we ignore it since it's a problem
363 # that will manifest immediately below with a more relevant
364 # error message
365 self.remove()
367 if transfer == "move":
368 # If a rename works we try that since that is guaranteed to
369 # be atomic. If that fails we copy and rename. We do this
370 # in case other processes are trying to move to the same
371 # file and we want the "winner" to not be corrupted.
372 try:
373 with transaction.undoWith(f"move from {local_src}", os.rename, newFullPath, local_src):
374 os.rename(local_src, newFullPath)
375 except OSError:
376 with self.temporary_uri(prefix=self.parent(), suffix=self.getExtension()) as temp_copy:
377 shutil.copy(local_src, temp_copy.ospath)
378 with transaction.undoWith(
379 f"move from {local_src}",
380 shutil.move,
381 newFullPath,
382 local_src,
383 copy_function=shutil.copy,
384 ):
385 os.rename(temp_copy.ospath, newFullPath)
386 os.remove(local_src)
387 elif transfer == "copy":
388 # We want atomic copies so first copy to a temp location in
389 # the same output directory. This at least guarantees that
390 # if multiple processes are writing to the same file
391 # simultaneously the file we end up with will not be corrupt.
392 if overwrite:
393 with self.temporary_uri(prefix=self.parent(), suffix=self.getExtension()) as temp_copy:
394 shutil.copy(local_src, temp_copy.ospath)
395 with transaction.undoWith(f"copy from {local_src}", os.remove, newFullPath):
396 os.rename(temp_copy.ospath, newFullPath)
397 else:
398 # Create the file exclusively to ensure that no others are
399 # trying to write.
400 temp_path = newFullPath + ".transfer-tmp"
401 try:
402 with open(temp_path, "x"):
403 pass
404 except FileExistsError:
405 raise FileExistsError(
406 f"Another process is writing to '{self}'."
407 f" Transfer from {src} cannot be completed."
408 )
409 with transaction.undoWith(f"copy from {local_src}", os.remove, temp_path):
410 # Make sure file is writable, no matter the umask.
411 st = os.stat(temp_path)
412 os.chmod(temp_path, st.st_mode | stat.S_IWUSR)
413 shutil.copy(local_src, temp_path)
414 # Use link/remove to atomically and exclusively move the
415 # file into place (only one concurrent linker can win).
416 try:
417 os.link(temp_path, newFullPath)
418 except FileExistsError:
419 raise FileExistsError(
420 f"Another process wrote to '{self}'. Transfer from {src} cannot be completed."
421 )
422 finally:
423 os.remove(temp_path)
424 elif transfer == "link":
425 # Try hard link and if that fails use a symlink
426 with transaction.undoWith(f"link to {local_src}", os.remove, newFullPath):
427 try:
428 os.link(local_src, newFullPath)
429 except OSError:
430 # Read through existing symlinks
431 os.symlink(local_src, newFullPath)
432 elif transfer == "hardlink":
433 with transaction.undoWith(f"hardlink to {local_src}", os.remove, newFullPath):
434 os.link(local_src, newFullPath)
435 elif transfer == "symlink":
436 # Read through existing symlinks
437 with transaction.undoWith(f"symlink to {local_src}", os.remove, newFullPath):
438 os.symlink(local_src, newFullPath)
439 elif transfer == "relsymlink":
440 # This is a standard symlink but using a relative path
441 # Need the directory name to give to relative root
442 # A full file path confuses it into an extra ../
443 newFullPathRoot = os.path.dirname(newFullPath)
444 relPath = os.path.relpath(local_src, newFullPathRoot)
445 with transaction.undoWith(f"relsymlink to {local_src}", os.remove, newFullPath):
446 os.symlink(relPath, newFullPath)
447 else:
448 raise NotImplementedError(f"Transfer type '{transfer}' not supported.")
450 # This was an explicit move requested from a remote resource
451 # try to remove that remote resource. We check is_temporary because
452 # the local file would have been moved by shutil.move already.
453 if requested_transfer == "move" and is_temporary and src != local_uri:
454 # Transactions do not work here
455 src.remove()
457 def walk(
458 self, file_filter: str | re.Pattern | None = None
459 ) -> Iterator[list | tuple[ResourcePath, list[str], list[str]]]:
460 """Walk the directory tree returning matching files and directories.
462 Parameters
463 ----------
464 file_filter : `str` or `re.Pattern`, optional
465 Regex to filter out files from the list before it is returned.
467 Yields
468 ------
469 dirpath : `ResourcePath`
470 Current directory being examined.
471 dirnames : `list` of `str`
472 Names of subdirectories within dirpath.
473 filenames : `list` of `str`
474 Names of all the files within dirpath.
475 """
476 if not self.isdir():
477 raise ValueError("Can not walk a non-directory URI")
479 if isinstance(file_filter, str):
480 file_filter = re.compile(file_filter)
482 for root, dirs, files in os.walk(self.ospath, followlinks=True):
483 # Filter by the regex
484 if file_filter is not None:
485 files = [f for f in files if file_filter.search(f)]
486 # Rebuild from the parsed URI rather than from the OS path, so
487 # that a subclass keeps its own scheme and netloc. Constructing
488 # from a plain path always resolves to a file URI.
489 path = os2posix(root)
490 if self.quotePaths: 490 ↛ 492line 490 didn't jump to line 492 because the condition on line 490 was always true
491 path = urllib.parse.quote(path)
492 yield self.replace(path=path, forceDirectory=True), dirs, files
494 @classmethod
495 def _fixupPathUri(
496 cls,
497 parsed: urllib.parse.ParseResult,
498 root: ResourcePath | None = None,
499 forceAbsolute: bool = False,
500 forceDirectory: bool | None = None,
501 ) -> tuple[urllib.parse.ParseResult, bool | None]:
502 """Fix up relative paths in URI instances.
504 Parameters
505 ----------
506 parsed : `~urllib.parse.ParseResult`
507 The result from parsing a URI using `urllib.parse`.
508 root : `ResourcePath`, optional
509 Path to use as root when converting relative to absolute.
510 If `None`, it will be the current working directory. It is only
511 used if a file-scheme is used incorrectly with a relative path.
512 forceAbsolute : `bool`, ignored
513 Has no effect for this subclass. ``file`` URIs are always
514 absolute.
515 forceDirectory : `bool`, optional
516 If `True` forces the URI to end with a separator, otherwise given
517 URI is interpreted as is.
519 Returns
520 -------
521 modified : `~urllib.parse.ParseResult`
522 Update result if a URI is being handled.
523 dirLike : `bool` or `None`
524 `True` if given parsed URI has a trailing separator or
525 ``forceDirectory`` is `True`. Otherwise can return the given
526 value of ``forceDirectory``.
528 Notes
529 -----
530 Relative paths are explicitly not supported by RFC8089 but `urllib`
531 does accept URIs of the form ``file:relative/path.ext``. They need
532 to be turned into absolute paths before they can be used. This is
533 always done regardless of the ``forceAbsolute`` parameter.
534 """
535 # assume we are not dealing with a directory like URI
536 dirLike = forceDirectory
538 # file URI implies POSIX path separators so split as POSIX,
539 # then join as os, and convert to abspath. Do not handle
540 # home directories since "file" scheme is explicitly documented
541 # to not do tilde expansion.
542 sep = posixpath.sep
544 # Consistency check.
545 if forceDirectory is False and parsed.path.endswith(sep):
546 raise ValueError(
547 f"URI {parsed.geturl()} ends with {sep} but "
548 "forceDirectory parameter declares it to be a file."
549 )
551 # For an absolute path all we need to do is check if we need
552 # to force the directory separator
553 if posixpath.isabs(parsed.path):
554 if forceDirectory:
555 if not parsed.path.endswith(sep):
556 parsed = parsed._replace(path=parsed.path + sep)
557 dirLike = True
558 return copy.copy(parsed), dirLike
560 # Relative path so must fix it to be compliant with the standard
562 # Replacement values for the URI
563 replacements = {}
565 if root is None:
566 root_str = os.path.abspath(os.path.curdir)
567 else:
568 if root.scheme and root.scheme != "file": 568 ↛ 569line 568 didn't jump to line 569 because the condition on line 568 was never true
569 raise RuntimeError(f"The override root must be a file URI not {root.scheme}")
570 root_str = os.path.abspath(root.ospath)
572 replacements["path"] = posixpath.normpath(posixpath.join(os2posix(root_str), parsed.path))
574 # normpath strips trailing "/" so put it back if necessary
575 # Acknowledge that trailing separator exists.
576 if forceDirectory or (parsed.path.endswith(sep) and not replacements["path"].endswith(sep)):
577 replacements["path"] += sep
578 dirLike = True
580 # ParseResult is a NamedTuple so _replace is standard API
581 parsed = parsed._replace(**replacements)
583 if parsed.params or parsed.query: 583 ↛ 584line 583 didn't jump to line 584 because the condition on line 583 was never true
584 log.warning("Additional items unexpectedly encountered in file URI: %s", parsed.geturl())
586 return parsed, dirLike
588 @contextlib.contextmanager
589 def _openImpl(
590 self,
591 mode: str = "r",
592 *,
593 encoding: str | None = None,
594 ) -> Generator[ResourceHandleProtocol]:
595 with FileResourceHandle(mode=mode, log=log, uri=self, encoding=encoding) as buffer:
596 yield buffer
598 def to_fsspec(self) -> tuple[AbstractFileSystem, str]:
599 """Return an abstract file system and path that can be used by fsspec.
601 Returns
602 -------
603 fs : `fsspec.spec.AbstractFileSystem`
604 A file system object suitable for use with the returned path.
605 path : `str`
606 A path that can be opened by the file system object.
607 """
608 if fsspec is None: 608 ↛ 609line 608 didn't jump to line 609 because the condition on line 608 was never true
609 raise ImportError("fsspec is not available")
610 # fsspec does not like URL encodings in file URIs so pass it the os
611 # path instead.
612 return fsspec.url_to_fs(self.ospath)
615def _create_directories(name: str | bytes) -> None:
616 """Create a directory and all of its parent directories that don't yet
617 exist.
619 Parameters
620 ----------
621 name : `str` or `bytes`
622 Path to the directory to be created
624 Notes
625 -----
626 The code in this function is duplicated from the Python standard library
627 function os.makedirs with one change: if the user has set a process umask
628 that prevents us from creating/accessing files in the newly created
629 directories, the permissions of the directories are altered to allow
630 owner-write and owner-traverse so that they can be used.
631 """
632 # These are optional parameters in the original function, but they can be
633 # constant here.
634 mode = 0o777
635 exist_ok = True
637 head, tail = os.path.split(name)
638 if not tail:
639 head, tail = os.path.split(head)
640 if head and tail and not os.path.exists(head):
641 try:
642 _create_directories(head)
643 except FileExistsError:
644 # Defeats race condition when another thread created the path
645 pass
646 cdir: str | bytes = os.curdir
647 if isinstance(tail, bytes): 647 ↛ 648line 647 didn't jump to line 648 because the condition on line 647 was never true
648 cdir = bytes(os.curdir, "ASCII")
649 if tail == cdir: # xxx/newdir/. exists if xxx/newdir exists 649 ↛ 650line 649 didn't jump to line 650 because the condition on line 649 was never true
650 return
651 try:
652 os.mkdir(name, mode)
653 # This is the portion that is modified relative to the standard library
654 # version of the function.
655 ensure_directory_is_writeable(name)
656 # end modified portion
657 except OSError:
658 # Cannot rely on checking for EEXIST, since the operating system
659 # could give priority to other errors like EACCES or EROFS
660 if not exist_ok or not os.path.isdir(name):
661 raise