Coverage for python/lsst/resources/file.py: 90%

227 statements  

« prev     ^ index     » next       coverage.py v7.11.0, created at 2025-11-19 01:29 -0800

1# This file is part of lsst-resources. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# Use of this source code is governed by a 3-clause BSD-style 

10# license that can be found in the LICENSE file. 

11 

12from __future__ import annotations 

13 

14__all__ = ("FileResourcePath",) 

15 

16import contextlib 

17import copy 

18import logging 

19import os 

20import os.path 

21import posixpath 

22import re 

23import shutil 

24import stat 

25import urllib.parse 

26from collections.abc import Iterator 

27from typing import IO, TYPE_CHECKING 

28 

29from ._resourceHandles._fileResourceHandle import FileResourceHandle 

30from ._resourcePath import ResourcePath 

31from .utils import NoTransaction, ensure_directory_is_writeable, os2posix, posix2os 

32 

33try: 

34 import fsspec 

35 from fsspec.spec import AbstractFileSystem 

36except ImportError: 

37 fsspec = None 

38 AbstractFileSystem = type 

39 

40if TYPE_CHECKING: 

41 from .utils import TransactionProtocol 

42 

43 

44log = logging.getLogger(__name__) 

45 

46 

47class FileResourcePath(ResourcePath): 

48 """Path for explicit ``file`` URI scheme.""" 

49 

50 transferModes = ("copy", "link", "symlink", "hardlink", "relsymlink", "auto", "move") 

51 transferDefault: str = "link" 

52 

53 # By definition refers to a local file 

54 isLocal = True 

55 

56 @property 

57 def ospath(self) -> str: 

58 """Path component of the URI localized to current OS. 

59 

60 Will unquote URI path since a formal URI must include the quoting. 

61 """ 

62 return urllib.parse.unquote(posix2os(self._uri.path)) 

63 

64 def exists(self) -> bool: 

65 """Indicate that the file exists.""" 

66 # Uses os.path.exists so if there is a soft link that points 

67 # to a file that no longer exists this will return False 

68 return os.path.exists(self.ospath) 

69 

70 def size(self) -> int: 

71 """Return the size of the file in bytes.""" 

72 if not os.path.isdir(self.ospath): 

73 stat = os.stat(self.ospath) 

74 sz = stat.st_size 

75 else: 

76 sz = 0 

77 return sz 

78 

79 def remove(self) -> None: 

80 """Remove the resource.""" 

81 os.remove(self.ospath) 

82 

83 @contextlib.contextmanager 

84 def _as_local( 

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

86 ) -> Iterator[ResourcePath]: 

87 """Return the local path of the file. 

88 

89 This is an internal helper for ``as_local()``. 

90 

91 Parameters 

92 ---------- 

93 multithreaded : `bool`, optional 

94 Unused. 

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

96 Unused. 

97 

98 Returns 

99 ------- 

100 local_uri : `ResourcePath` 

101 A local URI. In this case it will be itself. 

102 """ 

103 yield self 

104 

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

106 with open(self.ospath, "rb") as fh: 

107 return fh.read(size) 

108 

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

110 dir = os.path.dirname(self.ospath) 

111 if dir and not os.path.exists(dir): 

112 _create_directories(dir) 

113 mode = "wb" if overwrite else "xb" 

114 with open(self.ospath, mode) as f: 

115 f.write(data) 

116 

117 def mkdir(self) -> None: 

118 """Make the directory associated with this URI. 

119 

120 An attempt will be made to create the directory even if the URI 

121 looks like a file. 

122 

123 Raises 

124 ------ 

125 NotADirectoryError: 

126 Raised if a non-directory already exists. 

127 """ 

128 try: 

129 _create_directories(self.ospath) 

130 except FileExistsError: 

131 raise NotADirectoryError(f"{self.ospath} exists but is not a directory.") from None 

132 

133 def isdir(self) -> bool: 

134 """Return whether this URI is a directory. 

135 

136 Returns 

137 ------- 

138 isdir : `bool` 

139 `True` if this URI is a directory or looks like a directory, 

140 else `False`. 

141 """ 

142 if self.dirLike is None: 

143 # Cache state for next time. 

144 self.dirLike = os.path.isdir(self.ospath) 

145 return self.dirLike 

146 

147 def transfer_from( 

148 self, 

149 src: ResourcePath, 

150 transfer: str, 

151 overwrite: bool = False, 

152 transaction: TransactionProtocol | None = None, 

153 multithreaded: bool = True, 

154 ) -> None: 

