Coverage for python/lsst/resources/dav.py: 35%

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. 

11 

12from __future__ import annotations 

13 

14__all__ = ("DavResourcePath",) 

15 

16import contextlib 

17import datetime 

18import functools 

19import io 

20import logging 

21import os 

22import re 

23import sys 

24import threading 

25import urllib.parse 

26from collections.abc import Generator, Iterator 

27from typing import TYPE_CHECKING, Any, BinaryIO, cast 

28 

29if sys.version_info >= (3, 12): 29 ↛ 32line 29 didn't jump to line 32 because the condition on line 29 was always true

30 from typing import override 

31else: 

32 from typing_extensions import override 

33 

34try: 

35 import fsspec 

36 from fsspec.spec import AbstractFileSystem 

37except ImportError: 

38 # Hidden from type checkers so that the names above keep the types they 

39 # have when fsspec is installed. 

40 if not TYPE_CHECKING: 

41 fsspec = None 

42 AbstractFileSystem = type 

43 

44from ._resourceHandles import ResourceHandleProtocol 

45from ._resourceHandles._davResourceHandle import DavReadResourceHandle 

46from ._resourcePath import ResourceInfo, ResourcePath, ResourcePathExpression 

47from .davutils import ( 

48 DavClient, 

49 DavClientPool, 

50 DavConfigPool, 

51 DavFileMetadata, 

52 normalize_path, 

53 normalize_url, 

54) 

55from .utils import get_tempdir 

56 

57if TYPE_CHECKING: 

58 from .utils import TransactionProtocol 

59 

60from lsst.utils.logging import getLogger 

61 

62log = getLogger(__name__) 

63 

64 

65@functools.lru_cache 

66def _calc_tmpdir_buffer_size(tmpdir: str) -> int: 

67 """Compute the block size to use for writing files in `tmpdir` as 

68 256 blocks of typical size (i.e. 4096 bytes) or 10 times the file system 

69 block size, whichever is higher. 

70 

71 This is a reasonable compromise between using memory for buffering and 

72 the number of system calls issued to read from or write to temporary 

73 files. 

74 """ 

75 fsstats = os.statvfs(tmpdir) 

76 return max(10 * fsstats.f_bsize, 256 * 4096) 

77 

78 

79class DavResourcePathConfig: 

80 """Configuration class to encapsulate the configurable items used by 

81 all instances of class `DavResourcePath`. 

82 

83 Instantiating this class creates a thread-safe singleton. 

84 """ 

85 

86 _instance = None 

87 _lock = threading.Lock() 

88 

89 def __new__(cls) -> DavResourcePathConfig: 

90 if cls._instance is None: 90 ↛ 95line 90 didn't jump to line 95 because the condition on line 90 was always true

91 with cls._lock: 

92 if cls._instance is None: 92 ↛ 95line 92 didn't jump to line 95

93 cls._instance = super().__new__(cls) 

94 

95 return cls._instance 

96 

97 def __init__(self) -> None: 

98 # Path to the local temporary directory all instances of 

99 # `DavResourcePath` must use and its associated buffer size (in bytes). 

100 self._tmpdir_buffersize: tuple[str, int] | None = None 

101 

102 @property 

103 def tmpdir_buffersize(self) -> tuple[str, int]: 

104 """Return the path to a temporary directory and the preferred buffer 

105 size to use when reading/writing files from/to that directory. 

106 """ 

107 if self._tmpdir_buffersize is not None: 107 ↛ 108line 107 didn't jump to line 108 because the condition on line 107 was never true

108 return self._tmpdir_buffersize 

109 

110 # Retrieve and cache the path and the blocksize for the temporary 

111 # directory if no other thread has done that in the meantime. 

112 with DavResourcePathConfig._lock: 

113 if self._tmpdir_buffersize is None: 113 ↛ 118line 113 didn't jump to line 118

114 tmpdir = get_tempdir() 

115 bufsize = _calc_tmpdir_buffer_size(tmpdir) 

