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 threading 

24import urllib 

25from collections.abc import Iterator 

26from typing import TYPE_CHECKING, Any, BinaryIO, cast 

27 

28try: 

29 from typing import override # Python 3.12+ 

30except ImportError: 

31 from typing_extensions import override # Python 3.11 

32 

33try: 

34 import fsspec 

35 from fsspec.spec import AbstractFileSystem 

36except ImportError: 

37 fsspec = None 

38 AbstractFileSystem = type 

39 

40from ._resourceHandles import ResourceHandleProtocol 

41from ._resourceHandles._davResourceHandle import DavReadResourceHandle 

42from ._resourcePath import ResourceInfo, ResourcePath, ResourcePathExpression 

43from .davutils import ( 

44 DavClient, 

45 DavClientPool, 

46 DavConfigPool, 

47 DavFileMetadata, 

48 normalize_path, 

49 normalize_url, 

50) 

51from .utils import get_tempdir 

52 

53if TYPE_CHECKING: 

54 from .utils import TransactionProtocol 

55 

56from lsst.utils.logging import getLogger 

57 

58log = getLogger(__name__) 

59 

60 

61@functools.lru_cache 

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

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

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

65 block size, whichever is higher. 

66 

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

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

69 files. 

70 """ 

71 fsstats = os.statvfs(tmpdir) 

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

73 

74 

75class DavResourcePathConfig: 

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

77 all instances of class `DavResourcePath`. 

78 

79 Instantiating this class creates a thread-safe singleton. 

80 """ 

81 

82 _instance = None 

83 _lock = threading.Lock() 

84 

85 def __new__(cls) -> DavResourcePathConfig: 

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

87 with cls._lock: 

88 if cls._instance is None: 88 ↛ 91line 88 didn't jump to line 91

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

90 

91 return cls._instance 

92 

93 def __init__(self) -> None: 

94 # Path to the local temporary directory all instances of 

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

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

97 

98 @property 

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

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

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

102 """ 

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

104 return self._tmpdir_buffersize 

105 

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

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

108 with DavResourcePathConfig._lock: 

109 if self._tmpdir_buffersize is None: 109 ↛ 114line 109 didn't jump to line 114

110 tmpdir = get_tempdir() 

111 bufsize = _calc_tmpdir_buffer_size(tmpdir) 

112 self._tmpdir_buffersize = (tmpdir, bufsize) 

113 

114 return self._tmpdir_buffersize 

115 

116 def _destroy(self) -> None: 

117 """Destroy this class singleton instance. 

118 

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

120 """ 

121 with DavResourcePathConfig._lock: 

122 DavResourcePathConfig._instance = None 

123 

124 

125class DavGlobals: 

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

127 module. 

128 """ 

129 

130 def __init__(self) -> None: 

131 # Client pool used by all DavResourcePath instances. 

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

133 self._client_pool: Any = None 

134 

135 # Configuration used by all DavResourcePath instances. 

136 self._config: Any = None 

137 

138 # (Re)Initialize the objects above. 

139 self._reset() 

140 

141 def _reset(self) -> None: 

142 """Initialize all the globals. 

143 

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

145 """ 

146 # Initialize the singleton instance of the webdav endpoint 

147 # configuration pool. 

148 config_pool: DavConfigPool = DavConfigPool("LSST_RESOURCES_WEBDAV_CONFIG") 

149 

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

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

152 if self._client_pool is not None: 

153 self._client_pool._destroy() 

154 

155 self._client_pool = DavClientPool(config_pool) 

156 

157 # Initialize the singleton instance of the configuration shared 

158 # all DavResourcePath objects. 

159 if self._config is not None: 

160 self._config._destroy() 

161 

162 self._config = DavResourcePathConfig() 

163 

164 def client_pool(self) -> DavClientPool: 

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

166 return self._client_pool 

167 

168 def config(self) -> DavResourcePathConfig: 

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

170 objects. 

171 """ 

172 return self._config 

173 

174 

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

176dav_globals: DavGlobals = DavGlobals() 

177 

178 

179class DavResourcePath(ResourcePath): 

180 """WebDAV resource. 

181 

182 Parameters 

183 ---------- 

184 uri : `ResourcePathExpression` 

185 URI to store in object. 

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

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

188 forceAbsolute : `bool` 

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

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

