Coverage for python/lsst/resources/_resourcePath.py: 21%
548 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-10-09 03:18 -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__ = ("ResourcePath", "ResourcePathExpression")
16import concurrent.futures
17import contextlib
18import copy
19import io
20import locale
21import logging
22import os
23import posixpath
24import re
25import urllib.parse
26from collections import defaultdict
27from pathlib import Path, PurePath, PurePosixPath
28from random import Random
29from typing import TypeAlias
31try:
32 import fsspec
33 from fsspec.spec import AbstractFileSystem
34except ImportError:
35 fsspec = None
36 AbstractFileSystem = type
38from collections.abc import Iterable, Iterator
39from typing import TYPE_CHECKING, Any, Literal, NamedTuple, overload
41from ._resourceHandles._baseResourceHandle import ResourceHandleProtocol
42from .utils import _get_num_workers, get_tempdir
44if TYPE_CHECKING:
45 from .utils import TransactionProtocol
48log = logging.getLogger(__name__)
50# Regex for looking for URI escapes
51ESCAPES_RE = re.compile(r"%[A-F0-9]{2}")
53# Precomputed escaped hash
54ESCAPED_HASH = urllib.parse.quote("#")
57class MBulkResult(NamedTuple):
58 """Report on a bulk operation."""
60 success: bool
61 exception: Exception | None
64_EXECUTOR_TYPE: TypeAlias = type[
65 concurrent.futures.ThreadPoolExecutor | concurrent.futures.ProcessPoolExecutor
66]
68# Cache value for executor class so as not to issue warning multiple
69# times but still allow tests to override the value.
70_POOL_EXECUTOR_CLASS: _EXECUTOR_TYPE | None = None
73def _get_executor_class() -> _EXECUTOR_TYPE:
74 """Return the executor class used for parallelized execution.
76 Returns
77 -------
78 cls : `concurrent.futures.Executor`
79 The ``Executor`` class. Default is
80 `concurrent.futures.ThreadPoolExecutor`. Can be set explicitly by
81 setting the ``$LSST_RESOURCES_EXECUTOR`` environment variable to
82 "thread" or "process". Returns "thread" pool if the value of the
83 variable is not recognized.
84 """
85 global _POOL_EXECUTOR_CLASS
87 if _POOL_EXECUTOR_CLASS is not None:
88 return _POOL_EXECUTOR_CLASS
90 pool_executor_classes = {
91 "threads": concurrent.futures.ThreadPoolExecutor,
92 "process": concurrent.futures.ProcessPoolExecutor,
93 }
94 default_executor = "threads"
95 external = os.getenv("LSST_RESOURCES_EXECUTOR", default_executor)
96 if not external:
97 external = default_executor
98 if external not in pool_executor_classes:
99 log.warning(
100 "Unrecognized value of '%s' for LSST_RESOURCES_EXECUTOR env var. Using '%s'",
101 external,
102 default_executor,
103 )
104 external = default_executor
105 _POOL_EXECUTOR_CLASS = pool_executor_classes[external]
106 return _POOL_EXECUTOR_CLASS
109@contextlib.contextmanager
110def _patch_environ(new_values: dict[str, str]) -> Iterator[None]:
111 """Patch os.environ temporarily using the supplied values.
113 Parameters
114 ----------
115 new_values : `dict` [ `str`, `str` ]
116 New values to be stored in the environment.
117 """
118 old_values: dict[str, str] = {}
119 for k, v in new_values.items():
120 if k in os.environ:
121 old_values[k] = os.environ[k]
122 os.environ[k] = v
124 try:
125 yield
126 finally:
127 for k in new_values:
128 del os.environ[k]
129 if k in old_values:
130 os.environ[k] = old_values[k]
133class ResourcePath: # numpydoc ignore=PR02
134 """Convenience wrapper around URI parsers.
136 Provides access to URI components and can convert file
137 paths into absolute path URIs. Scheme-less URIs are treated as if
138 they are local file system paths and are converted to absolute URIs.
140 A specialist subclass is created for each supported URI scheme.
142 Parameters
143 ----------
144 uri : `str`, `pathlib.Path`, `urllib.parse.ParseResult`, or `ResourcePath`
145 URI in string form. Can be scheme-less if referring to a relative
146 path or an absolute path on the local file system.
147 root : `str` or `ResourcePath`, optional
148 When fixing up a relative path in a ``file`` scheme or if scheme-less,
149 use this as the root. Must be absolute. If `None` the current
150 working directory will be used. Can be any supported URI scheme.
151 Not used if ``forceAbsolute`` is `False`.
152 forceAbsolute : `bool`, optional
153 If `True`, scheme-less relative URI will be converted to an absolute
154 path using a ``file`` scheme. If `False` scheme-less URI will remain
155 scheme-less and will not be updated to ``file`` or absolute path unless
156 it is already an absolute path, in which case it will be updated to
157 a ``file`` scheme.
158 forceDirectory : `bool` or `None`, optional
159 If `True` forces the URI to end with a separator. If `False` the URI
160 is interpreted as a file-like entity. Default, `None`, is that the
161 given URI is interpreted as a directory if there is a trailing ``/`` or
162 for some schemes the system will check to see if it is a file or a
163 directory.
164 isTemporary : `bool`, optional
165 If `True` indicates that this URI points to a temporary resource.
166 The default is `False`, unless ``uri`` is already a `ResourcePath`
167 instance and ``uri.isTemporary is True``.
169 Notes
170 -----
171 A non-standard URI of the form ``file:dir/file.txt`` is always converted
172 to an absolute ``file`` URI.
173 """
175 _pathLib: type[PurePath] = PurePosixPath
176 """Path library to use for this scheme."""
178 _pathModule = posixpath
179 """Path module to use for this scheme."""
181 transferModes: tuple[str, ...] = ("copy", "auto", "move")
182 """Transfer modes supported by this implementation.
184 Move is special in that it is generally a copy followed by an unlink.
185 Whether that unlink works depends critically on whether the source URI
186 implements unlink. If it does not the move will be reported as a failure.
187 """
189 transferDefault: str = "copy"
190 """Default mode to use for transferring if ``auto`` is specified."""
192 quotePaths = True
193 """True if path-like elements modifying a URI should be quoted.
195 All non-schemeless URIs have to internally use quoted paths. Therefore
196 if a new file name is given (e.g. to updatedFile or join) a decision must
197 be made whether to quote it to be consistent.
198 """
200 isLocal = False
201 """If `True` this URI refers to a local file."""
203 # This is not an ABC with abstract methods because the __new__ being
204 # a factory confuses mypy such that it assumes that every constructor
205 # returns a ResourcePath and then determines that all the abstract methods
206 # are still abstract. If they are not marked abstract but just raise
207 # mypy is fine with it.
209 # mypy is confused without these
210 _uri: urllib.parse.ParseResult
211 isTemporary: bool
212 dirLike: bool | None
213 """Whether the resource looks like a directory resource. `None` means that
214 the status is uncertain."""
216 def __new__(
217 cls,
218 uri: ResourcePathExpression,
219 root: str | ResourcePath | None = None,
220 forceAbsolute: bool = True,
221 forceDirectory: bool | None = None,
222 isTemporary: bool | None = None,
223 ) -> ResourcePath:
224 """Create and return new specialist ResourcePath subclass."""
225 parsed: urllib.parse.ParseResult
226 dirLike: bool | None = forceDirectory
227 subclass: type[ResourcePath] | None = None
229 # Force root to be a ResourcePath -- this simplifies downstream
230 # code.
231 if root is None:
232 root_uri = None
233 elif isinstance(root, str):
234 root_uri = ResourcePath(root, forceDirectory=True, forceAbsolute=True)
235 else:
236 root_uri = root
238 if isinstance(uri, os.PathLike):
239 uri = str(uri)
241 # Record if we need to post process the URI components
242 # or if the instance is already fully configured
243 if isinstance(uri, str):
244 # Since local file names can have special characters in them
245 # we need to quote them for the parser but we can unquote
246 # later. Assume that all other URI schemes are quoted.
247 # Since sometimes people write file:/a/b and not file:///a/b
248 # we should not quote in the explicit case of file:
249 if "://" not in uri and not uri.startswith("file:"):
250 if ESCAPES_RE.search(uri):
251 log.warning("Possible double encoding of %s", uri)
252 else:
253 # Fragments are generally not encoded so we must search
254 # for the fragment boundary ourselves. This is making
255 # an assumption that the filename does not include a "#"
256 # and also that there is no "/" in the fragment itself.
257 to_encode = uri
258 fragment = ""
259 if "#" in uri:
260 dirpos = uri.rfind("/")
261 trailing = uri[dirpos + 1 :]
262 hashpos = trailing.rfind("#")
263 if hashpos != -1:
264 fragment = trailing[hashpos:]
265 to_encode = uri[: dirpos + hashpos + 1]
267 uri = urllib.parse.quote(to_encode) + fragment
269 parsed = urllib.parse.urlparse(uri)
270 elif isinstance(uri, urllib.parse.ParseResult):
271 parsed = copy.copy(uri)
272 # If we are being instantiated with a subclass, rather than
273 # ResourcePath, ensure that that subclass is used directly.
274 # This could lead to inconsistencies if this constructor
275 # is used externally outside of the ResourcePath.replace() method.
276 # S3ResourcePath(urllib.parse.urlparse("file://a/b.txt"))
277 # will be a problem.
278 # This is needed to prevent a schemeless absolute URI become
279 # a file URI unexpectedly when calling updatedFile or
280 # updatedExtension
281 if cls is not ResourcePath:
282 parsed, dirLike = cls._fixDirectorySep(parsed, forceDirectory)
283 subclass = cls
285 elif isinstance(uri, ResourcePath):
286 # Since ResourcePath is immutable we can return the argument
287 # unchanged if it already agrees with forceDirectory, isTemporary,
288 # and forceAbsolute.
289 # We invoke __new__ again with str(self) to add a scheme for
290 # forceAbsolute, but for the others that seems more likely to paper
291 # over logic errors than do something useful, so we just raise.
292 if forceDirectory is not None and uri.dirLike is not None and forceDirectory is not uri.dirLike:
293 # Can not force a file-like URI to become a dir-like one or
294 # vice versa.
295 raise RuntimeError(
296 f"{uri} can not be forced to change directory vs file state when previously declared."
297 )
298 if isTemporary is not None and isTemporary is not uri.isTemporary:
299 raise RuntimeError(
300 f"{uri} is already a {'temporary' if uri.isTemporary else 'permanent'} "
301 f"ResourcePath; cannot make it {'temporary' if isTemporary else 'permanent'}."
302 )
304 if forceAbsolute and not uri.scheme:
305 # Create new absolute from relative.
306 return ResourcePath(
307 str(uri),
308 root=root,
309 forceAbsolute=forceAbsolute,
310 forceDirectory=forceDirectory or uri.dirLike,
311 isTemporary=uri.isTemporary,
312 )
313 elif forceDirectory is not None and uri.dirLike is None:
314 # Clone but with a new dirLike status.
315 return uri.replace(forceDirectory=forceDirectory)
316 return uri
317 else:
318 raise ValueError(
319 f"Supplied URI must be string, Path, ResourcePath, or ParseResult but got '{uri!r}'"
320 )
322 if subclass is None:
323 # Work out the subclass from the URI scheme
324 if not parsed.scheme:
325 # Root may be specified as a ResourcePath that overrides
326 # the schemeless determination.
327 if (
328 root_uri is not None
329 and root_uri.scheme != "file" # file scheme has different code path
330 and not parsed.path.startswith("/") # Not already absolute path
331 ):
332 if root_uri.dirLike is False:
333 raise ValueError(
334 f"Root URI ({root}) was not a directory so can not be joined with"
335 f" path {parsed.path!r}"
336 )
337 # If root is temporary or this schemeless is temporary we
338 # assume this URI is temporary.
339 isTemporary = isTemporary or root_uri.isTemporary
340 joined = root_uri.join(
341 parsed.path, forceDirectory=forceDirectory, isTemporary=isTemporary
342 )
344 # Rather than returning this new ResourcePath directly we
345 # instead extract the path and the scheme and adjust the
346 # URI we were given -- we need to do this to preserve
347 # fragments since join() will drop them.
348 parsed = parsed._replace(scheme=joined.scheme, path=joined.path, netloc=joined.netloc)
349 subclass = type(joined)
351 # Clear the root parameter to indicate that it has
352 # been applied already.
353 root_uri = None
354 else:
355 from .schemeless import SchemelessResourcePath
357 subclass = SchemelessResourcePath
358 elif parsed.scheme == "file":
359 from .file import FileResourcePath
361 subclass = FileResourcePath
362 elif parsed.scheme == "s3":
363 from .s3 import S3ResourcePath
365 subclass = S3ResourcePath
366 elif parsed.scheme.startswith("http"):
367 from .http import HttpResourcePath
369 subclass = HttpResourcePath
370 elif parsed.scheme in {"dav", "davs"}:
371 from .dav import DavResourcePath
373 subclass = DavResourcePath
374 elif parsed.scheme == "gs":
375 from .gs import GSResourcePath
377 subclass = GSResourcePath
378 elif parsed.scheme == "resource":
379 # Rules for scheme names disallow pkg_resource
380 from .packageresource import PackageResourcePath
382 subclass = PackageResourcePath
383 elif parsed.scheme == "mem":
384 # in-memory datastore object
385 from .mem import InMemoryResourcePath
387 subclass = InMemoryResourcePath
388 else:
389 raise NotImplementedError(
390 f"No URI support for scheme: '{parsed.scheme}' in {parsed.geturl()}"
391 )
393 parsed, dirLike = subclass._fixupPathUri(
394 parsed, root=root_uri, forceAbsolute=forceAbsolute, forceDirectory=forceDirectory
395 )
397 # It is possible for the class to change from schemeless
398 # to file so handle that
399 if parsed.scheme == "file":
400 from .file import FileResourcePath
402 subclass = FileResourcePath
404 # Now create an instance of the correct subclass and set the
405 # attributes directly
406 self = object.__new__(subclass)
407 self._uri = parsed
408 self.dirLike = dirLike
409 if isTemporary is None:
410 isTemporary = False
411 self.isTemporary = isTemporary
412 return self
414 @property
415 def scheme(self) -> str:
416 """Return the URI scheme.
418 Notes
419 -----
420 (``://`` is not part of the scheme).
421 """
422 return self._uri.scheme
424 @property
425 def netloc(self) -> str:
426 """Return the URI network location."""
427 return self._uri.netloc
429 @property
430 def path(self) -> str:
431 """Return the path component of the URI."""
432 return self._uri.path
434 @property
435 def unquoted_path(self) -> str:
436 """Return path component of the URI with any URI quoting reversed."""
437 return urllib.parse.unquote(self._uri.path)
439 @property
440 def ospath(self) -> str:
441 """Return the path component of the URI localized to current OS."""
442 raise AttributeError(f"Non-file URI ({self}) has no local OS path.")
444 @property
445 def relativeToPathRoot(self) -> str:
446 """Return path relative to network location.
448 This is the path property with posix separator stripped
449 from the left hand side of the path.
451 Always unquotes.
452 """
453 relToRoot = self.path.lstrip("/")
454 if relToRoot == "":
455 return "./"
456 return urllib.parse.unquote(relToRoot)
458 @property
459 def is_root(self) -> bool:
460 """Return whether this URI points to the root of the network location.
462 This means that the path components refers to the top level.
463 """
464 relpath = self.relativeToPathRoot
465 if relpath == "./":
466 return True
467 return False
469 @property
470 def fragment(self) -> str:
471 """Return the fragment component of the URI. May be quoted."""
472 return self._uri.fragment
474 @property
475 def unquoted_fragment(self) -> str:
476 """Return unquoted fragment."""
477 return urllib.parse.unquote(self.fragment)
479 @property
480 def params(self) -> str:
481 """Return any parameters included in the URI."""
482 return self._uri.params
484 @property
485 def query(self) -> str:
486 """Return any query strings included in the URI."""
487 return self._uri.query
489 def geturl(self) -> str:
490 """Return the URI in string form.
492 Returns
493 -------
494 url : `str`
495 String form of URI.
496 """
497 return self._uri.geturl()
499 def to_fsspec(self) -> tuple[AbstractFileSystem, str]:
500 """Return an abstract file system and path that can be used by fsspec.
502 Returns
503 -------
504 fs : `fsspec.spec.AbstractFileSystem`
505 A file system object suitable for use with the returned path.
506 path : `str`
507 A path that can be opened by the file system object.
508 """
509 if fsspec is None:
510 raise ImportError("fsspec is not available")
511 # By default give the URL to fsspec and hope.
512 return fsspec.url_to_fs(self.geturl())
514 def root_uri(self) -> ResourcePath:
515 """Return the base root URI.
517 Returns
518 -------
519 uri : `ResourcePath`
520 Root URI.
521 """
522 return self.replace(path="", query="", fragment="", params="", forceDirectory=True)
524 def split(self) -> tuple[ResourcePath, str]:
525 """Split URI into head and tail.
527 Returns
528 -------
529 head: `ResourcePath`
530 Everything leading up to tail, expanded and normalized as per
531 ResourcePath rules.
532 tail : `str`
533 Last path component. Tail will be empty if path ends on a
534 separator or if the URI is known to be associated with a directory.
535 Tail will never contain separators. It will be unquoted.
537 Notes
538 -----
539 Equivalent to `os.path.split` where head preserves the URI
540 components. In some cases this method can result in a file system
541 check to verify whether the URI is a directory or not (only if
542 ``forceDirectory`` was `None` during construction). For a scheme-less
543 URI this can mean that the result might change depending on current
544 working directory.
545 """
546 if self.isdir():
547 # This is known to be a directory so must return itself and
548 # the empty string.
549 return self, ""
551 head, tail = self._pathModule.split(self.path)
552 headuri = self._uri._replace(path=head, fragment="", query="", params="")
554 # The file part should never include quoted metacharacters
555 tail = urllib.parse.unquote(tail)
557 # Schemeless is special in that it can be a relative path.
558 # We need to ensure that it stays that way. All other URIs will
559 # be absolute already.
560 forceAbsolute = self.isabs()
561 return ResourcePath(headuri, forceDirectory=True, forceAbsolute=forceAbsolute), tail
563 def basename(self) -> str:
564 """Return the base name, last element of path, of the URI.
566 Returns
567 -------
568 tail : `str`
569 Last part of the path attribute. Trail will be empty if path ends
570 on a separator.
572 Notes
573 -----
574 If URI ends on a slash returns an empty string. This is the second
575 element returned by `split()`.
577 Equivalent of `os.path.basename`.
578 """
579 return self.split()[1]
581 def dirname(self) -> ResourcePath:
582 """Return the directory component of the path as a new `ResourcePath`.
584 Returns
585 -------
586 head : `ResourcePath`
587 Everything except the tail of path attribute, expanded and
588 normalized as per ResourcePath rules.
590 Notes
591 -----
592 Equivalent of `os.path.dirname`. If this is a directory URI it will
593 be returned unchanged. If the parent directory is always required
594 use `parent`.
595 """
596 return self.split()[0]
598 def parent(self) -> ResourcePath:
599 """Return a `ResourcePath` of the parent directory.
601 Returns
602 -------
603 head : `ResourcePath`
604 Everything except the tail of path attribute, expanded and
605 normalized as per `ResourcePath` rules.
607 Notes
608 -----
609 For a file-like URI this will be the same as calling `dirname`.
610 For a directory-like URI this will always return the parent directory
611 whereas `dirname()` will return the original URI. This is consistent
612 with `os.path.dirname` compared to the `pathlib.Path` property
613 ``parent``.
614 """
615 if self.dirLike is False:
616 # os.path.split() is slightly faster than calling Path().parent.
617 return self.dirname()
618 # When self is dir-like, returns its parent directory,
619 # regardless of the presence of a trailing separator
620 originalPath = self._pathLib(self.path)
621 parentPath = originalPath.parent
622 return self.replace(path=str(parentPath), forceDirectory=True, fragment="", query="", params="")
624 def replace(
625 self, forceDirectory: bool | None = None, isTemporary: bool = False, **kwargs: Any
626 ) -> ResourcePath:
627 """Return new `ResourcePath` with specified components replaced.
629 Parameters
630 ----------
631 forceDirectory : `bool` or `None`, optional
632 Parameter passed to ResourcePath constructor to force this
633 new URI to be dir-like or file-like.
634 isTemporary : `bool`, optional
635 Indicate that the resulting URI is temporary resource.
636 **kwargs
637 Components of a `urllib.parse.ParseResult` that should be
638 modified for the newly-created `ResourcePath`.
640 Returns
641 -------
642 new : `ResourcePath`
643 New `ResourcePath` object with updated values.
645 Notes
646 -----
647 Does not, for now, allow a change in URI scheme.
648 """
649 # Disallow a change in scheme
650 if "scheme" in kwargs:
651 raise ValueError(f"Can not use replace() method to change URI scheme for {self}")
652 result = self.__class__(
653 self._uri._replace(**kwargs), forceDirectory=forceDirectory, isTemporary=isTemporary
654 )
655 result._copy_extra_attributes(self)
656 return result
658 def updatedFile(self, newfile: str) -> ResourcePath:
659 """Return new URI with an updated final component of the path.
661 Parameters
662 ----------
663 newfile : `str`
664 File name with no path component.
666 Returns
667 -------
668 updated : `ResourcePath`
669 Updated `ResourcePath` with new updated final component.
671 Notes
672 -----
673 Forces the ``ResourcePath.dirLike`` attribute to be false. The new file
674 path will be quoted if necessary. If the current URI is known to
675 refer to a directory, the new file will be joined to the current file.
676 It is recommended that this behavior no longer be used and a call
677 to `isdir` by the caller should be used to decide whether to join or
678 replace. In the future this method may be modified to always replace
679 the final element of the path.
680 """
681 if self.dirLike:
682 return self.join(newfile, forceDirectory=False)
683 return self.parent().join(newfile, forceDirectory=False)
685 def updatedExtension(self, ext: str | None) -> ResourcePath:
686 """Return a new `ResourcePath` with updated file extension.
688 All file extensions are replaced.
690 Parameters
691 ----------
692 ext : `str` or `None`
693 New extension. If an empty string is given any extension will
694 be removed. If `None` is given there will be no change.
696 Returns
697 -------
698 updated : `ResourcePath`
699 URI with the specified extension. Can return itself if
700 no extension was specified.
701 """
702 if ext is None:
703 return self
705 # Get the extension
706 current = self.getExtension()
708 # Nothing to do if the extension already matches
709 if current == ext:
710 return self
712 # Remove the current extension from the path
713 # .fits.gz counts as one extension do not use os.path.splitext
714 path = self.path
715 if current:
716 path = path.removesuffix(current)
718 # Ensure that we have a leading "." on file extension (and we do not
719 # try to modify the empty string)
720 if ext and not ext.startswith("."):
721 ext = "." + ext
723 return self.replace(path=path + ext, forceDirectory=False)
725 def getExtension(self) -> str:
726 """Return the extension(s) associated with this URI path.
728 Returns
729 -------
730 ext : `str`
731 The file extension (including the ``.``). Can be empty string
732 if there is no file extension. Usually returns only the last
733 file extension unless there is a special extension modifier
734 indicating file compression, in which case the combined
735 extension (e.g. ``.fits.gz``) will be returned.
737 Notes
738 -----
739 Does not distinguish between file and directory URIs when determining
740 a suffix. An extension is only determined from the final component
741 of the path.
742 """
743 special = {".gz", ".bz2", ".xz", ".fz"}
745 # path lib will ignore any "." in directories.
746 # path lib works well:
747 # extensions = self._pathLib(self.path).suffixes
748 # But the constructor is slow. Therefore write our own implementation.
749 # Strip trailing separator if present, do not care if this is a
750 # directory or not.
751 parts = self.path.rstrip("/").rsplit(self._pathModule.sep, 1)
752 _, *extensions = parts[-1].split(".")
754 if not extensions:
755 return ""
756 extensions = ["." + x for x in extensions]
758 ext = extensions.pop()
760 # Multiple extensions, decide whether to include the final two
761 if extensions and ext in special:
762 ext = f"{extensions[-1]}{ext}"
764 return ext
766 def join(
767 self, path: str | ResourcePath, isTemporary: bool | None = None, forceDirectory: bool | None = None
768 ) -> ResourcePath:
769 """Return new `ResourcePath` with additional path components.
771 Parameters
772 ----------
773 path : `str`, `ResourcePath`
774 Additional file components to append to the current URI. Will be
775 quoted depending on the associated URI scheme. If the path looks
776 like a URI referring to an absolute location, it will be returned
777 directly (matching the behavior of `os.path.join`). It can
778 also be a `ResourcePath`. Fragments are propagated.
779 isTemporary : `bool`, optional
780 Indicate that the resulting URI represents a temporary resource.
781 Default is ``self.isTemporary``.
782 forceDirectory : `bool` or `None`, optional
783 If `True` forces the URI to end with a separator. If `False` the
784 resultant URI is declared to refer to a file. `None` indicates
785 that the file directory status is unknown.
787 Returns
788 -------
789 new : `ResourcePath`
790 New URI with the path appended.
792 Notes
793 -----
794 Schemeless URIs assume local path separator but all other URIs assume
795 POSIX separator if the supplied path has directory structure. It
796 may be this never becomes a problem but datastore templates assume
797 POSIX separator is being used.
799 If an absolute `ResourcePath` is given for ``path`` is is assumed that
800 this should be returned directly. Giving a ``path`` of an absolute
801 scheme-less URI is not allowed for safety reasons as it may indicate
802 a mistake in the calling code.
804 It is an error to attempt to join to something that is known to
805 refer to a file. Use `updatedFile` if the file is to be
806 replaced.
808 If an unquoted ``#`` is included in the path it is assumed to be
809 referring to a fragment and not part of the file name.
811 Raises
812 ------
813 ValueError
814 Raised if the given path object refers to a directory but the
815 ``forceDirectory`` parameter insists the outcome should be a file,
816 and vice versa. Also raised if the URI being joined with is known
817 to refer to a file.
818 RuntimeError
819 Raised if this attempts to join a temporary URI to a non-temporary
820 URI.
821 """
822 if self.dirLike is False:
823 raise ValueError("Can not join a new path component to a file.")
824 if isTemporary is None:
825 isTemporary = self.isTemporary
826 elif not isTemporary and self.isTemporary:
827 raise RuntimeError("Cannot join temporary URI to non-temporary URI.")
828 # If we have a full URI in path we will use it directly
829 # but without forcing to absolute so that we can trap the
830 # expected option of relative path.
831 path_uri = ResourcePath(
832 path, forceAbsolute=False, forceDirectory=forceDirectory, isTemporary=isTemporary
833 )
834 if forceDirectory is not None and path_uri.dirLike is not forceDirectory:
835 raise ValueError(
836 "The supplied path URI to join has inconsistent directory state "
837 f"with forceDirectory parameter: {path_uri.dirLike} vs {forceDirectory}"
838 )
839 forceDirectory = path_uri.dirLike
841 if path_uri.isabs():
842 # Absolute URI so return it directly.
843 return path_uri
845 # We want to propagate fragments to the joined path and we rely on
846 # the ResourcePath parser to find these fragments for us even in plain
847 # strings. Must assume there are no `#` characters in filenames.
848 if not isinstance(path, str) or path_uri.fragment:
849 path = path_uri.unquoted_path
851 # Might need to quote the path.
852 if self.quotePaths:
853 path = urllib.parse.quote(path)
855 newpath = self._pathModule.normpath(self._pathModule.join(self.path, path))
857 # normpath can strip trailing / so we force directory if the supplied
858 # path ended with a /
859 has_dir_sep = path.endswith(self._pathModule.sep)
860 if forceDirectory is None and has_dir_sep:
861 forceDirectory = True
862 elif forceDirectory is False and has_dir_sep:
863 raise ValueError("Path to join has trailing / but is being forced to be a file.")
864 return self.replace(
865 path=newpath,
866 forceDirectory=forceDirectory,
867 isTemporary=isTemporary,
868 fragment=path_uri.fragment,
869 query=path_uri.query,
870 params=path_uri.params,
871 )
873 def relative_to(self, other: ResourcePath) -> str | None:
874 """Return the relative path from this URI to the other URI.
876 Parameters
877 ----------
878 other : `ResourcePath`
879 URI to use to calculate the relative path. Must be a parent
880 of this URI.
882 Returns
883 -------
884 subpath : `str`
885 The sub path of this URI relative to the supplied other URI.
886 Returns `None` if there is no parent child relationship.
887 Scheme and netloc must match.
888 """
889 # Scheme-less self is handled elsewhere.
890 if self.scheme != other.scheme:
891 return None
892 if self.netloc != other.netloc:
893 # Special case for localhost vs empty string.
894 # There can be many variants of localhost.
895 local_netlocs = {"", "localhost", "localhost.localdomain", "127.0.0.1"}
896 if not {self.netloc, other.netloc}.issubset(local_netlocs):
897 return None
899 enclosed_path = self._pathLib(self.relativeToPathRoot)
900 parent_path = other.relativeToPathRoot
901 subpath: str | None
902 try:
903 subpath = str(enclosed_path.relative_to(parent_path))
904 except ValueError:
905 subpath = None
906 else:
907 subpath = urllib.parse.unquote(subpath)
908 return subpath
910 def exists(self) -> bool:
911 """Indicate that the resource is available.
913 Returns
914 -------
915 exists : `bool`
916 `True` if the resource exists.
917 """
918 raise NotImplementedError()
920 @classmethod
921 def _group_uris(cls, uris: Iterable[ResourcePath]) -> dict[type[ResourcePath], list[ResourcePath]]:
922 """Group URIs by class/scheme."""
923 grouped: dict[type, list[ResourcePath]] = defaultdict(list)
924 for uri in uris:
925 grouped[uri.__class__].append(uri)
926 return grouped
928 @classmethod
929 def mexists(
930 cls, uris: Iterable[ResourcePath], *, num_workers: int | None = None
931 ) -> dict[ResourcePath, bool]:
932 """Check for existence of multiple URIs at once.
934 Parameters
935 ----------
936 uris : iterable of `ResourcePath`
937 The URIs to test.
938 num_workers : `int` or `None`, optional
939 The number of parallel workers to use when checking for existence
940 If `None`, the default value will be taken from the environment.
941 If this number is higher than the default and a thread pool is
942 used, there may not be enough cached connections available.
944 Returns
945 -------
946 existence : `dict` of [`ResourcePath`, `bool`]
947 Mapping of original URI to boolean indicating existence.
948 """
949 existence: dict[ResourcePath, bool] = {}
950 for uri_class, group in cls._group_uris(uris).items():
951 existence.update(uri_class._mexists(group, num_workers=num_workers))
953 return existence
955 @classmethod
956 def _mexists(
957 cls, uris: Iterable[ResourcePath], *, num_workers: int | None = None
958 ) -> dict[ResourcePath, bool]:
959 """Check for existence of multiple URIs at once.
961 Implementation helper method for `mexists`.
964 Parameters
965 ----------
966 uris : iterable of `ResourcePath`
967 The URIs to test.
968 num_workers : `int` or `None`, optional
969 The number of parallel workers to use when checking for existence
970 If `None`, the default value will be taken from the environment.
972 Returns
973 -------
974 existence : `dict` of [`ResourcePath`, `bool`]
975 Mapping of original URI to boolean indicating existence.
976 """
977 pool_executor_class = _get_executor_class()
978 if issubclass(pool_executor_class, concurrent.futures.ProcessPoolExecutor):
979 # Patch the environment to make it think there is only one worker
980 # for each subprocess.
981 with _patch_environ({"LSST_RESOURCES_NUM_WORKERS": "1"}):
982 return cls._mexists_pool(pool_executor_class, uris)
983 else:
984 return cls._mexists_pool(pool_executor_class, uris, num_workers=num_workers)
986 @classmethod
987 def _mexists_pool(
988 cls,
989 pool_executor_class: _EXECUTOR_TYPE,
990 uris: Iterable[ResourcePath],
991 *,
992 num_workers: int | None = None,
993 ) -> dict[ResourcePath, bool]:
994 """Check for existence of multiple URIs at once using specified pool
995 executor.
997 Implementation helper method for `_mexists`.
999 Parameters
1000 ----------
1001 pool_executor_class : `type` [ `concurrent.futures.Executor` ]
1002 Type of executor pool to use.
1003 uris : iterable of `ResourcePath`
1004 The URIs to test.
1005 num_workers : `int` or `None`, optional
1006 The number of parallel workers to use when checking for existence
1007 If `None`, the default value will be taken from the environment.
1009 Returns
1010 -------
1011 existence : `dict` of [`ResourcePath`, `bool`]
1012 Mapping of original URI to boolean indicating existence.
1013 """
1014 max_workers = num_workers if num_workers is not None else _get_num_workers()
1015 with pool_executor_class(max_workers=max_workers) as exists_executor:
1016 future_exists = {exists_executor.submit(uri.exists): uri for uri in uris}
1018 results: dict[ResourcePath, bool] = {}
1019 for future in concurrent.futures.as_completed(future_exists):
1020 uri = future_exists[future]
1021 try:
1022 exists = future.result()
1023 except Exception:
1024 exists = False
1025 results[uri] = exists
1026 return results
1028 @classmethod
1029 def mtransfer(
1030 cls,
1031 transfer: str,
1032 from_to: Iterable[tuple[ResourcePath, ResourcePath]],
1033 overwrite: bool = False,
1034 transaction: TransactionProtocol | None = None,
1035 do_raise: bool = True,
1036 ) -> dict[ResourcePath, MBulkResult]:
1037 """Transfer many files in bulk.
1039 Parameters
1040 ----------
1041 transfer : `str`
1042 Mode to use for transferring the resource. Generically there are
1043 many standard options: copy, link, symlink, hardlink, relsymlink.
1044 Not all URIs support all modes.
1045 from_to : `list` [ `tuple` [ `ResourcePath`, `ResourcePath` ] ]
1046 A sequence of the source URIs and the target URIs.
1047 overwrite : `bool`, optional
1048 Allow an existing file to be overwritten. Defaults to `False`.
1049 transaction : `~lsst.resources.utils.TransactionProtocol`, optional
1050 A transaction object that can (depending on implementation)
1051 rollback transfers on error. Not guaranteed to be implemented.
1052 The transaction object must be thread safe.
1053 do_raise : `bool`, optional
1054 If `True` an `ExceptionGroup` will be raised containing any
1055 exceptions raised by the individual transfers. If `False`, or if
1056 there were no exceptions, a dict reporting the status of each
1057 `ResourcePath` will be returned.
1059 Returns
1060 -------
1061 copy_status : `dict` [ `ResourcePath`, `MBulkResult` ]
1062 A dict of all the transfer attempts with a value indicating
1063 whether the transfer succeeded for the target URI. If ``do_raise``
1064 is `True`, this will only be returned if there are no errors.
1065 """
1066 pool_executor_class = _get_executor_class()
1067 if issubclass(pool_executor_class, concurrent.futures.ProcessPoolExecutor):
1068 # Patch the environment to make it think there is only one worker
1069 # for each subprocess.
1070 with _patch_environ({"LSST_RESOURCES_NUM_WORKERS": "1"}):
1071 return cls._mtransfer(
1072 pool_executor_class,
1073 transfer,
1074 from_to,
1075 overwrite=overwrite,
1076 transaction=transaction,
1077 do_raise=do_raise,
1078 )
1079 return cls._mtransfer(
1080 pool_executor_class,
1081 transfer,
1082 from_to,
1083 overwrite=overwrite,
1084 transaction=transaction,
1085 do_raise=do_raise,
1086 )
1088 @classmethod
1089 def _mtransfer(
1090 cls,
1091 pool_executor_class: _EXECUTOR_TYPE,
1092 transfer: str,
1093 from_to: Iterable[tuple[ResourcePath, ResourcePath]],
1094 overwrite: bool = False,
1095 transaction: TransactionProtocol | None = None,
1096 do_raise: bool = True,
1097 ) -> dict[ResourcePath, MBulkResult]:
1098 """Transfer many files in bulk.
1100 Parameters
1101 ----------
1102 transfer : `str`
1103 Mode to use for transferring the resource. Generically there are
1104 many standard options: copy, link, symlink, hardlink, relsymlink.
1105 Not all URIs support all modes.
1106 from_to : `list` [ `tuple` [ `ResourcePath`, `ResourcePath` ] ]
1107 A sequence of the source URIs and the target URIs.
1108 overwrite : `bool`, optional
1109 Allow an existing file to be overwritten. Defaults to `False`.
1110 transaction : `~lsst.resources.utils.TransactionProtocol`, optional
1111 A transaction object that can (depending on implementation)
1112 rollback transfers on error. Not guaranteed to be implemented.
1113 The transaction object must be thread safe.
1114 do_raise : `bool`, optional
1115 If `True` an `ExceptionGroup` will be raised containing any
1116 exceptions raised by the individual transfers. Else a dict
1117 reporting the status of each `ResourcePath` will be returned.
1119 Returns
1120 -------
1121 copy_status : `dict` [ `ResourcePath`, `MBulkResult` ]
1122 A dict of all the transfer attempts with a value indicating
1123 whether the transfer succeeded for the target URI.
1124 """
1125 with pool_executor_class(max_workers=_get_num_workers()) as transfer_executor:
1126 future_transfers = {
1127 transfer_executor.submit(
1128 to_uri.transfer_from,
1129 from_uri,
1130 transfer=transfer,
1131 overwrite=overwrite,
1132 transaction=transaction,
1133 multithreaded=False,
1134 ): to_uri
1135 for from_uri, to_uri in from_to
1136 }
1137 results: dict[ResourcePath, MBulkResult] = {}
1138 failed = False
1139 for future in concurrent.futures.as_completed(future_transfers):
1140 to_uri = future_transfers[future]
1141 try:
1142 future.result()
1143 except Exception as e:
1144 transferred = MBulkResult(False, e)
1145 failed = True
1146 else:
1147 transferred = MBulkResult(True, None)
1148 results[to_uri] = transferred
1150 if do_raise and failed:
1151 raise ExceptionGroup(
1152 f"Errors transferring {len(results)} artifacts",
1153 tuple(res.exception for res in results.values() if res.exception is not None),
1154 )
1156 return results
1158 def remove(self) -> None:
1159 """Remove the resource."""
1160 raise NotImplementedError()
1162 @classmethod
1163 def mremove(
1164 cls, uris: Iterable[ResourcePath], *, do_raise: bool = True
1165 ) -> dict[ResourcePath, MBulkResult]:
1166 """Remove multiple URIs at once.
1168 Parameters
1169 ----------
1170 uris : iterable of `ResourcePath`
1171 URIs to remove.
1172 do_raise : `bool`, optional
1173 If `True` an `ExceptionGroup` will be raised containing any
1174 exceptions raised by the individual transfers. If `False`, or if
1175 there were no exceptions, a dict reporting the status of each
1176 `ResourcePath` will be returned.
1178 Returns
1179 -------
1180 results : `dict` [ `ResourcePath`, `MBulkResult` ]
1181 Dictionary mapping each URI to a result object indicating whether
1182 the removal succeeded or resulted in an exception. If ``do_raise``
1183 is `True` this will only be returned if everything succeeded.
1184 """
1185 # Group URIs by scheme since some URI schemes support native bulk
1186 # APIs.
1187 results: dict[ResourcePath, MBulkResult] = {}
1188 for uri_class, group in cls._group_uris(uris).items():
1189 results.update(uri_class._mremove(group))
1190 if do_raise:
1191 failed = any(not r.success for r in results.values())
1192 if failed:
1193 s = "s" if len(results) != 1 else ""
1194 raise ExceptionGroup(
1195 f"Error{s} removing {len(results)} artifact{s}",
1196 tuple(res.exception for res in results.values() if res.exception is not None),
1197 )
1199 return results
1201 @classmethod
1202 def _mremove(cls, uris: Iterable[ResourcePath]) -> dict[ResourcePath, MBulkResult]:
1203 """Remove multiple URIs using futures."""
1204 pool_executor_class = _get_executor_class()
1205 if issubclass(pool_executor_class, concurrent.futures.ProcessPoolExecutor):
1206 # Patch the environment to make it think there is only one worker
1207 # for each subprocess.
1208 with _patch_environ({"LSST_RESOURCES_NUM_WORKERS": "1"}):
1209 return cls._mremove_pool(pool_executor_class, uris)
1210 else:
1211 return cls._mremove_pool(pool_executor_class, uris)
1213 @classmethod
1214 def _mremove_pool(
1215 cls,
1216 pool_executor_class: _EXECUTOR_TYPE,
1217 uris: Iterable[ResourcePath],
1218 *,
1219 num_workers: int | None = None,
1220 ) -> dict[ResourcePath, MBulkResult]:
1221 """Remove URIs using a futures pool."""
1222 max_workers = num_workers if num_workers is not None else _get_num_workers()
1223 results: dict[ResourcePath, MBulkResult] = {}
1224 with pool_executor_class(max_workers=max_workers) as remove_executor:
1225 future_remove = {remove_executor.submit(uri.remove): uri for uri in uris}
1226 for future in concurrent.futures.as_completed(future_remove):
1227 try:
1228 future.result()
1229 except Exception as e:
1230 removed = MBulkResult(False, e)
1231 else:
1232 removed = MBulkResult(True, None)
1233 uri = future_remove[future]
1234 results[uri] = removed
1235 return results
1237 def isabs(self) -> bool:
1238 """Indicate that the resource is fully specified.
1240 For non-schemeless URIs this is always true.
1242 Returns
1243 -------
1244 isabs : `bool`
1245 `True` in all cases except schemeless URI.
1246 """
1247 return True
1249 def abspath(self) -> ResourcePath:
1250 """Return URI using an absolute path.
1252 Returns
1253 -------
1254 abs : `ResourcePath`
1255 Absolute URI. For non-schemeless URIs this always returns itself.
1256 Schemeless URIs are upgraded to file URIs.
1257 """
1258 return self
1260 @contextlib.contextmanager
1261 def _as_local(
1262 self, multithreaded: bool = True, tmpdir: ResourcePath | None = None
1263 ) -> Iterator[ResourcePath]:
1264 """Return the location of the (possibly remote) resource as local file.
1266 This is a helper function for `as_local` context manager.
1268 Parameters
1269 ----------
1270 multithreaded : `bool`, optional
1271 If `True` the transfer will be allowed to attempt to improve
1272 throughput by using parallel download streams. This may of no
1273 effect if the URI scheme does not support parallel streams or
1274 if a global override has been applied. If `False` parallel
1275 streams will be disabled.
1276 tmpdir : `ResourcePath` or `None`, optional
1277 Explicit override of the temporary directory to use for remote
1278 downloads.
1280 Returns
1281 -------
1282 local_uri : `ResourcePath`
1283 A URI to a local POSIX file. This can either be the same resource
1284 or a local downloaded copy of the resource.
1285 """
1286 raise NotImplementedError()
1288 @contextlib.contextmanager
1289 def as_local(
1290 self, multithreaded: bool = True, tmpdir: ResourcePathExpression | None = None
1291 ) -> Iterator[ResourcePath]:
1292 """Return the location of the (possibly remote) resource as local file.
1294 Parameters
1295 ----------
1296 multithreaded : `bool`, optional
1297 If `True` the transfer will be allowed to attempt to improve
1298 throughput by using parallel download streams. This may of no
1299 effect if the URI scheme does not support parallel streams or
1300 if a global override has been applied. If `False` parallel
1301 streams will be disabled.
1302 tmpdir : `lsst.resources.ResourcePathExpression` or `None`, optional
1303 Explicit override of the temporary directory to use for remote
1304 downloads. This directory must be a local POSIX directory and
1305 must exist.
1307 Yields
1308 ------
1309 local : `ResourcePath`
1310 If this is a remote resource, it will be a copy of the resource
1311 on the local file system, probably in a temporary directory.
1312 For a local resource this should be the actual path to the
1313 resource.
1315 Notes
1316 -----
1317 The context manager will automatically delete any local temporary
1318 file.
1320 Examples
1321 --------
1322 Should be used as a context manager:
1324 .. code-block:: py
1326 with uri.as_local() as local:
1327 ospath = local.ospath
1328 """
1329 if self.isdir():
1330 raise IsADirectoryError(f"Directory-like URI {self} cannot be fetched as local.")
1331 temp_dir = ResourcePath(tmpdir, forceDirectory=True) if tmpdir is not None else None
1332 if temp_dir is not None and not temp_dir.isLocal:
1333 raise ValueError(f"Temporary directory for as_local must be local resource not {temp_dir}")
1334 with self._as_local(multithreaded=multithreaded, tmpdir=temp_dir) as local_uri:
1335 yield local_uri
1337 @classmethod
1338 @contextlib.contextmanager
1339 def temporary_uri(
1340 cls,
1341 prefix: ResourcePath | None = None,
1342 suffix: str | None = None,
1343 delete: bool = True,
1344 ) -> Iterator[ResourcePath]:
1345 """Create a temporary file-like URI.
1347 Parameters
1348 ----------
1349 prefix : `ResourcePath`, optional
1350 Temporary directory to use (can be any scheme). Without this the
1351 path will be formed as a local file URI in a temporary directory
1352 obtained from `lsst.resources.utils.get_tempdir`. Ensuring that the
1353 prefix location exists is the responsibility of the caller.
1354 suffix : `str`, optional
1355 A file suffix to be used. The ``.`` should be included in this
1356 suffix.
1357 delete : `bool`, optional
1358 By default the resource will be deleted when the context manager
1359 is exited. Setting this flag to `False` will leave the resource
1360 alone.
1362 Yields
1363 ------
1364 uri : `ResourcePath`
1365 The temporary URI. Will be removed when the context is completed.
1366 """
1367 if prefix is None:
1368 prefix = ResourcePath(get_tempdir(), forceDirectory=True)
1370 # Need to create a randomized file name. For consistency do not
1371 # use mkstemp for local and something else for remote. Additionally
1372 # this method does not create the file to prevent name clashes.
1373 characters = "abcdefghijklmnopqrstuvwxyz0123456789_"
1374 rng = Random()
1375 tempname = "".join(rng.choice(characters) for _ in range(16))
1376 if suffix:
1377 tempname += suffix
1378 temporary_uri = prefix.join(tempname, isTemporary=True)
1379 if temporary_uri.isdir():
1380 # If we had a safe way to clean up a remote temporary directory, we
1381 # could support this.
1382 raise NotImplementedError("temporary_uri cannot be used to create a temporary directory.")
1383 try:
1384 yield temporary_uri
1385 finally:
1386 if delete:
1387 with contextlib.suppress(FileNotFoundError):
1388 # It's okay if this does not work because the user
1389 # removed the file.
1390 temporary_uri.remove()
1392 def read(self, size: int = -1) -> bytes:
1393 """Open the resource and return the contents in bytes.
1395 Parameters
1396 ----------
1397 size : `int`, optional
1398 The number of bytes to read. Negative or omitted indicates
1399 that all data should be read.
1400 """
1401 raise NotImplementedError()
1403 def write(self, data: bytes, overwrite: bool = True) -> None:
1404 """Write the supplied bytes to the new resource.
1406 Parameters
1407 ----------
1408 data : `bytes`
1409 The bytes to write to the resource. The entire contents of the
1410 resource will be replaced.
1411 overwrite : `bool`, optional
1412 If `True` the resource will be overwritten if it exists. Otherwise
1413 the write will fail.
1414 """
1415 raise NotImplementedError()
1417 def mkdir(self) -> None:
1418 """For a dir-like URI, create the directory resource if needed."""
1419 raise NotImplementedError()
1421 def isdir(self) -> bool:
1422 """Return True if this URI looks like a directory, else False."""
1423 return bool(self.dirLike)
1425 def size(self) -> int:
1426 """For non-dir-like URI, return the size of the resource.
1428 Returns
1429 -------
1430 sz : `int`
1431 The size in bytes of the resource associated with this URI.
1432 Returns 0 if dir-like.
1433 """
1434 raise NotImplementedError()
1436 def __str__(self) -> str:
1437 """Convert the URI to its native string form."""
1438 return self.geturl()
1440 def __repr__(self) -> str:
1441 """Return string representation suitable for evaluation."""
1442 return f'ResourcePath("{self.geturl()}")'
1444 def __eq__(self, other: Any) -> bool:
1445 """Compare supplied object with this `ResourcePath`."""
1446 if not isinstance(other, ResourcePath):
1447 return NotImplemented
1448 return self.geturl() == other.geturl()
1450 def __hash__(self) -> int:
1451 """Return hash of this object."""
1452 return hash(str(self))
1454 def __lt__(self, other: ResourcePath) -> bool:
1455 return self.geturl() < other.geturl()
1457 def __le__(self, other: ResourcePath) -> bool:
1458 return self.geturl() <= other.geturl()
1460 def __gt__(self, other: ResourcePath) -> bool:
1461 return self.geturl() > other.geturl()
1463 def __ge__(self, other: ResourcePath) -> bool:
1464 return self.geturl() >= other.geturl()
1466 def __copy__(self) -> ResourcePath:
1467 """Copy constructor.
1469 Object is immutable so copy can return itself.
1470 """
1471 # Implement here because the __new__ method confuses things
1472 return self
1474 def __deepcopy__(self, memo: Any) -> ResourcePath:
1475 """Deepcopy the object.
1477 Object is immutable so copy can return itself.
1478 """
1479 # Implement here because the __new__ method confuses things
1480 return self
1482 def __getnewargs__(self) -> tuple:
1483 """Support pickling."""
1484 return (str(self),)
1486 @classmethod
1487 def _fixDirectorySep(
1488 cls, parsed: urllib.parse.ParseResult, forceDirectory: bool | None = None
1489 ) -> tuple[urllib.parse.ParseResult, bool | None]:
1490 """Ensure that a path separator is present on directory paths.
1492 Parameters
1493 ----------
1494 parsed : `~urllib.parse.ParseResult`
1495 The result from parsing a URI using `urllib.parse`.
1496 forceDirectory : `bool` or `None`, optional
1497 If `True` forces the URI to end with a separator, otherwise given
1498 URI is interpreted as is. Specifying that the URI is conceptually
1499 equivalent to a directory can break some ambiguities when
1500 interpreting the last element of a path.
1502 Returns
1503 -------
1504 modified : `~urllib.parse.ParseResult`
1505 Update result if a URI is being handled.
1506 dirLike : `bool` or `None`
1507 `True` if given parsed URI has a trailing separator or
1508 ``forceDirectory`` is `True`. Otherwise returns the given value of
1509 ``forceDirectory``.
1510 """
1511 # Assume the forceDirectory flag can give us a clue.
1512 dirLike = forceDirectory
1514 # Directory separator
1515 sep = cls._pathModule.sep
1517 # URI is dir-like if explicitly stated or if it ends on a separator
1518 endsOnSep = parsed.path.endswith(sep)
1520 if forceDirectory is False and endsOnSep:
1521 raise ValueError(
1522 f"URI {parsed.geturl()} ends with {sep} but "
1523 "forceDirectory parameter declares it to be a file."
1524 )
1526 if forceDirectory or endsOnSep:
1527 dirLike = True
1528 # only add the separator if it's not already there
1529 if not endsOnSep:
1530 parsed = parsed._replace(path=parsed.path + sep)
1532 return parsed, dirLike
1534 @classmethod
1535 def _fixupPathUri(
1536 cls,
1537 parsed: urllib.parse.ParseResult,
1538 root: ResourcePath | None = None,
1539 forceAbsolute: bool = False,
1540 forceDirectory: bool | None = None,
1541 ) -> tuple[urllib.parse.ParseResult, bool | None]:
1542 """Correct any issues with the supplied URI.
1544 Parameters
1545 ----------
1546 parsed : `~urllib.parse.ParseResult`
1547 The result from parsing a URI using `urllib.parse`.
1548 root : `ResourcePath`, ignored
1549 Not used by the this implementation since all URIs are
1550 absolute except for those representing the local file system.
1551 forceAbsolute : `bool`, ignored.
1552 Not used by this implementation. URIs are generally always
1553 absolute.
1554 forceDirectory : `bool` or `None`, optional
1555 If `True` forces the URI to end with a separator, otherwise given
1556 URI is interpreted as is. Specifying that the URI is conceptually
1557 equivalent to a directory can break some ambiguities when
1558 interpreting the last element of a path.
1560 Returns
1561 -------
1562 modified : `~urllib.parse.ParseResult`
1563 Update result if a URI is being handled.
1564 dirLike : `bool`
1565 `True` if given parsed URI has a trailing separator or
1566 ``forceDirectory`` is `True`. Otherwise returns the given value
1567 of ``forceDirectory``.
1569 Notes
1570 -----
1571 Relative paths are explicitly not supported by RFC8089 but `urllib`
1572 does accept URIs of the form ``file:relative/path.ext``. They need
1573 to be turned into absolute paths before they can be used. This is
1574 always done regardless of the ``forceAbsolute`` parameter.
1576 AWS S3 differentiates between keys with trailing POSIX separators (i.e
1577 ``/dir`` and ``/dir/``) whereas POSIX does not necessarily.
1579 Scheme-less paths are normalized.
1580 """
1581 return cls._fixDirectorySep(parsed, forceDirectory)
1583 def transfer_from(
1584 self,
1585 src: ResourcePath,
1586 transfer: str,
1587 overwrite: bool = False,
1588 transaction: TransactionProtocol | None = None,
1589 multithreaded: bool = True,
1590 ) -> None:
1591 """Transfer to this URI from another.
1593 Parameters
1594 ----------
1595 src : `ResourcePath`
1596 Source URI.
1597 transfer : `str`
1598 Mode to use for transferring the resource. Generically there are
1599 many standard options: copy, link, symlink, hardlink, relsymlink.
1600 Not all URIs support all modes.
1601 overwrite : `bool`, optional
1602 Allow an existing file to be overwritten. Defaults to `False`.
1603 transaction : `~lsst.resources.utils.TransactionProtocol`, optional
1604 A transaction object that can (depending on implementation)
1605 rollback transfers on error. Not guaranteed to be implemented.
1606 multithreaded : `bool`, optional
1607 If `True` the transfer will be allowed to attempt to improve
1608 throughput by using parallel download streams. This may of no
1609 effect if the URI scheme does not support parallel streams or
1610 if a global override has been applied. If `False` parallel
1611 streams will be disabled.
1613 Notes
1614 -----
1615 Conceptually this is hard to scale as the number of URI schemes
1616 grow. The destination URI is more important than the source URI
1617 since that is where all the transfer modes are relevant (with the
1618 complication that "move" deletes the source).
1620 Local file to local file is the fundamental use case but every
1621 other scheme has to support "copy" to local file (with implicit
1622 support for "move") and copy from local file.
1623 All the "link" options tend to be specific to local file systems.
1625 "move" is a "copy" where the remote resource is deleted at the end.
1626 Whether this works depends on the source URI rather than the
1627 destination URI. Reverting a move on transaction rollback is
1628 expected to be problematic if a remote resource was involved.
1629 """
1630 raise NotImplementedError(f"No transfer modes supported by URI scheme {self.scheme}")
1632 def walk(
1633 self, file_filter: str | re.Pattern | None = None
1634 ) -> Iterator[list | tuple[ResourcePath, list[str], list[str]]]:
1635 """Walk the directory tree returning matching files and directories.
1637 Parameters
1638 ----------
1639 file_filter : `str` or `re.Pattern`, optional
1640 Regex to filter out files from the list before it is returned.
1642 Yields
1643 ------
1644 dirpath : `ResourcePath`
1645 Current directory being examined.
1646 dirnames : `list` of `str`
1647 Names of subdirectories within dirpath.
1648 filenames : `list` of `str`
1649 Names of all the files within dirpath.
1650 """
1651 raise NotImplementedError()
1653 @overload
1654 @classmethod
1655 def findFileResources( 1655 ↛ exitline 1655 didn't return from function 'findFileResources' because
1656 cls,
1657 candidates: Iterable[ResourcePathExpression],
1658 file_filter: str | re.Pattern | None,
1659 grouped: Literal[True],
1660 ) -> Iterator[Iterator[ResourcePath]]: ...
1662 @overload
1663 @classmethod
1664 def findFileResources( 1664 ↛ exitline 1664 didn't return from function 'findFileResources' because
1665 cls,
1666 candidates: Iterable[ResourcePathExpression],
1667 *,
1668 grouped: Literal[True],
1669 ) -> Iterator[Iterator[ResourcePath]]: ...
1671 @overload
1672 @classmethod
1673 def findFileResources( 1673 ↛ exitline 1673 didn't return from function 'findFileResources' because
1674 cls,
1675 candidates: Iterable[ResourcePathExpression],
1676 file_filter: str | re.Pattern | None = None,
1677 grouped: Literal[False] = False,
1678 ) -> Iterator[ResourcePath]: ...
1680 @classmethod
1681 def findFileResources(
1682 cls,
1683 candidates: Iterable[ResourcePathExpression],
1684 file_filter: str | re.Pattern | None = None,
1685 grouped: bool = False,
1686 ) -> Iterator[ResourcePath | Iterator[ResourcePath]]:
1687 """Get all the files from a list of values.
1689 Parameters
1690 ----------
1691 candidates : iterable [`str` or `ResourcePath`]
1692 The files to return and directories in which to look for files to
1693 return.
1694 file_filter : `str` or `re.Pattern`, optional
1695 The regex to use when searching for files within directories.
1696 By default returns all the found files.
1697 grouped : `bool`, optional
1698 If `True` the results will be grouped by directory and each
1699 yielded value will be an iterator over URIs. If `False` each
1700 URI will be returned separately.
1702 Yields
1703 ------
1704 found_file: `ResourcePath`
1705 The passed-in URIs and URIs found in passed-in directories.
1706 If grouping is enabled, each of the yielded values will be an
1707 iterator yielding members of the group. Files given explicitly
1708 will be returned as a single group at the end.
1710 Notes
1711 -----
1712 If a value is a file it is yielded immediately without checking that it
1713 exists. If a value is a directory, all the files in the directory
1714 (recursively) that match the regex will be yielded in turn.
1715 """
1716 fileRegex = None if file_filter is None else re.compile(file_filter)
1718 singles = []
1720 # Find all the files of interest
1721 for location in candidates:
1722 uri = ResourcePath(location)
1723 if uri.isdir():
1724 for found in uri.walk(fileRegex):
1725 if not found:
1726 # This means the uri does not exist and by
1727 # convention we ignore it
1728 continue
1729 root, dirs, files = found
1730 if not files:
1731 continue
1732 if grouped:
1733 yield (root.join(name) for name in files)
1734 else:
1735 for name in files:
1736 yield root.join(name)
1737 else:
1738 if grouped:
1739 singles.append(uri)
1740 else:
1741 yield uri
1743 # Finally, return any explicitly given files in one group
1744 if grouped and singles:
1745 yield iter(singles)
1747 @contextlib.contextmanager
1748 def open(
1749 self,
1750 mode: str = "r",
1751 *,
1752 encoding: str | None = None,
1753 prefer_file_temporary: bool = False,
1754 ) -> Iterator[ResourceHandleProtocol]:
1755 """Return a context manager that wraps an object that behaves like an
1756 open file at the location of the URI.
1758 Parameters
1759 ----------
1760 mode : `str`
1761 String indicating the mode in which to open the file. Values are
1762 the same as those accepted by `open`, though intrinsically
1763 read-only URI types may only support read modes, and
1764 `io.IOBase.seekable` is not guaranteed to be `True` on the returned
1765 object.
1766 encoding : `str`, optional
1767 Unicode encoding for text IO; ignored for binary IO. Defaults to
1768 ``locale.getpreferredencoding(False)``, just as `open`
1769 does.
1770 prefer_file_temporary : `bool`, optional
1771 If `True`, for implementations that require transfers from a remote
1772 system to temporary local storage and/or back, use a temporary file
1773 instead of an in-memory buffer; this is generally slower, but it
1774 may be necessary to avoid excessive memory usage by large files.
1775 Ignored by implementations that do not require a temporary.
1777 Yields
1778 ------
1779 cm : `~contextlib.AbstractContextManager`
1780 A context manager that wraps a `ResourceHandleProtocol` file-like
1781 object.
1783 Notes
1784 -----
1785 The default implementation of this method uses a local temporary buffer
1786 (in-memory or file, depending on ``prefer_file_temporary``) with calls
1787 to `read`, `write`, `as_local`, and `transfer_from` as necessary to
1788 read and write from/to remote systems. Remote writes thus occur only
1789 when the context manager is exited. `ResourcePath` implementations
1790 that can return a more efficient native buffer should do so whenever
1791 possible (as is guaranteed for local files). `ResourcePath`
1792 implementations for which `as_local` does not return a temporary are
1793 required to reimplement `open`, though they may delegate to `super`
1794 when ``prefer_file_temporary`` is `False`.
1795 """
1796 if self.isdir():
1797 raise IsADirectoryError(f"Directory-like URI {self} cannot be opened.")
1798 if "x" in mode and self.exists():
1799 raise FileExistsError(f"File at {self} already exists.")
1800 if prefer_file_temporary:
1801 if "r" in mode or "a" in mode:
1802 local_cm = self.as_local()
1803 else:
1804 local_cm = self.temporary_uri(suffix=self.getExtension())
1805 with local_cm as local_uri:
1806 assert local_uri.isTemporary, (
1807 "ResourcePath implementations for which as_local is not "
1808 "a temporary must reimplement `open`."
1809 )
1810 with open(local_uri.ospath, mode=mode, encoding=encoding) as file_buffer:
1811 if "a" in mode:
1812 file_buffer.seek(0, io.SEEK_END)
1813 yield file_buffer
1814 if "r" not in mode or "+" in mode:
1815 self.transfer_from(local_uri, transfer="copy", overwrite=("x" not in mode))
1816 else:
1817 with self._openImpl(mode, encoding=encoding) as handle:
1818 yield handle
1820 @contextlib.contextmanager
1821 def _openImpl(self, mode: str = "r", *, encoding: str | None = None) -> Iterator[ResourceHandleProtocol]:
1822 """Implement opening of a resource handle.
1824 This private method may be overridden by specific `ResourcePath`
1825 implementations to provide a customized handle like interface.
1827 Parameters
1828 ----------
1829 mode : `str`
1830 The mode the handle should be opened with
1831 encoding : `str`, optional
1832 The byte encoding of any binary text
1834 Yields
1835 ------
1836 handle : `~._resourceHandles.BaseResourceHandle`
1837 A handle that conforms to the
1838 `~._resourceHandles.BaseResourceHandle` interface
1840 Notes
1841 -----
1842 The base implementation of a file handle reads in a files entire
1843 contents into a buffer for manipulation, and then writes it back out
1844 upon close. Subclasses of this class may offer more fine grained
1845 control.
1846 """
1847 in_bytes = self.read() if "r" in mode or "a" in mode else b""
1848 if "b" in mode:
1849 bytes_buffer = io.BytesIO(in_bytes)
1850 bytes_buffer.name = str(self)
1851 if "a" in mode:
1852 bytes_buffer.seek(0, io.SEEK_END)
1853 yield bytes_buffer
1854 out_bytes = bytes_buffer.getvalue()
1855 else:
1856 if encoding is None:
1857 encoding = locale.getpreferredencoding(False)
1858 str_buffer = io.StringIO(in_bytes.decode(encoding))
1859 str_buffer.name = str(self)
1860 if "a" in mode:
1861 str_buffer.seek(0, io.SEEK_END)
1862 yield str_buffer
1863 out_bytes = str_buffer.getvalue().encode(encoding)
1864 if "r" not in mode or "+" in mode:
1865 self.write(out_bytes, overwrite=("x" not in mode))
1867 def generate_presigned_get_url(self, *, expiration_time_seconds: int) -> str:
1868 """Return a pre-signed URL that can be used to retrieve this resource
1869 using an HTTP GET without supplying any access credentials.
1871 Parameters
1872 ----------
1873 expiration_time_seconds : `int`
1874 Number of seconds until the generated URL is no longer valid.
1876 Returns
1877 -------
1878 url : `str`
1879 HTTP URL signed for GET.
1880 """
1881 raise NotImplementedError(f"URL signing is not supported for '{self.scheme}'")
1883 def generate_presigned_put_url(self, *, expiration_time_seconds: int) -> str:
1884 """Return a pre-signed URL that can be used to upload a file to this
1885 path using an HTTP PUT without supplying any access credentials.
1887 Parameters
1888 ----------
1889 expiration_time_seconds : `int`
1890 Number of seconds until the generated URL is no longer valid.
1892 Returns
1893 -------
1894 url : `str`
1895 HTTP URL signed for PUT.
1896 """
1897 raise NotImplementedError(f"URL signing is not supported for '{self.scheme}'")
1899 def _copy_extra_attributes(self, original_uri: ResourcePath) -> None:
1900 # May be overridden by subclasses to transfer attributes when a
1901 # ResourcePath is constructed using the "clone" version of the
1902 # ResourcePath constructor by passing in a ResourcePath object.
1903 pass
1906ResourcePathExpression = str | urllib.parse.ParseResult | ResourcePath | Path
1907"""Type-annotation alias for objects that can be coerced to ResourcePath.
1908"""