116 self._tmpdir_buffersize = (tmpdir, bufsize) 

117 

118 return self._tmpdir_buffersize 

119 

120 def _destroy(self) -> None: 

121 """Destroy this class singleton instance. 

122 

123 Helper method to be used in tests to reset global configuration. 

124 """ 

125 with DavResourcePathConfig._lock: 

126 DavResourcePathConfig._instance = None 

127 

128 

129class DavGlobals: 

130 """Helper container to encapsulate all the gloal objects needed by this 

131 module. 

132 """ 

133 

134 def __init__(self) -> None: 

135 # Client pool used by all DavResourcePath instances. 

136 # Use Any as type annotation to keep mypy happy. 

137 self._client_pool: Any = None 

138 

139 # Configuration used by all DavResourcePath instances. 

140 self._config: Any = None 

141 

142 # (Re)Initialize the objects above. 

143 self._reset() 

144 

145 def _reset(self) -> None: 

146 """Initialize all the globals. 

147 

148 This method is a helper for reinitializing globals in tests. 

149 """ 

150 # Initialize the singleton instance of the webdav endpoint 

151 # configuration pool. 

152 config_pool: DavConfigPool = DavConfigPool("LSST_RESOURCES_WEBDAV_CONFIG") 

153 

154 # Initialize the singleton instance of the webdav client pool. This is 

155 # a thread-safe singleton shared by all instances of DavResourcePath. 

156 if self._client_pool is not None: 

157 self._client_pool._destroy() 

158 

159 self._client_pool = DavClientPool(config_pool) 

160 

161 # Initialize the singleton instance of the configuration shared 

162 # all DavResourcePath objects. 

163 if self._config is not None: 

164 self._config._destroy() 

165 

166 self._config = DavResourcePathConfig() 

167 

168 def client_pool(self) -> DavClientPool: 

169 """Return the pool of reusable webDAV clients.""" 

170 return self._client_pool 

171 

172 def config(self) -> DavResourcePathConfig: 

173 """Return the configuration settings for all `DavResourcePath` 

174 objects. 

175 """ 

176 return self._config 

177 

178 

179# Convenience object to encapsulate all global objects needed by this module. 

180dav_globals: DavGlobals = DavGlobals() 

181 

182 

183class DavResourcePath(ResourcePath): 

184 """WebDAV resource. 

185 

186 Parameters 

187 ---------- 

188 uri : `ResourcePathExpression` 

189 URI to store in object. 

190 root : `str` or `ResourcePath` or `None`, optional 

191 Root for relative URIs. Not used in this constructor. 

192 forceAbsolute : `bool` 

193 Whether to force absolute URI. A WebDAV URI is always absolute. 

194 forceDirectory : `bool` or `None`, optional 

195 Whether this URI represents a directory. 

196 isTemporary : `bool` or `None`, optional 

197 Whether this URI represents a temporary resource. 

198 """ 

199 

200 def __init__( 

201 self, 

202 uri: ResourcePathExpression, 

203 root: str | ResourcePath | None = None, 

204 forceAbsolute: bool = True, 

205 forceDirectory: bool | None = None, 

206 isTemporary: bool | None = None, 

207 ) -> None: 

208 # Build the internal URL we use to talk to the server, which 

209 # uses "http" or "https" as scheme instead of "dav" or "davs". 

210 self._internal_url: str = normalize_url(self.geturl()) 

211 

212 # WebDAV client this path must use to interact with the server. 

213 self._dav_client: DavClient | None = None 

214 

215 # Retrieve the configuration shared by all instances of this class. 

216 self._config: DavResourcePathConfig = dav_globals.config() 

217 

218 log.debug("created instance of DavResourcePath %s [%#x]", self, id(self)) 

219 

220 @classmethod 

221 def _fixupPathUri( 

222 cls, 

223 parsed: urllib.parse.ParseResult, 

224 root: ResourcePath | None = None, 

225 forceAbsolute: bool = False, 

226 forceDirectory: bool | None = None, 

227 ) -> tuple[urllib.parse.ParseResult, bool | None]: 