191 Whether this URI represents a directory. 

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

193 Whether this URI represents a temporary resource. 

194 """ 

195 

196 def __init__( 

197 self, 

198 uri: ResourcePathExpression, 

199 root: str | ResourcePath | None = None, 

200 forceAbsolute: bool = True, 

201 forceDirectory: bool | None = None, 

202 isTemporary: bool | None = None, 

203 ) -> None: 

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

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

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

207 

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

209 self._dav_client: DavClient | None = None 

210 

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

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

213 

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

215 

216 @classmethod 

217 def _fixupPathUri( 

218 cls, 

219 parsed: urllib.parse.ParseResult, 

220 root: ResourcePath | None = None, 

221 forceAbsolute: bool = False, 

222 forceDirectory: bool | None = None, 

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

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

225 

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

227 """ 

228 # Call the superclass' _fixupPathUri. 

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

230 

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

232 path = normalize_path(parsed.path) 

233 if dirLike and path != "/": 

234 path += "/" 

235 

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

237 

238 @property 

239 def _client(self) -> DavClient: 

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

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

242 if self._dav_client is not None: 

243 return self._dav_client 

244 

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

246 # server from the global client pool. 

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

248 return self._dav_client 

249 

250 def _stat(self) -> DavFileMetadata: 

251 """Retrieve metadata about this resource.""" 

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

253 

254 @override 

255 def mkdir(self) -> None: 

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

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

258 

259 if not self.isdir(): 

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

261 

262 stat = self._stat() 

263 if stat.is_dir: 

264 return 

265 

266 if stat.is_file: 

267 # A file exists at this path. 

268 raise NotADirectoryError( 

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

270 ) 

271 

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

273 # the specific server to create the requested directory 

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

275 self._client.mkcol(self._internal_url) 

276 

277 @override 

278 def exists(self) -> bool: 

279 """Check that this resource exists.""" 

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

281 

282 return self._stat().exists 

283 

284 @override 

285 def size(self) -> int: 

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

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

288 

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

290 

291 @override 

292 def get_info(self) -> ResourceInfo: 

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

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

295 

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

297 if info["type"] is None: 

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

299 

300 return ResourceInfo( 

301 uri=str(self), 

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

303 size=info["size"], 

304 last_modified=info["last_modified"], 

305 checksums=info["checksums"], 

306 ) 

307 

308 @override 

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

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

311 

312 Parameters 

313 ---------- 

314 size : `int`, optional 

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

316 all data should be read. 

317 """ 

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

319 

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

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

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

323 # 

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

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

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

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

328 # metadata. 

329 if self.isdir(): 

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

331 

332 if size < 0: 

333 # Read the entire file content 

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

335 return data 

336 

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

338 stat = self._stat() 

339 if not stat.is_file: 

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

341 

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

343 return b"" 

344 

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

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

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

348 self._client._close(url=url) 

349 return data 

350 

351 @override 

352 @contextlib.contextmanager 

353 def _as_local( 

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

355 ) -> Iterator[ResourcePath]: 

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

357 

358 Parameters 

359 ---------- 

360 multithreaded : `bool`, optional 

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

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

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

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

365 streams will be disabled. 

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

367 Explicit override of the temporary directory to use for remote 

368 downloads. 

369 

370 Returns 

371 ------- 

372 local_uri : `ResourcePath` 

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

374 downloaded copy of the resource. 

375 """ 

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

377 

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

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

380 # several ways, according to RFC 4818. 

381 if self.isdir(): 

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

383 

384 if tmpdir is None: 

385 local_dir, buffer_size = self._config.tmpdir_buffersize 

386 tmpdir = ResourcePath(local_dir, forceDirectory=True) 

387 else: 

388 buffer_size = _calc_tmpdir_buffer_size(tmpdir.ospath) 

389 

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

391 log.debug( 

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

393 self, 

394 id(self), 

395 tmp_uri.ospath, 

396 buffer_size, 

397 ) 

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

399 yield tmp_uri 

400 

401 @override 

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

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

404 

405 Parameters 

406 ---------- 

407 data : `bytes` 

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

409 resource will be replaced. 

410 overwrite : `bool`, optional 

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

412 the write will fail. 

