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