228 """Correct any issues with the supplied URI. 

229 

230 This function ensures that the path of the URI is normalized. 

231 """ 

232 # Call the superclass' _fixupPathUri. 

233 parsed, dirLike = super()._fixupPathUri(parsed, forceDirectory=forceDirectory) 

234 

235 # Clean the URL's path and ensure dir-like paths end by "/". 

236 path = normalize_path(parsed.path) 

237 if dirLike and path != "/": 

238 path += "/" 

239 

240 return parsed._replace(path=path), dirLike 

241 

242 @property 

243 def _client(self) -> DavClient: 

244 """Return the webDAV client for this resource.""" 

245 # If we already have a client, use it. 

246 if self._dav_client is not None: 

247 return self._dav_client 

248 

249 # Retrieve the client this resource must use to interact with the 

250 # server from the global client pool. 

251 self._dav_client = dav_globals.client_pool().get_client_for_url(self._internal_url) 

252 return self._dav_client 

253 

254 def _stat(self) -> DavFileMetadata: 

255 """Retrieve metadata about this resource.""" 

256 return self._client.stat(self._internal_url) 

257 

258 @override 

259 def mkdir(self) -> None: 

260 """Create the directory resource if it does not already exist.""" 

261 log.debug("mkdir %s [%#x]", self, id(self)) 

262 

263 if not self.isdir(): 

264 raise NotADirectoryError(f"Can not create a directory for file-like URI {self}") 

265 

266 stat = self._stat() 

267 if stat.is_dir: 

268 return 

269 

270 if stat.is_file: 

271 # A file exists at this path. 

272 raise NotADirectoryError( 

273 f"Can not create a directory for {self} because a file already exists at that URL" 

274 ) 

275 

276 # The underlying webDAV client will use the knowledge it has about 

277 # the specific server to create the requested directory 

278 # hierarchy by issueing the minimum possible number of requests. 

279 self._client.mkcol(self._internal_url) 

280 

281 @override 

282 def exists(self) -> bool: 

283 """Check that this resource exists.""" 

284 log.debug("exists %s [%#x]", self, id(self)) 

285 

286 return self._stat().exists 

287 

288 @override 

289 def size(self) -> int: 

290 """Return the size of the remote resource in bytes.""" 

291 log.debug("size %s [%#x]", self, id(self)) 

292 

293 return 0 if self.isdir() else self._client.size(self._internal_url) 

294 

295 @override 

296 def get_info(self) -> ResourceInfo: 

297 """Return lightweight metadata details about this resource.""" 

298 log.debug("get_info %s [%#x]", self, id(self)) 

299 

300 info = self._client.info(self._internal_url) 

301 if info["type"] is None: 

302 raise FileNotFoundError(f"Resource {self} does not exist") 

303 

304 return ResourceInfo( 

305 uri=str(self), 

306 is_file=info["type"] == "file", 

307 size=info["size"], 

308 last_modified=info["last_modified"], 

309 checksums=info["checksums"], 

310 ) 

311 

312 @override 

313 def read(self, size: int = -1) -> bytes: 

314 """Open the resource and return the contents in bytes. 

315 

316 Parameters 

317 ---------- 

318 size : `int`, optional 

319 The number of bytes to read. Negative or omitted indicates that 

320 all data should be read. 

321 """ 

322 log.debug("read %s [%#x] size=%d", self, id(self), size) 

323 

324 # A GET request on a dCache directory returns the contents of the 

325 # directory in HTML, to be visualized with a browser. This means 

326 # that we need to check first that this resource is not a directory. 

327 # 

328 # Since isdir() only checks that the URL of the resource ends in "/" 

329 # without actually asking the server, this check is not robust. 

330 # However, it is a reasonable compromise since it prevents doing 

331 # an additional roundtrip to the server to retrieve this resource's 

332 # metadata. 

333 if self.isdir(): 

334 raise ValueError(f"method read() is not implemented for directory {self}") 

335 

336 if size < 0: 