155 """Transfer the current resource to a local file. 

156 

157 Parameters 

158 ---------- 

159 src : `ResourcePath` 

160 Source URI. 

161 transfer : `str` 

162 Mode to use for transferring the resource. Supports the following 

163 options: copy, link, symlink, hardlink, relsymlink. 

164 overwrite : `bool`, optional 

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

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

167 If a transaction is provided, undo actions will be registered. 

168 multithreaded : `bool`, optional 

169 Whether threads are allowed to be used or not. 

170 """ 

171 # Fail early to prevent delays if remote resources are requested 

172 if transfer not in self.transferModes: 

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

174 

175 # Existence checks can take time so only try if the log message 

176 # will be issued. 

177 if log.isEnabledFor(logging.DEBUG): 177 ↛ 191line 177 didn't jump to line 191 because the condition on line 177 was always true

178 log.debug( 

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

180 src, 

181 src.exists(), 

182 self, 

183 self.exists(), 

184 transfer, 

185 ) 

186 

187 # The output location should not exist unless overwrite=True. 

188 # Rather than use `exists()`, use os.stat since we might need 

189 # the full answer later. 

190 dest_stat: os.stat_result | None 

191 try: 

192 # Do not read through links of the file itself. 

193 dest_stat = os.lstat(self.ospath) 

194 except FileNotFoundError: 

195 dest_stat = None 

196 

197 # It is possible that the source URI and target URI refer 

198 # to the same file. This can happen for a number of reasons 

199 # (such as soft links in the path, or they really are the same). 

200 # In that case log a message and return as if the transfer 

201 # completed (it technically did). A temporary file download 

202 # can't be the same so the test can be skipped. 

203 if dest_stat and src.isLocal and not src.isTemporary: 

204 # Be consistent and use lstat here (even though realpath 

205 # has been called). It does not harm. 

206 local_src_stat = os.lstat(src.ospath) 

207 if dest_stat.st_ino == local_src_stat.st_ino and dest_stat.st_dev == local_src_stat.st_dev: 

208 log.debug( 

209 "Destination URI %s is the same file as source URI %s, returning immediately." 

210 " No further action required.", 

211 self, 

212 src, 

213 ) 

214 return 

215 

216 if not overwrite and dest_stat: 

217 raise FileExistsError( 

218 f"Destination path '{self}' already exists. Transfer from {src} cannot be completed." 

219 ) 

220 

221 # Make the destination path absolute (but don't follow links since 

222 # that would possibly cause us to end up in the wrong place if the 

223 # file existed already as a soft link) 

224 newFullPath = os.path.abspath(self.ospath) 

225 outputDir = os.path.dirname(newFullPath) 

226 

227 # We do not have to special case FileResourcePath here because 

228 # as_local handles that. If remote download, download it to the 

229 # destination directory to allow an atomic rename but only if that 

230 # directory exists because we do not want to create a directory 

231 # but then end up with the download failing. 

232 tmpdir = outputDir if os.path.exists(outputDir) else None 

233 with src.as_local(multithreaded=multithreaded, tmpdir=tmpdir) as local_uri: 

234 is_temporary = local_uri.isTemporary 

235 local_src = local_uri.ospath 

236 

237 # Short circuit if the URIs are identical immediately. 

238 if self == local_uri: 

239 log.debug( 

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

241 " No further action required.", 

242 self, 

243 ) 

244 return 

245 

246 # Default transfer mode depends on whether we have a temporary 

247 # file or not. 

248 if transfer == "auto": 

249 transfer = self.transferDefault if not is_temporary else "copy" 

250 

251 if not os.path.exists(local_src): 

252 if is_temporary: 

253 if src == local_uri: 253 ↛ 257line 253 didn't jump to line 257 because the condition on line 253 was always true

254 msg = f"Local temporary file {src} has gone missing." 

255 else: 

256 # This will not happen in normal scenarios. 

257 msg = f"Local file {local_uri} downloaded from {src} has gone missing" 

258 else: 

259 msg = f"Source URI {src} does not exist" 

260 raise FileNotFoundError(msg) 

261 

262 # Follow soft links 

263 local_src = os.path.realpath(os.path.normpath(local_src)) 

264 

265 # Creating a symlink to a local copy of a remote resource 

266 # should never work. Creating a hardlink will work but should 

267 # not be allowed since it is highly unlikely that this is ever 

268 # an intended option and depends on the local target being 

