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