337 # Read the entire file content 

338 _, data = self._client.read(self._internal_url) 

339 return data 

340 

341 # This is a partial read. Retrieve the file size. 

342 stat = self._stat() 

343 if not stat.is_file: 

344 raise FileNotFoundError(f"No file found at {self}") 

345 

346 if size == 0 or stat.size == 0: 

347 return b"" 

348 

349 # Read the requested chunk of data and release the backend server. 

350 end_range = min(stat.size, size) - 1 

351 url, data = self._client.read_range(self._internal_url, start=0, end=end_range) 

352 self._client._close(url=url) 

353 return data 

354 

355 @override 

356 @contextlib.contextmanager 

357 def _as_local( 

358 self, multithreaded: bool = True, tmpdir: ResourcePath | None = None 

359 ) -> Generator[ResourcePath]: 

360 """Download object and place in temporary directory. 

361 

362 Parameters 

363 ---------- 

364 multithreaded : `bool`, optional 

365 If `True` the transfer will be allowed to attempt to improve 

366 throughput by using parallel download streams. This may of no 

367 effect if the URI scheme does not support parallel streams or 

368 if a global override has been applied. If `False` parallel 

369 streams will be disabled. 

370 tmpdir : `ResourcePath` or `None`, optional 

371 Explicit override of the temporary directory to use for remote 

372 downloads. 

373 

374 Returns 

375 ------- 

376 local_uri : `ResourcePath` 

377 A URI to a local POSIX file corresponding to a local temporary 

378 downloaded copy of the resource. 

379 """ 

380 log.debug("_as_local %s [%#x] tmpdir: %s", self, id(self), tmpdir) 

381 

382 # We need to ensure that this resource is actually a file since 

383 # the response to a GET request on a directory may be implemented in 

384 # several ways, according to RFC 4818. 

385 if self.isdir(): 

386 raise FileNotFoundError(f"{self} is a directory") 

387 

388 if tmpdir is None: 

389 local_dir, buffer_size = self._config.tmpdir_buffersize 

390 tmpdir = ResourcePath(local_dir, forceDirectory=True) 

391 else: 

392 buffer_size = _calc_tmpdir_buffer_size(tmpdir.ospath) 

393 

394 with ResourcePath.temporary_uri(suffix=self.getExtension(), prefix=tmpdir, delete=True) as tmp_uri: 

395 log.debug( 

396 "downloading %s [%#x] to local file %s [buffer_size %d]", 

397 self, 

398 id(self), 

399 tmp_uri.ospath, 

400 buffer_size, 

401 ) 

402 self._client.download(self._internal_url, tmp_uri.ospath, buffer_size) 

403 yield tmp_uri 

404 

405 @override 

406 def write(self, data: BinaryIO | bytes, overwrite: bool = True) -> None: 

407 """Write the supplied bytes to the new resource. 

408 

409 Parameters 

410 ---------- 

411 data : `bytes` 

412 The bytes to write to the resource. The entire contents of the 

413 resource will be replaced. 

414 overwrite : `bool`, optional 

415 If `True` the resource will be overwritten if it exists. Otherwise 

416 the write will fail. 

417 """ 

418 log.debug("write %s [%#x] overwrite=%s", self, id(self), overwrite) 

419 

420 if self.isdir(): 

421 raise ValueError(f"Method write() is not implemented for directory {self}") 

422 

423 if not overwrite and self._stat().is_file: 

424 raise FileExistsError(f"File {self} exists and overwrite has been disabled") 

425 

426 self._client.write(self._internal_url, data) 

427 

428 @override 

429 def remove(self) -> None: 

430 """Remove the resource. 

431 

432 If the resource is a directory, it must be empty otherwise this 

433 method raises. Removing a non-existent file or directory is not 

434 considered an error. 

435 """ 

436 log.debug("remove %s [%#x]", self, id(self)) 

437 

438 stat = self._stat() 

439 if not stat.exists: 

440 # There is no resource at this uri. There is nothing to do. 

441 return 