269 # on the same file system as was used for the temporary file 

270 # download. 

271 # If a symlink is being requested for a local temporary file 

272 # that is likely undesirable but should not be refused. 

273 if is_temporary and src != local_uri and "link" in transfer: 

274 raise RuntimeError( 

275 f"Can not use local file system transfer mode {transfer} for remote resource ({src})" 

276 ) 

277 elif is_temporary and src == local_uri and "symlink" in transfer: 

278 log.debug( 

279 "Using a symlink for a temporary resource may lead to unexpected downstream failures." 

280 ) 

281 

282 # For temporary files we can own them if we created it. 

283 requested_transfer = transfer 

284 if src != local_uri and is_temporary and transfer == "copy": 

285 transfer = "move" 

286 

287 if not os.path.isdir(outputDir): 

288 # Must create the directory -- this can not be rolled back 

289 # since another transfer running concurrently may 

290 # be relying on this existing. 

291 _create_directories(outputDir) 

292 

293 if transaction is None: 293 ↛ 300line 293 didn't jump to line 300 because the condition on line 293 was always true

294 # Use a no-op transaction to reduce code duplication 

295 transaction = NoTransaction() 

296 

297 # For links the OS doesn't let us overwrite so if something does 

298 # exist we have to remove it before we do the actual "transfer" 

299 # below 

300 if "link" in transfer and overwrite and dest_stat: 

301 with contextlib.suppress(Exception): 

302 # If this fails we ignore it since it's a problem 

303 # that will manifest immediately below with a more relevant 

304 # error message 

305 self.remove() 

306 

307 if transfer == "move": 

308 # If a rename works we try that since that is guaranteed to 

309 # be atomic. If that fails we copy and rename. We do this 

310 # in case other processes are trying to move to the same 

311 # file and we want the "winner" to not be corrupted. 

312 try: 

313 with transaction.undoWith(f"move from {local_src}", os.rename, newFullPath, local_src): 

314 os.rename(local_src, newFullPath) 

315 except OSError: 

316 with self.temporary_uri(prefix=self.parent(), suffix=self.getExtension()) as temp_copy: 

317 shutil.copy(local_src, temp_copy.ospath) 

318 with transaction.undoWith( 

319 f"move from {local_src}", 

320 shutil.move, 

321 newFullPath, 

322 local_src, 

323 copy_function=shutil.copy, 

324 ): 

325 os.rename(temp_copy.ospath, newFullPath) 

326 os.remove(local_src) 

327 elif transfer == "copy": 

328 # We want atomic copies so first copy to a temp location in 

329 # the same output directory. This at least guarantees that 

330 # if multiple processes are writing to the same file 

331 # simultaneously the file we end up with will not be corrupt. 

332 if overwrite: 

333 with self.temporary_uri(prefix=self.parent(), suffix=self.getExtension()) as temp_copy: 

334 shutil.copy(local_src, temp_copy.ospath) 

335 with transaction.undoWith(f"copy from {local_src}", os.remove, newFullPath): 

336 os.rename(temp_copy.ospath, newFullPath) 

337 else: 

338 # Create the file exclusively to ensure that no others are 

339 # trying to write. 

340 temp_path = newFullPath + ".transfer-tmp" 

341 try: 

342 with open(temp_path, "x"): 

343 pass 

344 except FileExistsError: 

345 raise FileExistsError( 

346 f"Another process is writing to '{self}'." 

347 f" Transfer from {src} cannot be completed." 

348 ) 

349 with transaction.undoWith(f"copy from {local_src}", os.remove, temp_path): 

350 # Make sure file is writable, no matter the umask. 

351 st = os.stat(temp_path) 

352 os.chmod(temp_path, st.st_mode | stat.S_IWUSR) 

353 shutil.copy(local_src, temp_path) 

354 # Use link/remove to atomically and exclusively move the 

355 # file into place (only one concurrent linker can win). 

356 try: 

357 os.link(temp_path, newFullPath) 

358 except FileExistsError: 

359 raise FileExistsError( 

360 f"Another process wrote to '{self}'. Transfer from {src} cannot be completed." 

361 ) 

362 finally: 

363 os.remove(temp_path) 

364 elif transfer == "link": 

365 # Try hard link and if that fails use a symlink 

366 with transaction.undoWith(f"link to {local_src}", os.remove, newFullPath): 

367 try: 

368 os.link(local_src, newFullPath) 

369 except OSError: 

370 # Read through existing symlinks 

371 os.symlink(local_src, newFullPath) 