413 """ 

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

415 

416 if self.isdir(): 

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

418 

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

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

421 

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

423 

424 @override 

425 def remove(self) -> None: 

426 """Remove the resource. 

427 

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

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

430 considered an error. 

431 """ 

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

433 

434 stat = self._stat() 

435 if not stat.exists: 

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

437 return 

438 

439 if stat.is_dir: 

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

441 if len(entries) > 0: 

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

443 

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

445 # it. 

446 self._client.delete(self._internal_url) 

447 

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

449 """Remove a directory if empty. 

450 

451 Parameters 

452 ---------- 

453 recursive : `bool` 

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

455 directory. 

456 

457 Notes 

458 ----- 

459 This method is not present in the superclass. 

460 """ 

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

462 

463 if not self.isdir(): 

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

465 

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

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

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

469 

470 for file in files: 

471 root.join(file).remove() 

472 

473 for subdir in subdirs: 

474 DavResourcePath(root.join(subdir, forceDirectory=True)).remove_dir(recursive=recursive) 

475 

476 # Remove empty top directory 

477 self.remove() 

478 

479 @override 

480 def transfer_from( 

481 self, 

482 src: ResourcePath, 

483 transfer: str = "copy", 

484 overwrite: bool = False, 

485 transaction: TransactionProtocol | None = None, 

486 multithreaded: bool = True, 

487 ) -> None: 

488 """Transfer to this URI from another. 

489 

490 Parameters 

491 ---------- 

492 src : `ResourcePath` 

493 Source URI. 

494 transfer : `str` 

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

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

497 Not all URIs support all modes. 

498 overwrite : `bool`, optional 

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

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

501 A transaction object that can (depending on implementation) 

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

503 multithreaded : `bool`, optional 

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

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

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

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

508 streams will be disabled. 

509 """ 

510 log.debug( 

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

512 self, 

513 id(self), 

514 src, 

515 transfer, 

516 overwrite, 

517 ) 

518 

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

520 if transfer not in self.transferModes: 

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

522 

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

524 # that debugging is enabled. 

525 destination_exists = None 

526 if log.isEnabledFor(logging.DEBUG): 

527 destination_exists = self.exists() 

528 log.debug( 

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

530 src.geturl(), 

531 src.exists(), 

532 self, 

533 destination_exists, 

534 transfer, 

535 ) 

536 

537 # Short circuit immediately if the URIs are identical. 

538 if self == src: 

539 log.debug( 

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

541 " No further action required.", 

542 self, 

543 ) 

544 return 

545 

546 if not overwrite: 

547 if destination_exists is None: 

548 destination_exists = self.exists() 

549 

550 if destination_exists: 

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

552 

553 if transfer == "auto": 

554 transfer = self.transferDefault 

555 

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

557 # resources are located in the same server. 

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

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

560 return ( 

561 self._move_from(src, overwrite=overwrite) 

562 if transfer == "move" 

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

564 ) 

565 

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

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

568 # destination. 

569 self._copy_via_local(src) 

570 

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

572 if transfer == "move": 

573 src.remove() 

574 

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

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

577 resource by using a local temporary file. 

578 

579 Parameters 

580 ---------- 

581 source : `ResourcePath` 

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

583 """ 

584 with source.as_local() as local_uri: 

585 log.debug( 

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

587 source.geturl(), 

588 self, 

589 id(self), 

590 local_uri, 

591 ) 

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

593 self.write(data=f) 

594 

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

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

597 be a file. 

598 """ 

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

600 

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

602 if self.isdir(): 

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

604 

605 if source.isdir(): 

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

607 

608 if not source.exists(): 

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

610 

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

612 if self._client.supports_duplicate: 

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

614 

615 # Make this copy via a local file 

616 if not overwrite and self.exists(): 

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

618 

619 self._copy_via_local(source) 

620 

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

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

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

624 

625 Parameters 

626 ---------- 

627 source : `DavResourcePath` 

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

629 """ 

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

631 

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

633 if self.isdir(): 

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

635 

636 if source.isdir(): 

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

638 

639 if not source.exists(): 

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

641 

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

643 

644 @override 