442 

443 if stat.is_dir: 

444 entries = self._client.read_dir(self._internal_url) 

445 if len(entries) > 0: 

446 raise IsADirectoryError(f"Directory {self} is not empty") 

447 

448 # This resource is a either file or an empty directory, we can remove 

449 # it. 

450 self._client.delete(self._internal_url) 

451 

452 def remove_dir(self, recursive: bool = False) -> None: 

453 """Remove a directory if empty. 

454 

455 Parameters 

456 ---------- 

457 recursive : `bool` 

458 If `True` recursively remove all files and directories under this 

459 directory. 

460 

461 Notes 

462 ----- 

463 This method is not present in the superclass. 

464 """ 

465 log.debug("remove_dir %s [%#x] recursive=%s", self, id(self), recursive) 

466 

467 if not self.isdir(): 

468 raise NotADirectoryError(f"{self} is not a directory") 

469 

470 for root, subdirs, files in self.walk(): 

471 if not recursive and (len(subdirs) > 0 or len(files) > 0): 

472 raise IsADirectoryError(f"Directory at {self} is not empty and recursive argument is False") 

473 

474 for file in files: 

475 root.join(file).remove() 

476 

477 for subdir in subdirs: 

478 child = DavResourcePath(root.join(subdir, forceDirectory=True)) 

479 # ResourcePath.__new__ is a scheme-dispatching factory 

480 # declared as returning the base class; ty honors that 

481 # declaration and mypy does not. 

482 child.remove_dir(recursive=recursive) # ty: ignore[unresolved-attribute] 

483 

484 # Remove empty top directory 

485 self.remove() 

486 

487 @override 

488 def transfer_from( 

489 self, 

490 src: ResourcePath, 

491 transfer: str = "copy", 

492 overwrite: bool = False, 

493 transaction: TransactionProtocol | None = None, 

494 multithreaded: bool = True, 

495 ) -> None: 

496 """Transfer to this URI from another. 

497 

498 Parameters 

499 ---------- 

500 src : `ResourcePath` 

501 Source URI. 

502 transfer : `str` 

503 Mode to use for transferring the resource. Generically there are 

504 many standard options: copy, link, symlink, hardlink, relsymlink. 

505 Not all URIs support all modes. 

506 overwrite : `bool`, optional 

507 Allow an existing file to be overwritten. Defaults to `False`. 

508 transaction : `~lsst.resources.utils.TransactionProtocol`, optional 

509 A transaction object that can (depending on implementation) 

510 rollback transfers on error. Not guaranteed to be implemented. 

511 multithreaded : `bool`, optional 

512 If `True` the transfer will be allowed to attempt to improve 

513 throughput by using parallel download streams. This may of no 

514 effect if the URI scheme does not support parallel streams or 

515 if a global override has been applied. If `False` parallel 

516 streams will be disabled. 

517 """ 

518 log.debug( 

519 "transfer_from %s [%#x] src=%s transfer=%s overwrite=%s", 

520 self, 

521 id(self), 

522 src, 

523 transfer, 

524 overwrite, 

525 ) 

526 

527 # Fail early to prevent delays if remote resources are requested. 

528 if transfer not in self.transferModes: 

529 raise ValueError(f"Transfer mode {transfer} not supported by URI scheme {self.scheme}") 

530 

531 # Existence checks cost time so do not call this unless we know 

532 # that debugging is enabled. 

533 destination_exists = None 

534 if log.isEnabledFor(logging.DEBUG): 

535 destination_exists = self.exists() 

536 log.debug( 

537 "Transferring %s [exists: %s] -> %s [exists: %s] (transfer=%s)", 

538 src.geturl(), 

539 src.exists(), 

540 self, 

541 destination_exists, 

542 transfer, 

543 ) 

544 

545 # Short circuit immediately if the URIs are identical. 

546 if self == src: 

547 log.debug( 

548 "Target and destination URIs are identical: %s, returning immediately." 

549 " No further action required.", 

550 self, 

551 ) 

552 return 