372 elif transfer == "hardlink": 

373 with transaction.undoWith(f"hardlink to {local_src}", os.remove, newFullPath): 

374 os.link(local_src, newFullPath) 

375 elif transfer == "symlink": 

376 # Read through existing symlinks 

377 with transaction.undoWith(f"symlink to {local_src}", os.remove, newFullPath): 

378 os.symlink(local_src, newFullPath) 

379 elif transfer == "relsymlink": 

380 # This is a standard symlink but using a relative path 

381 # Need the directory name to give to relative root 

382 # A full file path confuses it into an extra ../ 

383 newFullPathRoot = os.path.dirname(newFullPath) 

384 relPath = os.path.relpath(local_src, newFullPathRoot) 

385 with transaction.undoWith(f"relsymlink to {local_src}", os.remove, newFullPath): 

386 os.symlink(relPath, newFullPath) 

387 else: 

388 raise NotImplementedError(f"Transfer type '{transfer}' not supported.") 

389 

390 # This was an explicit move requested from a remote resource 

391 # try to remove that remote resource. We check is_temporary because 

392 # the local file would have been moved by shutil.move already. 

393 if requested_transfer == "move" and is_temporary and src != local_uri: 

394 # Transactions do not work here 

395 src.remove() 

396 

397 def walk( 

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

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

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

401 

402 Parameters 

403 ---------- 

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

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

406 

407 Yields 

408 ------ 

409 dirpath : `ResourcePath` 

410 Current directory being examined. 

411 dirnames : `list` of `str` 

412 Names of subdirectories within dirpath. 

413 filenames : `list` of `str` 

414 Names of all the files within dirpath. 

415 """ 

416 if not self.isdir(): 

417 raise ValueError("Can not walk a non-directory URI") 

418 

419 if isinstance(file_filter, str): 419 ↛ 420line 419 didn't jump to line 420 because the condition on line 419 was never true

420 file_filter = re.compile(file_filter) 

421 

422 for root, dirs, files in os.walk(self.ospath, followlinks=True): 

423 # Filter by the regex 

424 if file_filter is not None: 

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

426 yield type(self)(root, forceAbsolute=False, forceDirectory=True), dirs, files 

427 

428 @classmethod 

429 def _fixupPathUri( 

430 cls, 

431 parsed: urllib.parse.ParseResult, 

432 root: ResourcePath | None = None, 

433 forceAbsolute: bool = False, 

434 forceDirectory: bool | None = None, 

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

436 """Fix up relative paths in URI instances. 

437 

438 Parameters 

439 ---------- 

440 parsed : `~urllib.parse.ParseResult` 

441 The result from parsing a URI using `urllib.parse`. 

442 root : `ResourcePath`, optional 

443 Path to use as root when converting relative to absolute. 

444 If `None`, it will be the current working directory. It is only 

445 used if a file-scheme is used incorrectly with a relative path. 

446 forceAbsolute : `bool`, ignored 

447 Has no effect for this subclass. ``file`` URIs are always 

448 absolute. 

449 forceDirectory : `bool`, optional 

450 If `True` forces the URI to end with a separator, otherwise given 

451 URI is interpreted as is. 

452 

453 Returns 

454 ------- 

455 modified : `~urllib.parse.ParseResult` 

456 Update result if a URI is being handled. 

457 dirLike : `bool` or `None` 

458 `True` if given parsed URI has a trailing separator or 

459 ``forceDirectory`` is `True`. Otherwise can return the given 

460 value of ``forceDirectory``. 

461 

462 Notes 

463 ----- 

464 Relative paths are explicitly not supported by RFC8089 but `urllib` 

465 does accept URIs of the form ``file:relative/path.ext``. They need 

466 to be turned into absolute paths before they can be used. This is 

467 always done regardless of the ``forceAbsolute`` parameter. 

468 """ 

469 # assume we are not dealing with a directory like URI 

470 dirLike = forceDirectory 

471 

472 # file URI implies POSIX path separators so split as POSIX, 

473 # then join as os, and convert to abspath. Do not handle 

474 # home directories since "file" scheme is explicitly documented 

475 # to not do tilde expansion. 

476 sep = posixpath.sep 

477 

478 # Consistency check. 

479 if forceDirectory is False and parsed.path.endswith(sep): 

480 raise ValueError( 

481 f"URI {parsed.geturl()} ends with {sep} but " 

482 "forceDirectory parameter declares it to be a file." 

483 ) 

484 

485 # For an absolute path all we need to do is check if we need 

486 # to force the directory separator 

487 if posixpath.isabs(parsed.path): 

488 if forceDirectory: 

489 if not parsed.path.endswith(sep): 

490 parsed = parsed._replace(path=parsed.path + sep) 

491 dirLike = True 

492 return copy.copy(parsed), dirLike 

493 

494 # Relative path so must fix it to be compliant with the standard 

495 

496 # Replacement values for the URI 

497 replacements = {} 

498 

499 if root is None: 

500 root_str = os.path.abspath(os.path.curdir) 

501 else: 

502 if root.scheme and root.scheme != "file": 502 ↛ 503line 502 didn't jump to line 503 because the condition on line 502 was never true

503 raise RuntimeError(f"The override root must be a file URI not {root.scheme}") 

504 root_str = os.path.abspath(root.ospath) 

505 

506 replacements["path"] = posixpath.normpath(posixpath.join(os2posix(root_str), parsed.path)) 

507 

508 # normpath strips trailing "/" so put it back if necessary 

509 # Acknowledge that trailing separator exists. 

510 if forceDirectory or (parsed.path.endswith(sep) and not replacements["path"].endswith(sep)): 

511 replacements["path"] += sep 

512 dirLike = True 

513 

514 # ParseResult is a NamedTuple so _replace is standard API 

515 parsed = parsed._replace(**replacements) 

516 

517 if parsed.params or parsed.query: 517 ↛ 518line 517 didn't jump to line 518 because the condition on line 517 was never true

518 log.warning("Additional items unexpectedly encountered in file URI: %s", parsed.geturl()) 

519 

520 return parsed, dirLike 

521 

522 @contextlib.contextmanager 

523 def _openImpl( 

524 self, 

525 mode: str = "r", 

526 *, 

527 encoding: str | None = None, 

528 ) -> Iterator[IO]: 

529 with FileResourceHandle(mode=mode, log=log, uri=self, encoding=encoding) as buffer: 

530 yield buffer # type: ignore 

531 

532 def to_fsspec(self) -> tuple[AbstractFileSystem, str]: 

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

534 

535 Returns 

536 ------- 

537 fs : `fsspec.spec.AbstractFileSystem` 

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

539 path : `str` 

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

541 """ 

542 if fsspec is None: 542 ↛ 543line 542 didn't jump to line 543 because the condition on line 542 was never true

543 raise ImportError("fsspec is not available") 

544 # fsspec does not like URL encodings in file URIs so pass it the os 

545 # path instead. 

546 return fsspec.url_to_fs(self.ospath) 

547 

548 

549def _create_directories(name: str | bytes) -> None: 

550 """Create a directory and all of its parent directories that don't yet 

551 exist. 

552 

553 Parameters 

554 ---------- 

555 name : `str` or `bytes` 

556 Path to the directory to be created 

557 

558 Notes 

559 ----- 

560 The code in this function is duplicated from the Python standard library 

561 function os.makedirs with one change: if the user has set a process umask 

562 that prevents us from creating/accessing files in the newly created 

563 directories, the permissions of the directories are altered to allow 

564 owner-write and owner-traverse so that they can be used. 

565 """ 

566 # These are optional parameters in the original function, but they can be 

567 # constant here. 

568 mode = 0o777 

569 exist_ok = True 

570 

571 head, tail = os.path.split(name) 

572 if not tail: 

573 head, tail = os.path.split(head) 

574 if head and tail and not os.path.exists(head): 

575 try: 

576 _create_directories(head) 

577 except FileExistsError: 

578 # Defeats race condition when another thread created the path 

579 pass 

580 cdir: str | bytes = os.curdir 

581 if isinstance(tail, bytes): 581 ↛ 582line 581 didn't jump to line 582 because the condition on line 581 was never true

582 cdir = bytes(os.curdir, "ASCII") 

583 if tail == cdir: # xxx/newdir/. exists if xxx/newdir exists 583 ↛ 584line 583 didn't jump to line 584 because the condition on line 583 was never true

584 return 

585 try: 

586 os.mkdir(name, mode) 

587 # This is the portion that is modified relative to the standard library 

588 # version of the function. 

589 ensure_directory_is_writeable(name) 

590 # end modified portion 

591 except OSError: 

592 # Cannot rely on checking for EEXIST, since the operating system 

593 # could give priority to other errors like EACCES or EROFS 

594 if not exist_ok or not os.path.isdir(name): 

595 raise