645 def walk( 

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

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

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

649 

650 Parameters 

651 ---------- 

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

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

654 

655 Yields 

656 ------ 

657 dirpath : `ResourcePath` 

658 Current directory being examined. 

659 dirnames : `list` of `str` 

660 Names of subdirectories within dirpath. 

661 filenames : `list` of `str` 

662 Names of all the files within dirpath. 

663 """ 

664 if not self.isdir(): 

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

666 

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

668 if not self.exists(): 

669 return 

670 

671 # Retrieve the entries in this directory 

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

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

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

675 

676 # Filter files 

677 if isinstance(file_filter, str): 

678 file_filter = re.compile(file_filter) 

679 

680 if file_filter is not None: 

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

682 

683 if not subdirs and not files: 

684 return 

685 else: 

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

687 

688 for subdir in subdirs: 

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

690 yield from new_uri.walk(file_filter) 

691 

692 @override 

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

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

695 using an HTTP GET without supplying any access credentials. 

696 

697 Parameters 

698 ---------- 

699 expiration_time_seconds : `int` 

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

701 

702 Returns 

703 ------- 

704 url : `str` 

705 HTTP URL signed for GET. 

706 """ 

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

708 

709 @override 

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

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

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

713 

714 Parameters 

715 ---------- 

716 expiration_time_seconds : `int` 

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

718 

719 Returns 

720 ------- 

721 url : `str` 

722 HTTP URL signed for PUT. 

723 """ 

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

725 

726 @override 

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

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

729 

730 Returns 

731 ------- 

732 fs : `fsspec.spec.AbstractFileSystem` 

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

734 path : `str` 

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

736 """ 

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

738 raise ImportError("fsspec is not available") 

739 

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

741 fsys = DavFileSystem(self) 

742 return fsys, fsys._path 

743 

744 @override 

745 @contextlib.contextmanager 

746 def _openImpl( 

747 self, 

748 mode: str = "r", 

749 *, 

750 encoding: str | None = None, 

751 ) -> Iterator[ResourceHandleProtocol]: 

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

753 

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

755 stat = self._stat() 

756 if stat.is_dir: 

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

758 

759 if not stat.is_file: 

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

761 

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

763 if mode == "r": 

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

765 # have BytesIO in the inheritance tree 

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

767 else: 

768 yield handle 

769 else: 

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

771 yield handle 

772 

773 

774class DavFileSystem(AbstractFileSystem): 

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

776 file. 

777 

778 Parameters 

779 ---------- 

780 uri : `DavResourcePath` 

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

782 """ 

783 

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

785 

786 def __init__(self, uri: DavResourcePath): 

787 super().__init__() 

788 self._uri: DavResourcePath = uri 

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

790 self._size: int | None = None 

791 

792 @override 

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

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

795 if path != self._path: 

796 raise FileNotFoundError(path) 

797 

798 return { 

799 "name": path, 

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

801 "type": "file", 

802 } 

803 

804 @override 

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

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

807 if path != self._path: 

808 raise FileNotFoundError(path) 

809 

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

811 

812 @override 

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

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

815 if path != self._path: 

816 raise FileNotFoundError(path) 

817 

818 return self._uri._stat().last_modified 

819 

820 @override 

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

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

823 if path != self._path: 

824 raise FileNotFoundError(path) 

825 

826 if self._size is None: 

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

828 

829 return self._size 

830 

831 @override 

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

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

834 return path == self._path 

835 

836 @override 

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

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

839 return False 

840 

841 @override 

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

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

844 return path == self._path 

845 

846 @override 

847 def open( 

848 self, 

849 path: str, 

850 mode: str = "rb", 

851 encoding: str | None = None, 

852 block_size: int | None = None, 

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

854 compression: str | None = None, 

855 **kwargs: Any, 

856 ) -> DavReadResourceHandle | io.TextIOWrapper: 

857 log.debug( 

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

859 path, 

860 mode, 

861 encoding, 

862 block_size, 

863 ) 

864 if path != self._path: 

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

866 

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

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

869 

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

871 if mode == "rb": 

872 return handle 

873 else: 

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

875 

876 @property 

877 def fsid(self) -> Any: 

878 return "davs" 

879 

880 @override 

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

882 raise NotImplementedError 

883 

884 @override 

885 def makedirs(self, path: str, exist_ok: bool = False) -> None: 

886 raise NotImplementedError 

887 

888 @override 

889 def rmdir(self, path: str) -> None: 

890 raise NotImplementedError 

891