553 

554 if not overwrite: 

555 if destination_exists is None: 

556 destination_exists = self.exists() 

557 

558 if destination_exists: 

559 raise FileExistsError(f"Destination path {self} already exists.") 

560 

561 if transfer == "auto": 

562 transfer = self.transferDefault 

563 

564 # We can use webDAV 'COPY' or 'MOVE' if both the current and source 

565 # resources are located in the same server. 

566 if isinstance(src, type(self)) and self.root_uri() == src.root_uri(): 

567 log.debug("Transfer from %s to %s [%#x] directly", src, self, id(self)) 

568 return ( 

569 self._move_from(src, overwrite=overwrite) 

570 if transfer == "move" 

571 else self._copy_from(src, overwrite=overwrite) 

572 ) 

573 

574 # For resources of different classes we can perform the copy or move 

575 # operation by downloading to a local file and uploading to the 

576 # destination. 

577 self._copy_via_local(src) 

578 

579 # This was an explicit move, try to remove the source. 

580 if transfer == "move": 

581 src.remove() 

582 

583 def _copy_via_local(self, source: ResourcePath) -> None: 

584 """Replace the contents of this resource with the contents of a remote 

585 resource by using a local temporary file. 

586 

587 Parameters 

588 ---------- 

589 source : `ResourcePath` 

590 The source of the contents to copy to `self`. 

591 """ 

592 with source.as_local() as local_uri: 

593 log.debug( 

594 "Transfer from %s to %s [%#x] via local file %s", 

595 source.geturl(), 

596 self, 

597 id(self), 

598 local_uri, 

599 ) 

600 with open(local_uri.ospath, "rb") as f: 

601 self.write(data=f) 

602 

603 def _copy_from(self, source: DavResourcePath, overwrite: bool = False) -> None: 

604 """Copy the contents of `source` to this resource. `source` must 

605 be a file. 

606 """ 

607 log.debug("_copy_from %s [%#x] source=%s overwrite=%s", self, id(self), source, overwrite) 

608 

609 # Copy is only supported for files, not directories. 

610 if self.isdir(): 

611 raise ValueError(f"Copy is not supported because destination {self} is a directory") 

612 

613 if source.isdir(): 

614 raise ValueError(f"Copy is not supported for directory {source}") 

615 

616 if not source.exists(): 

617 raise FileNotFoundError(f"No file found at {source}") 

618 

619 # If the server supports file duplication, use that method. 

620 if self._client.supports_duplicate: 

621 return self._client.duplicate(source._internal_url, self._internal_url, overwrite) 

622 

623 # Make this copy via a local file 

624 if not overwrite and self.exists(): 

625 raise FileExistsError(f"Destination path {self} already exists.") 

626 

627 self._copy_via_local(source) 

628 

629 def _move_from(self, source: DavResourcePath, overwrite: bool = False) -> None: 

630 """Send a MOVE webDAV request to replace the contents of this resource 

631 with the contents of another resource located in the same server. 

632 

633 Parameters 

634 ---------- 

635 source : `DavResourcePath` 

636 The source of the contents to move to `self`. 

637 """ 

638 log.debug("_move_from %s [%#x] source=%s overwrite=%s", self, id(self), source, overwrite) 

639 

640 # Move is only supported for files, not directories. 

641 if self.isdir(): 

642 raise ValueError(f"Move is not supported for destination directory {self}") 

643 

644 if source.isdir(): 

645 raise ValueError(f"Move is not supported for directory {source}") 

646 

647 if not source.exists(): 

648 raise FileNotFoundError(f"No file found at {source}") 

649 

650 self._client.rename(source._internal_url, self._internal_url, overwrite) 

651 

652 @override 

653 def walk( 

654 self, file_filter: str | re.Pattern | None = None 

655 ) -> Iterator[list | tuple[ResourcePath, list[str], list[str]]]: 

