( # noqa: C901, PLR0912, PLR0915
self,
rpath,
lpath,
recursive=False,
callback=DEFAULT_CALLBACK,
maxdepth=None,
batch_size=None,
**kwargs,
)
| 510 | ) |
| 511 | |
| 512 | def _get( # noqa: C901, PLR0912, PLR0915 |
| 513 | self, |
| 514 | rpath, |
| 515 | lpath, |
| 516 | recursive=False, |
| 517 | callback=DEFAULT_CALLBACK, |
| 518 | maxdepth=None, |
| 519 | batch_size=None, |
| 520 | **kwargs, |
| 521 | ) -> list[tuple[str, str, Optional[dict]]]: |
| 522 | if ( |
| 523 | isinstance(rpath, list) |
| 524 | or isinstance(lpath, list) |
| 525 | or has_magic(rpath) |
| 526 | or not self.exists(rpath) |
| 527 | or not recursive |
| 528 | ): |
| 529 | super().get( |
| 530 | rpath, |
| 531 | lpath, |
| 532 | recursive=recursive, |
| 533 | callback=callback, |
| 534 | maxdepth=maxdepth, |
| 535 | **kwargs, |
| 536 | ) |
| 537 | return [] |
| 538 | |
| 539 | if os.path.isdir(lpath) or lpath.endswith(os.path.sep): |
| 540 | lpath = self.join(lpath, os.path.basename(rpath)) |
| 541 | |
| 542 | if self.isfile(rpath): |
| 543 | with callback.branched(rpath, lpath) as child: |
| 544 | self.get_file(rpath, lpath, callback=child, **kwargs) |
| 545 | return [(rpath, lpath, None)] |
| 546 | |
| 547 | result: list[tuple[str, str, Optional[dict]]] = [] |
| 548 | _dirs: list[str] = [] |
| 549 | _files: dict[FileSystem, list[tuple[str, str, Optional[dict]]]] |
| 550 | _files = defaultdict(list) |
| 551 | |
| 552 | for root, dirs, files in self.walk(rpath, maxdepth=maxdepth, detail=True): |
| 553 | if files: |
| 554 | callback.set_size((callback.size or 0) + len(files)) |
| 555 | |
| 556 | parts = self.relparts(root, rpath) |
| 557 | if parts in ((os.curdir,), ("",)): |
| 558 | parts = () |
| 559 | dest_root = os.path.join(lpath, *parts) |
| 560 | if not maxdepth or len(parts) < maxdepth - 1: |
| 561 | _dirs.extend(f"{dest_root}{os.path.sep}{d}" for d in dirs) |
| 562 | |
| 563 | key = self._get_key_from_relative(root) |
| 564 | _, dvc_fs, _ = self._get_subrepo_info(key) |
| 565 | |
| 566 | for name, info in files.items(): |
| 567 | dvc_info = info.get("dvc_info") |
| 568 | fs_info = info.get("fs_info") |
| 569 | if dvc_fs and dvc_info and not fs_info: |
no test coverage detected