656 """Walk the directory tree returning matching files and directories. 

657 

658 Parameters 

659 ---------- 

660 file_filter : `str` or `re.Pattern`, optional 

661 Regex to filter out files from the list before it is returned. 

662 

663 Yields 

664 ------ 

665 dirpath : `ResourcePath` 

666 Current directory being examined. 

667 dirnames : `list` of `str` 

668 Names of subdirectories within dirpath. 

669 filenames : `list` of `str` 

670 Names of all the files within dirpath. 

671 """ 

672 if not self.isdir(): 

673 raise ValueError(f"Can not walk non-directory URI {self}") 

674 

675 # We must return no entries for non-existent directories. 

676 if not self.exists(): 

677 return 

678 

679 # Retrieve the entries in this directory 

680 entries = self._client.read_dir(self._internal_url) 

681 files = [e.name for e in entries if e.is_file] 

682 subdirs = [e.name for e in entries if e.is_dir] 

683 

684 # Filter files 

685 if isinstance(file_filter, str): 

686 file_filter = re.compile(file_filter) 

687 

688 if file_filter is not None: 

689 files = [f for f in files if file_filter.search(f)] 

690 

691 if not subdirs and not files: 

692 return 

693 else: 

694 yield type(self)(self, forceAbsolute=False, forceDirectory=True), subdirs, files 

695 

696 for subdir in subdirs: 

697 new_uri = self.join(subdir, forceDirectory=True) 

698 yield from new_uri.walk(file_filter) 

699 

700 @override 

701 def generate_presigned_get_url(self, *, expiration_time_seconds: int) -> str: 

702 """Return a pre-signed URL that can be used to retrieve this resource 

703 using an HTTP GET without supplying any access credentials. 

704 

705 Parameters 

706 ---------- 

707 expiration_time_seconds : `int` 

708 Number of seconds until the generated URL is no longer valid. 

709 

710 Returns 

711 ------- 

712 url : `str` 

713 HTTP URL signed for GET. 

714 """ 

715 return self._client.generate_presigned_get_url(self._internal_url, expiration_time_seconds) 

716 

717 @override 

718 def generate_presigned_put_url(self, *, expiration_time_seconds: int) -> str: 

719 """Return a pre-signed URL that can be used to upload a file to this 

720 path using an HTTP PUT without supplying any access credentials. 

721 

722 Parameters 

723 ---------- 

724 expiration_time_seconds : `int` 

725 Number of seconds until the generated URL is no longer valid. 

726 

727 Returns 

728 ------- 

729 url : `str` 

730 HTTP URL signed for PUT. 

731 """ 

732 return self._client.generate_presigned_put_url(self._internal_url, expiration_time_seconds) 

733 

734 @override 

735 def to_fsspec(self) -> tuple[DavFileSystem, str]: 

736 """Return an abstract file system and path that can be used by fsspec. 

737 

738 Returns 

739 ------- 

740 fs : `fsspec.spec.AbstractFileSystem` 

741 A file system object suitable for use with the returned path. 

742 path : `str` 

743 A path that can be opened by the file system object. 

744 """ 

745 if fsspec is None or not self._client._config.enable_fsspec: 

746 raise ImportError("fsspec is not available") 

747 

748 log.debug("DavResourcePath.to_fsspec: %s", self) 

749 fsys = DavFileSystem(self) 

750 return fsys, fsys._path 

751 

752 @override 

753 @contextlib.contextmanager 

754 def _openImpl( 

755 self, 

756 mode: str = "r", 

757 *, 

758 encoding: str | None = None, 

759 ) -> Generator[ResourceHandleProtocol]: 

760 log.debug("DavResourcePath._openImpl: %s mode: %s", self, mode) 

761 

762 if mode in ("rb", "r") and self._client.accepts_ranges(self._internal_url): 

763 stat = self._stat() 

764 if stat.is_dir: 

765 raise OSError(f"open is not implemented for directory {self}") 

766 

767 if not stat.is_file: 

768 raise FileNotFoundError(f"No such file {self}") 

769 

770 with DavReadResourceHandle(mode, log.logger, uri=self, file_size=stat.size) as handle: 

771 if mode == "r": 

772 # cast because the protocol is compatible, but does not 

773 # have BytesIO in the inheritance tree 

774 yield io.TextIOWrapper(cast(Any, handle), encoding=encoding) 

775 else: 

776 yield handle 

777 else: 

778 with super()._openImpl(mode, encoding=encoding) as handle: 

779 yield handle 

780 

781 

782class DavFileSystem(AbstractFileSystem): 

783 """Minimal fsspec-compatible read-only file system which contains a single 

784 file. 

785 

786 Parameters 

787 ---------- 

788 uri : `DavResourcePath` 

789 URI of the single resource contained in the file system. 

790 """ 

791 

792 protocol = ("davs", "dav") 

793 

794 def __init__(self, uri: DavResourcePath): 

795 super().__init__() 

796 self._uri: DavResourcePath = uri 

797 self._path: str = self._uri.geturl() 

798 self._size: int | None = None 

799 

800 @override 

801 def info(self, path: str, **kwargs: Any) -> dict[str, Any]: 

802 log.debug("DavFileSystem.info %s", path) 

803 if path != self._path: 

804 raise FileNotFoundError(path) 

805 

806 return { 

807 "name": path, 

808 "size": self.size(self._path), 

809 "type": "file", 

810 } 

811 

812 @override 

813 def ls(self, path: str, detail: bool = True, **kwargs: Any) -> list[str] | list[dict[str, str]]: 

814 log.debug("DavFileSystem.ls %s", path) 

815 if path != self._path: 

816 raise FileNotFoundError(path) 

817 

818 return list(self.info(path)) if detail else list(path) 

819 

820 @override 

821 def modified(self, path: str) -> datetime.datetime: 

822 log.debug("DavFileSystem.modified %s", path) 

823 if path != self._path: 

824 raise FileNotFoundError(path) 

825 

826 return self._uri._stat().last_modified 

827 

828 @override 

829 def size(self, path: str) -> int: 

830 log.debug("DavFileSystem.size %s", path) 

831 if path != self._path: 

832 raise FileNotFoundError(path) 

833 

834 if self._size is None: 

835 self._size = self._uri.size() 

836 

837 return self._size 

838 

839 @override 

840 def isfile(self, path: str) -> bool: 

841 log.debug("DavFileSystem.isfile %s", path) 

842 return path == self._path 

843 

844 @override 

845 def isdir(self, path: str) -> bool: 

846 log.debug("DavFileSystem.isdir %s", path) 

847 return False 

848 

849 @override 

850 def exists(self, path: str, **kwargs: Any) -> bool: 

851 log.debug("DavFileSystem.exists %s", path) 

852 return path == self._path 

853 

854 @override 

855 def open( 

856 self, 

857 path: str, 

858 mode: str = "rb", 

859 block_size: int | None = None, 

860 cache_options: dict[Any, Any] | None = None, 

861 compression: str | None = None, 

862 *, 

863 encoding: str | None = None, 

864 **kwargs: Any, 

865 ) -> DavReadResourceHandle | io.TextIOWrapper: 

866 log.debug( 

867 "DavFileSystem.open path: %s mode: %s encoding: %s blocksize: %s", 

868 path, 

869 mode, 

870 encoding, 

871 block_size, 

872 ) 

873 if path != self._path: 

874 raise FileNotFoundError(f"File {path} does not exist") 

875 

876 if mode not in ("rb", "r"): 

877 raise OSError(f"Opening {path} for writing is not supported") 

878 

879 handle = DavReadResourceHandle(mode, log.logger, self._uri, self.size(self._path)) 

880 if mode == "rb": 

881 return handle 

882 else: 

883 return io.TextIOWrapper(cast(Any, handle), encoding=encoding) 

884 

885 @property 

886 def fsid(self) -> Any: 

887 return "davs" 

888 

889 @override 

890 def mkdir(self, path: str, create_parents: bool = True, **kwargs: Any) -> None: 

891 raise NotImplementedError 

892 

893 @override